from __future__ import annotations import argparse import hashlib import json from pathlib import Path from typing import Any import pandas as pd from app.soat_crosswalk.domain.models import ( CrosswalkCatalog, CrosswalkRelationship, CrosswalkSource, ) from app.soat_crosswalk.infrastructure.catalogs import DEFAULT_CROSSWALK, XlsxCupsCatalog from app.soat_tariffs.infrastructure.json_catalog import JsonSoatTariffCatalog def _read_source(path: Path) -> pd.DataFrame: if not path.is_file() or path.stat().st_size <= 0: raise FileNotFoundError(f"Fuente de correlación ausente o vacía: {path}") try: if path.suffix.casefold() == ".csv": return pd.read_csv(path, dtype=str, keep_default_na=False) if path.suffix.casefold() in {".xlsx", ".xls"}: return pd.read_excel(path, dtype=str, keep_default_na=False) except Exception as exc: raise ValueError(f"No fue posible leer la fuente de correlación: {path.name}") from exc raise ValueError("La fuente de correlación debe ser CSV, XLS o XLSX") def _required_column(frame: pd.DataFrame, name: str) -> None: if name not in frame.columns: raise ValueError(f"Columna requerida ausente: {name}") def _normalize_cups(value: Any) -> str: raw = str(value or "").strip() if raw.endswith(".0"): raw = raw[:-2] normalized = raw.zfill(6) if not normalized.isdigit() or len(normalized) != 6: raise ValueError(f"Código CUPS inválido: {value}") return normalized def _normalize_soat(value: Any) -> str: raw = str(value or "").strip() if raw.endswith(".0"): raw = raw[:-2] if not raw.isdigit() or len(raw) not in {4, 5}: raise ValueError(f"Código SOAT inválido: {value}") return raw.zfill(5) def import_crosswalk( source_path: str | Path, *, cups_column: str, soat_column: str, output_path: str | Path = DEFAULT_CROSSWALK, reference_column: str | None = None, observation_column: str | None = None, group_column: str | None = None, source_reference: str = "Archivo de correlación validado suministrado por el usuario", cups_catalog: XlsxCupsCatalog | None = None, tariff_catalog: JsonSoatTariffCatalog | None = None, ) -> CrosswalkCatalog: source = Path(source_path) output = Path(output_path) frame = _read_source(source) for column in (cups_column, soat_column): _required_column(frame, column) for column in (reference_column, observation_column, group_column): if column: _required_column(frame, column) cups = cups_catalog or XlsxCupsCatalog() tariffs = tariff_catalog or JsonSoatTariffCatalog() cups.load() soat_catalog = tariffs.load(2026) soat_index = {entry.code: entry for entry in soat_catalog.entries} relationships: list[CrosswalkRelationship] = [] seen_pairs: set[tuple[str, str]] = set() for row_number, (_, row) in enumerate(frame.iterrows(), start=2): cups_code = _normalize_cups(row[cups_column]) soat_code = _normalize_soat(row[soat_column]) if not cups.contains(cups_code): raise ValueError(f"Fila {row_number}: CUPS desconocido {cups_code}") soat_entry = soat_index.get(soat_code) if soat_entry is None: raise ValueError(f"Fila {row_number}: SOAT desconocido {soat_code}") pair = (cups_code, soat_code) if pair in seen_pairs: raise ValueError(f"Fila {row_number}: par CUPS-SOAT duplicado {cups_code}/{soat_code}") seen_pairs.add(pair) if group_column: try: supplied_group = int(str(row[group_column]).strip()) except ValueError as exc: raise ValueError(f"Fila {row_number}: grupo quirúrgico inválido") from exc if supplied_group != soat_entry.surgical_group: raise ValueError( f"Fila {row_number}: grupo {supplied_group} contradice el catálogo SOAT " f"({soat_entry.surgical_group})" ) relationships.append( CrosswalkRelationship( cups_code=cups_code, cups_description=cups.description(cups_code) or "", soat_code=soat_code, soat_description=soat_entry.description, surgical_group=soat_entry.surgical_group, reference=( str(row[reference_column]).strip() if reference_column else f"{source_reference}; fila {row_number}" ), observation=str(row[observation_column]).strip() if observation_column else "", ) ) digest = hashlib.sha256(source.read_bytes()).hexdigest() relationships.sort(key=lambda item: (item.cups_code, item.soat_code)) catalog = CrosswalkCatalog( version=f"cups-soat-2026-{digest[:12]}", effective_from="2026-01-01", normative_version="CUPS Resolución 2706 de 2025 / Manual SOAT vigente 2026", cups_catalog_version=cups.version, soat_catalog_version=soat_catalog.version, source=CrosswalkSource( filename=source.name, sha256=digest, reference=source_reference, status="confirmed", cups_column=cups_column, soat_column=soat_column, ), relationships=relationships, ) payload = catalog.model_dump(mode="json") output.parent.mkdir(parents=True, exist_ok=True) output.write_text( json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) return catalog def main() -> None: parser = argparse.ArgumentParser(description="Importa un cruce autoritativo CUPS-SOAT 2026") parser.add_argument("source", type=Path) parser.add_argument("--cups-column", required=True) parser.add_argument("--soat-column", required=True) parser.add_argument("--reference-column") parser.add_argument("--observation-column") parser.add_argument("--group-column") parser.add_argument("--source-reference", default="Archivo validado suministrado por el usuario") parser.add_argument("--output", type=Path, default=DEFAULT_CROSSWALK) args = parser.parse_args() catalog = import_crosswalk( args.source, cups_column=args.cups_column, soat_column=args.soat_column, output_path=args.output, reference_column=args.reference_column, observation_column=args.observation_column, group_column=args.group_column, source_reference=args.source_reference, ) print( f"Cruce generado: {args.output} " f"({len(catalog.relationships)} relaciones, versión {catalog.version})" ) if __name__ == "__main__": main()