327 lines
12 KiB
Python
Executable File
327 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Script d'import du référentiel CCAM depuis un fichier Excel.
|
|
|
|
Ce script:
|
|
1. Lit le fichier Excel CCAM_V81.xls
|
|
2. Extrait les codes CCAM avec leurs descriptions
|
|
3. Convertit en format texte structuré
|
|
4. Importe dans le ReferentielsManager
|
|
5. Génère les chunks et l'index vectoriel
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
# Configuration du logging avant les imports qui l'utilisent
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
import pandas as pd
|
|
import openpyxl
|
|
EXCEL_SUPPORT = True
|
|
except ImportError:
|
|
EXCEL_SUPPORT = False
|
|
logger.error("pandas ou openpyxl non installé. Installez avec: pip install pandas openpyxl")
|
|
sys.exit(1)
|
|
|
|
from pipeline_mco_pmsi.rag.referentiels_manager import ReferentielsManager
|
|
|
|
|
|
def extract_ccam_from_excel(excel_path: Path) -> str:
|
|
"""
|
|
Extrait le contenu du référentiel CCAM depuis un fichier Excel.
|
|
|
|
Args:
|
|
excel_path: Chemin vers le fichier Excel CCAM
|
|
|
|
Returns:
|
|
Texte structuré du référentiel CCAM
|
|
"""
|
|
logger.info(f"Lecture du fichier Excel: {excel_path}")
|
|
|
|
# Utiliser pandas pour lire le fichier Excel (supporte .xls et .xlsx)
|
|
try:
|
|
df = pd.read_excel(excel_path, engine='xlrd')
|
|
except Exception as e:
|
|
logger.warning(f"Échec avec xlrd, tentative avec openpyxl: {e}")
|
|
try:
|
|
df = pd.read_excel(excel_path, engine='openpyxl')
|
|
except Exception as e2:
|
|
logger.error(f"Impossible de lire le fichier Excel: {e2}")
|
|
raise RuntimeError(f"Échec de lecture du fichier Excel: {e2}")
|
|
|
|
logger.info(f"DataFrame chargé: {len(df)} lignes, {len(df.columns)} colonnes")
|
|
logger.info(f"Colonnes: {list(df.columns)}")
|
|
|
|
# Structure pour stocker le contenu
|
|
lines = []
|
|
current_chapter = ""
|
|
|
|
# Analyser la structure des colonnes
|
|
# Adapter selon la structure réelle du fichier CCAM
|
|
col_names = list(df.columns)
|
|
|
|
# Essayer de détecter les colonnes importantes
|
|
code_col = None
|
|
desc_col = None
|
|
|
|
for i, col in enumerate(col_names):
|
|
col_lower = str(col).lower()
|
|
if 'code' in col_lower and code_col is None:
|
|
code_col = i
|
|
elif any(keyword in col_lower for keyword in ['libellé', 'libelle', 'description', 'texte']) and desc_col is None:
|
|
desc_col = i
|
|
|
|
# Si pas trouvé, utiliser les premières colonnes par défaut
|
|
if code_col is None:
|
|
code_col = 0
|
|
logger.warning("Colonne 'code' non détectée, utilisation de la colonne 0")
|
|
if desc_col is None:
|
|
desc_col = 2 if len(col_names) > 2 else 1
|
|
logger.warning(f"Colonne 'description' non détectée, utilisation de la colonne {desc_col}")
|
|
|
|
logger.info(f"Colonnes utilisées: code={code_col}, description={desc_col}")
|
|
|
|
# Parcourir les lignes
|
|
for idx, row in df.iterrows():
|
|
# Ignorer les lignes vides
|
|
if row.isna().all():
|
|
continue
|
|
|
|
code = str(row.iloc[code_col]).strip() if pd.notna(row.iloc[code_col]) else ""
|
|
text = str(row.iloc[desc_col]).strip() if pd.notna(row.iloc[desc_col]) and desc_col < len(row) else ""
|
|
|
|
# Nettoyer les valeurs NaN
|
|
if code == "nan":
|
|
code = ""
|
|
if text == "nan":
|
|
text = ""
|
|
|
|
# Ligne d'en-tête (première ligne)
|
|
if idx == 0 and text and not code:
|
|
lines.append(f"# RÉFÉRENTIEL CCAM")
|
|
lines.append("")
|
|
continue
|
|
|
|
# Chapitre (numéro seul dans la colonne code)
|
|
if code and code.replace(".", "").replace(",", "").isdigit() and text:
|
|
current_chapter = text
|
|
lines.append(f"\n## CHAPITRE {code}: {text}")
|
|
lines.append("")
|
|
continue
|
|
|
|
# Note ou exclusion (pas de code, mais du texte)
|
|
if not code and text:
|
|
if "exclusion" in text.lower():
|
|
lines.append(f"**Exclusion**: {text}")
|
|
elif text.startswith("Par ") or text.startswith("Note"):
|
|
lines.append(f"**Note**: {text}")
|
|
else:
|
|
lines.append(text)
|
|
lines.append("")
|
|
continue
|
|
|
|
# Code CCAM (format: XXXX000 ou XXXX000+XXX pour extensions ATIH)
|
|
# Accepter aussi les codes avec extensions
|
|
if code and len(code) >= 7:
|
|
# Vérifier le format de base (4 lettres + 3 chiffres)
|
|
base_code = code[:7]
|
|
if len(base_code) == 7 and base_code[:4].isalpha() and base_code[4:].isdigit():
|
|
# Extraire les métadonnées supplémentaires si disponibles
|
|
activite = ""
|
|
phase = ""
|
|
|
|
if len(row) > 3 and pd.notna(row.iloc[3]):
|
|
activite = str(row.iloc[3]).strip()
|
|
if len(row) > 4 and pd.notna(row.iloc[4]):
|
|
phase = str(row.iloc[4]).strip()
|
|
|
|
# Formater l'entrée CCAM
|
|
lines.append(f"### {code}")
|
|
if text:
|
|
lines.append(f"**Description**: {text}")
|
|
|
|
if activite and activite != "nan":
|
|
lines.append(f"**Activité**: {activite}")
|
|
if phase and phase != "nan":
|
|
lines.append(f"**Phase**: {phase}")
|
|
|
|
# Ajouter le chapitre pour contexte
|
|
if current_chapter:
|
|
lines.append(f"**Chapitre**: {current_chapter}")
|
|
|
|
# Détecter les extensions ATIH (format +XXX)
|
|
if "+" in code:
|
|
extension = code.split("+")[1] if len(code.split("+")) > 1 else ""
|
|
if extension:
|
|
lines.append(f"**Extension ATIH**: +{extension}")
|
|
|
|
lines.append("")
|
|
|
|
full_text = "\n".join(lines)
|
|
logger.info(f"Extraction terminée: {len(lines)} lignes, {len(full_text)} caractères")
|
|
|
|
return full_text
|
|
|
|
|
|
def main():
|
|
"""Point d'entrée principal du script."""
|
|
parser = argparse.ArgumentParser(
|
|
description="Import du référentiel CCAM dans le système"
|
|
)
|
|
parser.add_argument(
|
|
"--excel-file",
|
|
type=Path,
|
|
default=Path("data/referentiels/CCAM_V81.xls"),
|
|
help="Chemin vers le fichier Excel CCAM (défaut: data/referentiels/CCAM_V81.xls)"
|
|
)
|
|
parser.add_argument(
|
|
"--version",
|
|
type=str,
|
|
default="V81",
|
|
help="Version du référentiel CCAM (défaut: V81)"
|
|
)
|
|
parser.add_argument(
|
|
"--data-dir",
|
|
type=Path,
|
|
default=Path("data/referentiels"),
|
|
help="Répertoire de stockage des référentiels (défaut: data/referentiels)"
|
|
)
|
|
parser.add_argument(
|
|
"--skip-indexing",
|
|
action="store_true",
|
|
help="Ne pas créer l'index vectoriel (seulement import et chunking)"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Vérifier que le fichier existe
|
|
if not args.excel_file.exists():
|
|
logger.error(f"Fichier Excel introuvable: {args.excel_file}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
# 1. Extraire le contenu du fichier Excel
|
|
logger.info("=" * 60)
|
|
logger.info("ÉTAPE 1: Extraction du contenu Excel")
|
|
logger.info("=" * 60)
|
|
ccam_text = extract_ccam_from_excel(args.excel_file)
|
|
|
|
# Sauvegarder le texte extrait
|
|
text_output_path = args.data_dir / f"ccam_{args.version}_extracted.txt"
|
|
with open(text_output_path, "w", encoding="utf-8") as f:
|
|
f.write(ccam_text)
|
|
logger.info(f"Texte extrait sauvegardé dans: {text_output_path}")
|
|
|
|
# 2. Importer dans le ReferentielsManager
|
|
logger.info("")
|
|
logger.info("=" * 60)
|
|
logger.info("ÉTAPE 2: Import dans ReferentielsManager")
|
|
logger.info("=" * 60)
|
|
|
|
manager = ReferentielsManager(data_dir=args.data_dir)
|
|
|
|
# Créer un fichier PDF temporaire pour l'import
|
|
# (le ReferentielsManager attend un PDF, mais on va contourner ça)
|
|
# Pour l'instant, on va directement sauvegarder le texte et créer la version
|
|
|
|
# Sauvegarder le texte pour chunking
|
|
text_file_path = args.data_dir / f"ccam_{args.version}_text.txt"
|
|
with open(text_file_path, "w", encoding="utf-8") as f:
|
|
f.write(ccam_text)
|
|
|
|
# Créer manuellement la version du référentiel
|
|
import hashlib
|
|
from datetime import datetime
|
|
from pipeline_mco_pmsi.models.metadata import ReferentielVersion
|
|
|
|
file_hash = hashlib.sha256(ccam_text.encode()).hexdigest()
|
|
placeholder_hash = "0" * 64
|
|
|
|
referentiel_version = ReferentielVersion(
|
|
type="ccam",
|
|
version=args.version,
|
|
import_date=datetime.now(),
|
|
file_hash=file_hash,
|
|
chunk_count=0,
|
|
index_hash=placeholder_hash,
|
|
)
|
|
|
|
logger.info(f"Référentiel CCAM {args.version} créé avec hash: {file_hash[:16]}...")
|
|
|
|
# 3. Chunking
|
|
logger.info("")
|
|
logger.info("=" * 60)
|
|
logger.info("ÉTAPE 3: Chunking du référentiel")
|
|
logger.info("=" * 60)
|
|
|
|
chunks = manager.chunk_referentiel(referentiel_version)
|
|
logger.info(f"Chunking terminé: {len(chunks)} chunks créés")
|
|
|
|
# Créer une nouvelle version avec le chunk_count mis à jour
|
|
referentiel_version = ReferentielVersion(
|
|
type=referentiel_version.type,
|
|
version=referentiel_version.version,
|
|
import_date=referentiel_version.import_date,
|
|
file_hash=referentiel_version.file_hash,
|
|
chunk_count=len(chunks),
|
|
index_hash=referentiel_version.index_hash,
|
|
)
|
|
|
|
# 4. Indexation (optionnel)
|
|
if not args.skip_indexing:
|
|
logger.info("")
|
|
logger.info("=" * 60)
|
|
logger.info("ÉTAPE 4: Construction de l'index vectoriel")
|
|
logger.info("=" * 60)
|
|
|
|
vector_index = manager.build_index(chunks)
|
|
logger.info(f"Index vectoriel créé:")
|
|
logger.info(f" - Hash: {vector_index.index_hash[:16]}...")
|
|
logger.info(f" - Dimension: {vector_index.dimension}")
|
|
logger.info(f" - Nombre de vecteurs: {vector_index.num_vectors}")
|
|
logger.info(f" - Type d'index: {vector_index.index_type}")
|
|
|
|
# Créer une nouvelle version avec l'index_hash mis à jour
|
|
referentiel_version = ReferentielVersion(
|
|
type=referentiel_version.type,
|
|
version=referentiel_version.version,
|
|
import_date=referentiel_version.import_date,
|
|
file_hash=referentiel_version.file_hash,
|
|
chunk_count=referentiel_version.chunk_count,
|
|
index_hash=vector_index.index_hash,
|
|
)
|
|
else:
|
|
logger.info("")
|
|
logger.info("Indexation ignorée (--skip-indexing)")
|
|
|
|
# 5. Résumé
|
|
logger.info("")
|
|
logger.info("=" * 60)
|
|
logger.info("IMPORT TERMINÉ AVEC SUCCÈS")
|
|
logger.info("=" * 60)
|
|
logger.info(f"Référentiel: CCAM {args.version}")
|
|
logger.info(f"Hash du fichier: {referentiel_version.file_hash[:16]}...")
|
|
logger.info(f"Nombre de chunks: {referentiel_version.chunk_count}")
|
|
if not args.skip_indexing:
|
|
logger.info(f"Hash de l'index: {referentiel_version.index_hash[:16]}...")
|
|
logger.info(f"Date d'import: {referentiel_version.import_date}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de l'import: {e}", exc_info=True)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|