from __future__ import annotations import hashlib import pickle import re import unicodedata from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any from app.case_epicrisis.domain.curation import CatalogReference, CodeSystem from app.soat_crosswalk.infrastructure.catalogs import XlsxCupsCatalog _NON_CODE = re.compile(r"[^A-Z0-9]+") _WORD = re.compile(r"[A-Z0-9]+") _IGNORED_WORDS = { "DE", "DEL", "LA", "LAS", "EL", "LOS", "CON", "SIN", "POR", "PARA", "Y", "O", "EN", "UN", "UNA", "MAS", } _ANATOMY_WORDS = { "BRAZO", "CLAVICULA", "CODO", "HOMBRO", "HUMERO", "MANO", "PIE", "RODILLA", "TIBIA", "TOBILLO", } def _ascii_upper(value: Any) -> str: normalized = unicodedata.normalize("NFKD", str(value or "")) return "".join(char for char in normalized if not unicodedata.combining(char)).upper() def normalize_catalog_code(system: CodeSystem, value: Any) -> str: code = _NON_CODE.sub("", _ascii_upper(value)) if system == CodeSystem.CIE10: return code return code def _description_words(value: Any) -> set[str]: return { token for token in _WORD.findall(_ascii_upper(value)) if len(token) > 2 and token not in _IGNORED_WORDS and not token.isdigit() } @dataclass(frozen=True) class _CatalogFiles: index_name: str directory: Path source_name: str class RepositoryCodingCatalogRegistry: """Anti-corruption adapter over the repository-owned FAISS docstores.""" def __init__(self, base_dir: Path): self.base_dir = Path(base_dir) self._files = { CodeSystem.CIE10: _CatalogFiles( "faiss_principal_jerarquico", self.base_dir / "faiss_principal_jerarquico", "modules/processing/cie10/CIE10.xlsx", ), CodeSystem.CUPS: _CatalogFiles( "cups_faiss", self.base_dir / "cups_faiss", "TablaReferencia_CUPS__1.xlsx", ), CodeSystem.SOAT: _CatalogFiles( "soat_faiss", self.base_dir / "soat_faiss", "Manual-SOAT-2025.pdf", ), } self._references: dict[CodeSystem, CatalogReference] = {} self._entries: dict[CodeSystem, dict[str, str]] = {} self._cups_catalog = XlsxCupsCatalog() def reference(self, system: CodeSystem) -> CatalogReference: if system == CodeSystem.BILLING_INTERNAL: return CatalogReference( system=system, index_name="documento_fuente", fingerprint="not-applicable", source_name="Código interno de facturación", available=True, ) if system not in self._references: self._references[system] = self._build_reference(system) return self._references[system] def references(self) -> list[CatalogReference]: return [self.reference(system) for system in (CodeSystem.CIE10, CodeSystem.CUPS, CodeSystem.SOAT)] def contains(self, system: CodeSystem, code: str) -> bool: normalized = normalize_catalog_code(system, code) if not normalized: return False return normalized in self._load_entries(system) def description_matches(self, system: CodeSystem, code: str, description: str) -> bool: normalized = normalize_catalog_code(system, code) catalog_description = self._load_entries(system).get(normalized, "") if not catalog_description or not str(description or "").strip(): return False expected = _description_words(catalog_description) observed = _description_words(description) expected_anatomy = expected & _ANATOMY_WORDS observed_anatomy = observed & _ANATOMY_WORDS if expected_anatomy and observed_anatomy and expected_anatomy.isdisjoint(observed_anatomy): return False return bool(expected & observed) def classification(self, system: CodeSystem, code: str) -> str | None: if system != CodeSystem.CUPS: return None return self._cups_catalog.classification(normalize_catalog_code(system, code)) def _build_reference(self, system: CodeSystem) -> CatalogReference: files = self._files[system] paths = [files.directory / "index.faiss", files.directory / "index.pkl"] available = all(path.is_file() for path in paths) digest = hashlib.sha256() built_at = "" if available: for path in paths: digest.update(path.name.encode()) with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) built_at = datetime.fromtimestamp(max(path.stat().st_mtime for path in paths)).isoformat() return CatalogReference( system=system, index_name=files.index_name, fingerprint=digest.hexdigest() if available else "unavailable", built_at=built_at, source_name=files.source_name, available=available, ) def _load_entries(self, system: CodeSystem) -> dict[str, str]: if system in self._entries: return self._entries[system] if system == CodeSystem.BILLING_INTERNAL or not self.reference(system).available: self._entries[system] = {} return self._entries[system] # These pickle files are repository-owned artifacts already loaded elsewhere with # allow_dangerous_deserialization=True. Never accept a user-provided path here. with (self._files[system].directory / "index.pkl").open("rb") as stream: docstore, _index_to_docstore_id = pickle.load(stream) # noqa: S301 documents = getattr(docstore, "_dict", {}) entries: dict[str, str] = {} for document in documents.values(): content = str(getattr(document, "page_content", "") or "") metadata = getattr(document, "metadata", {}) or {} code = metadata.get("codigo") or metadata.get("code") or "" description = " ".join( str(value or "") for value in ( metadata.get("nombre"), metadata.get("descripcion") or metadata.get("description"), content, ) if str(value or "").strip() ) if code: entries[normalize_catalog_code(system, code)] = str(description) continue self._collect_codes_from_content(system, content, entries) self._entries[system] = entries return entries @staticmethod def _collect_codes_from_content( system: CodeSystem, content: str, entries: dict[str, str], ) -> None: patterns = { CodeSystem.CIE10: re.compile(r"\b([A-TV-Z]\d{2}(?:[0-9A-Z])?(?:\.[0-9A-Z]{1,2})?)\b"), CodeSystem.CUPS: re.compile(r"\b(\d{6}|[0-9]{3}[A-Z][0-9]{2})\b"), CodeSystem.SOAT: re.compile(r"\b(\d{4,6})\b"), } pattern = patterns.get(system) if pattern is None: return for match in pattern.finditer(content): entries.setdefault(normalize_catalog_code(system, match.group(1)), content[:500])