import json import os from pathlib import Path import pandas as pd from langchain_community.vectorstores import FAISS from modules.processing.embeddings import get_hf_embeddings EMBEDDINGS_MODEL = os.getenv("EMBEDDINGS_MODEL", "intfloat/e5-small-v2") PROJECT_ROOT = Path(__file__).resolve().parents[3] SCRIPT_DIR = Path(__file__).resolve().parent DEFAULT_STRUCTURE_JSON = PROJECT_ROOT / "cie10_estructura_completa.json" DEFAULT_INDEX_DIR = PROJECT_ROOT / "faiss_principal_jerarquico" DEFAULT_EXCEL_PATH = SCRIPT_DIR / "CIE10.xlsx" DEFAULT_BLOQUES_PATH = SCRIPT_DIR / "bloques.json" def _resolve_path(env_var: str, default_path: Path) -> Path: raw = os.getenv(env_var) path = Path(raw) if raw else default_path if not path.is_absolute(): path = PROJECT_ROOT / path return path def normalizar_texto(texto: str) -> str: texto = str(texto) texto = texto.replace("TRAZO SUGESTIVO DE FRACTURA", "FRACTURA") texto = texto.replace("QUEMADURA POR FRICCION", "QUEMADURA POR FRICCION") return texto.strip() def leer_excel(ruta_archivo: Path): df = pd.read_excel(ruta_archivo, skiprows=2) df.columns = df.columns.str.strip() if "COD_3" not in df.columns or "DESRIPCION CATEGORIAS DE TRES CARACTERES" not in df.columns: raise ValueError( "No se encontraron columnas COD_3 y DESRIPCION CATEGORIAS DE TRES CARACTERES en el Excel" ) categorias = {} for _, row in df.iterrows(): if pd.notna(row["COD_3"]) and pd.notna(row["DESRIPCION CATEGORIAS DE TRES CARACTERES"]): codigo_cat = str(row["COD_3"]).strip() descripcion_cat = normalizar_texto(row["DESRIPCION CATEGORIAS DE TRES CARACTERES"]) categorias.setdefault( codigo_cat, {"codigo": codigo_cat, "descripcion": descripcion_cat, "subcategorias": []}, ) if "COD_4" in df.columns and "DESCRIPCION CODIGOS DE CUATRO CARACTERES" in df.columns: if pd.notna(row["COD_4"]) and pd.notna(row["DESCRIPCION CODIGOS DE CUATRO CARACTERES"]): codigo_sub = str(row["COD_4"]).strip() descripcion_sub = normalizar_texto(row["DESCRIPCION CODIGOS DE CUATRO CARACTERES"]) codigo_categoria = codigo_sub[:3] categorias.setdefault( codigo_categoria, { "codigo": codigo_categoria, "descripcion": "Categoria no encontrada en el Excel", "subcategorias": [], }, ) categorias[codigo_categoria]["subcategorias"].append( {"codigo": codigo_sub, "descripcion": descripcion_sub} ) return list(categorias.values()) def _codigo_en_rango(codigo: str, inicio: str, fin: str) -> bool: base = codigo.split(".")[0] return inicio <= base <= fin def generar_estructura_jerarquica(bloques: list, categorias: list): mapeo_bloques = { "A00-B99": {"start": "A00", "end": "B99"}, "C00-D48": {"start": "C00", "end": "D48"}, "D50-D89": {"start": "D50", "end": "D89"}, "E00-E89": {"start": "E00", "end": "E89"}, "F01-F99": {"start": "F01", "end": "F99"}, "G00-G99": {"start": "G00", "end": "G99"}, "H00-H59": {"start": "H00", "end": "H59"}, "H60-H95": {"start": "H60", "end": "H95"}, "I00-I99": {"start": "I00", "end": "I99"}, "J00-J99": {"start": "J00", "end": "J99"}, "K00-K95": {"start": "K00", "end": "K95"}, "L00-L99": {"start": "L00", "end": "L99"}, "M00-M99": {"start": "M00", "end": "M99"}, "N00-N99": {"start": "N00", "end": "N99"}, "O00-O9A": {"start": "O00", "end": "O9A"}, "P00-P96": {"start": "P00", "end": "P96"}, "Q00-Q99": {"start": "Q00", "end": "Q99"}, "R00-R99": {"start": "R00", "end": "R99"}, "S00-T88": {"start": "S00", "end": "T88"}, "V00-Y99": {"start": "V00", "end": "Y99"}, "Z00-Z99": {"start": "Z00", "end": "Z99"}, "U00-U85": {"start": "U00", "end": "U85"}, } estructura = {"bloques": []} for bloque in bloques: codigo_bloque = bloque["codigo"] if codigo_bloque not in mapeo_bloques: continue bloque_entry = { "codigo": codigo_bloque, "nombre": bloque.get("nombre", codigo_bloque), "categorias": [], } start = mapeo_bloques[codigo_bloque]["start"] end = mapeo_bloques[codigo_bloque]["end"] for categoria in categorias: codigo_cat = categoria["codigo"] if _codigo_en_rango(codigo_cat, start, end): bloque_entry["categorias"].append( { "codigo": codigo_cat, "descripcion": categoria["descripcion"], "subcategorias": categoria["subcategorias"], } ) estructura["bloques"].append(bloque_entry) return estructura def guardar_json(datos: dict, ruta_archivo: Path) -> None: ruta_archivo.parent.mkdir(parents=True, exist_ok=True) with open(ruta_archivo, "w", encoding="utf-8") as f: json.dump(datos, f, ensure_ascii=False, indent=2) def cargar_o_generar_estructura() -> tuple[dict, Path]: estructura_path = _resolve_path("CIE10_ESTRUCTURA_JSON", DEFAULT_STRUCTURE_JSON) if estructura_path.exists(): with open(estructura_path, "r", encoding="utf-8") as f: return json.load(f), estructura_path excel_path = _resolve_path("CIE10_EXCEL_PATH", DEFAULT_EXCEL_PATH) bloques_path = _resolve_path("CIE10_BLOQUES_JSON", DEFAULT_BLOQUES_PATH) if not excel_path.exists(): raise FileNotFoundError(f"No se encontro Excel CIE10: {excel_path}") if not bloques_path.exists(): raise FileNotFoundError( f"No se encontro bloques.json: {bloques_path}. " "Alternativa: coloca cie10_estructura_completa.json en la raiz del proyecto." ) with open(bloques_path, "r", encoding="utf-8") as f: bloques_data = json.load(f) bloques_cie10 = bloques_data.get("bloques") if not isinstance(bloques_cie10, list): raise ValueError("El archivo bloques.json no contiene la clave 'bloques' como lista") categorias_cie10 = leer_excel(excel_path) estructura = generar_estructura_jerarquica(bloques_cie10, categorias_cie10) guardar_json(estructura, estructura_path) return estructura, estructura_path def configurar_base_datos(estructura_completa: dict): embeddings = get_hf_embeddings(EMBEDDINGS_MODEL) textos = [] metadatos = [] for bloque in estructura_completa.get("bloques", []): texto_bloque = f"BLOQUE {bloque['codigo']}: {bloque.get('nombre', '')}" textos.append(texto_bloque) metadatos.append( { "nivel": 1, "tipo": "bloque", "codigo": bloque["codigo"], "ruta": bloque["codigo"], } ) for categoria in bloque.get("categorias", []): texto_categoria = f"{texto_bloque}\nCATEGORIA {categoria['codigo']}: {categoria.get('descripcion', '')}" textos.append(texto_categoria) metadatos.append( { "nivel": 2, "tipo": "categoria", "codigo": categoria["codigo"], "codigo_padre": bloque["codigo"], "ruta": f"{bloque['codigo']} > {categoria['codigo']}", } ) for subcategoria in categoria.get("subcategorias", []): texto_sub = ( f"{texto_categoria}\nSUBCATEGORIA {subcategoria['codigo']}: " f"{subcategoria.get('descripcion', '')}" ) textos.append(texto_sub) metadatos.append( { "nivel": 3, "tipo": "subcategoria", "codigo": subcategoria["codigo"], "codigo_padre": categoria["codigo"], "bloque_padre": bloque["codigo"], "ruta": ( f"{bloque['codigo']} > {categoria['codigo']} > " f"{subcategoria['codigo']}" ), } ) if not textos: raise ValueError("No hay datos CIE-10 para indexar en FAISS") return FAISS.from_texts(textos, embedding=embeddings, metadatas=metadatos) def main() -> None: estructura_completa, estructura_path = cargar_o_generar_estructura() db = configurar_base_datos(estructura_completa) index_dir = _resolve_path("CIE10_FAISS_INDEX", DEFAULT_INDEX_DIR) index_dir.mkdir(parents=True, exist_ok=True) db.save_local(str(index_dir)) print(f"Estructura CIE-10 usada: {estructura_path}") print(f"Indice FAISS CIE-10 generado en: {index_dir}") if __name__ == "__main__": main()