from __future__ import annotations import re from typing import Any, cast from app.services.soat_processing import _parse_cie10_lines from .common import ( _normalize_ascii, _normalize_whitespace, ) from .pdf_draft import ( coerce_pdf_draft, ) _CIE10_INLINE_PATTERN = re.compile( r"^\s*(?P[A-TV-Z]\d{2}[0-9A-Z]?(?:\.[0-9A-Z]{1,2})?)\s*(?:[-:–—]\s*|\s+)(?P.+)$", flags=re.IGNORECASE, ) _CIE10_CODE_PATTERN = re.compile(r"^[A-TV-Z]\d{2}[0-9A-Z]?(?:\.[0-9A-Z]{1,2})?$", flags=re.IGNORECASE) _TECHNICAL_CODE_VALUES = {"", "error", "pendiente", "sin codigo", "sin código", "no disponible"} _TECHNICAL_DESCRIPTION_SNIPPETS = ( "codificación automática no disponible", "codificación automática no disponible", "no se encontró código cie-10 correspondiente", "no se encontró código cie-10 correspondiente", "no se pudo codificar automáticamente este diagnostico", "no se pudo codificar automáticamente este diagnóstico", ) _CIE10_PENDING_REASONS = { "sin_codigo": "El diagnóstico no tiene un código CIE-10 explícito.", "codigo_no_encontrado": "El código explícito no existe en el catálogo CIE-10.", "codigo_no_hoja": "El código corresponde a una categoría grupal y requiere una subcategoría hoja.", "catalogo_no_disponible": "El catálogo CIE-10 no está disponible para validar el concepto oficial.", "inconsistencia_catalogo": "El código y la descripción no tienen una coincidencia segura en el catálogo CIE-10.", } _POST_CONTEXT_PATTERN = re.compile( r"\b(post|postquir|postop|post operatorio|posoperatorio)\b", flags=re.IGNORECASE ) _PRE_CONTEXT_PATTERN = re.compile(r"\b(pre|prequir|pre operatorio)\b", flags=re.IGNORECASE) def _is_real_cie10_code(value: Any) -> bool: return bool(_CIE10_CODE_PATTERN.fullmatch(str(value or "").strip().upper())) def _normalize_cie10_code(value: Any) -> str: code = str(value or "").strip().upper().replace(" ", "") return code.replace(".", "") if _CIE10_CODE_PATTERN.fullmatch(code) else code def normalize_clinical_diagnosis_text(value: Any) -> str: """Apply editorial cleanup while retaining the words and meaning extracted from the source.""" text = str(value or "") text = re.sub(r"\s+", " ", text) text = re.sub(r"\s*([,;:.])\s*", r"\1 ", text) text = re.sub(r"\s+([)\]])", r"\1", text) text = re.sub(r"([(\[])\s+", r"\1", text) text = re.sub(r"(?:\s*\.\s*){2,}", ". ", text) return text.strip(" \n\r\t") def _deduplicate_clinical_texts(values: list[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for value in values: normalized = normalize_clinical_diagnosis_text(value) key = _normalize_ascii(normalized).casefold() if normalized and key not in seen: seen.add(key) result.append(normalized) return result def _clinical_text_candidates(seed: dict[str, Any], *, official_concept: str = "") -> list[tuple[str, str]]: """Return source prose in documentary order as (original, normalized) pairs.""" candidates: list[Any] = [] for key in ("descripcion_clinica_normalizada", "descripcion"): value = seed.get(key) if value not in (None, ""): text = str(value) inline = extraer_cie10_inline(text) if inline: text = inline["descripcion"] candidates.append(text) official_key = _normalize_ascii(official_concept).casefold() if not any( _normalize_ascii(normalize_clinical_diagnosis_text(value)).casefold() != official_key for value in candidates ): candidates = [] if not candidates: value = seed.get("diagnostico") or seed.get("texto_original") if value not in (None, ""): text = str(value) inline = extraer_cie10_inline(text) candidates.append(inline["descripcion"] if inline else text) result: list[tuple[str, str]] = [] seen: set[str] = set() for original in candidates: normalized = normalize_clinical_diagnosis_text(original) key = _normalize_ascii(normalized).casefold() if not normalized or key in seen: continue # The catalog concept is not documentary prose when it was copied into a legacy field. if official_key and key == official_key and len(candidates) > 1: continue seen.add(key) result.append((str(original).strip(), normalized)) return result def _resolution_state_and_reasons( *, code: str, resolution: dict[str, Any], retriever_available: bool ) -> tuple[str, list[str]]: if not code: return "pendiente_revision", [_CIE10_PENDING_REASONS["sin_codigo"]] if not retriever_available: return "pendiente_revision", [_CIE10_PENDING_REASONS["catalogo_no_disponible"]] if not resolution: return "pendiente_revision", [_CIE10_PENDING_REASONS["codigo_no_encontrado"]] if not resolution.get("es_hoja"): return "pendiente_revision", [_CIE10_PENDING_REASONS["codigo_no_hoja"]] if resolution.get("inconsistente"): return "pendiente_revision", [_CIE10_PENDING_REASONS["inconsistencia_catalogo"]] return "oficial", [] def _is_technical_code(value: Any) -> bool: return ( _normalize_whitespace(value).casefold() in {_normalize_whitespace(item).casefold() for item in _TECHNICAL_CODE_VALUES} or not str(value or "").strip() ) def _is_technical_description(value: Any) -> bool: normalized = _normalize_ascii(value).casefold() return any( _normalize_ascii(snippet).casefold() in normalized for snippet in _TECHNICAL_DESCRIPTION_SNIPPETS ) def _as_text_list(value: Any) -> list[str]: if not isinstance(value, list): return [] return [ normalize_clinical_diagnosis_text(item) for item in value if normalize_clinical_diagnosis_text(item) ] def extraer_cie10_inline(diagnostico: str) -> dict[str, str] | None: if not diagnostico: return None match = _CIE10_INLINE_PATTERN.match(str(diagnostico).strip()) if not match: return None return { "codigo": match.group("codigo").upper(), "descripcion": match.group("descripcion").strip(), } def _parse_codigo_asignado(value: Any) -> dict[str, str] | None: raw = _normalize_whitespace(value) if not raw: return None parsed = _parse_cie10_lines(raw) if parsed: return { "codigo": _normalize_cie10_code(parsed[0].get("codigo")), "descripcion": _normalize_whitespace(parsed[0].get("descripcion")), } inline = extraer_cie10_inline(raw) if inline: return inline return None def _coerce_tipo_contexto(value: Any) -> str: normalized = _normalize_ascii(value).lower() if _POST_CONTEXT_PATTERN.search(normalized): return "post" if _PRE_CONTEXT_PATTERN.search(normalized): return "pre" return "no_clasificado" def _merge_tipo_contexto(current: str, incoming: str) -> str: current = str(current or "").strip() or "no_clasificado" incoming = str(incoming or "").strip() or "no_clasificado" if current == incoming: return current if "no_clasificado" in {current, incoming}: return incoming if current == "no_clasificado" else current return "mixto" def _normalize_diagnostico_key_text(value: Any) -> str: text = _normalize_ascii(value).lower() text = re.sub( r"\b(post|postquirurgico|postquirúrgico|prequirurgico|prequirúrgico|postop|preop|pre operatorio|post operatorio)\b", " ", text, ) text = re.sub(r"[^a-z0-9]+", " ", text) return re.sub(r"\s+", " ", text).strip() def _source_priority(source: str) -> int: priorities = { "manual_pdf": 5, "manual_soat": 4, "historia_clinica": 3, "quirurgico": 3, "qx_agent": 2, "auto_soat": 1, "soat": 1, } return priorities.get(str(source or "").strip(), 0) def _coding_state_priority(state: Any) -> int: return { "codificado": 4, "inferido_revisable": 3, "provisional_grupo": 2, "pendiente_validacion": 1, }.get(str(state or "").strip(), 0) def _build_diagnostico_texto_presentacion(item: dict[str, Any]) -> str: codigo = _normalize_cie10_code(item.get("codigo")) concept = normalize_clinical_diagnosis_text( item.get("concepto_diagnostico_presentacion") or item.get("diagnostico") or item.get("descripcion") or item.get("descripcion_original") ) clinical = normalize_clinical_diagnosis_text( item.get("descripcion_clinica_normalizada") or item.get("descripcion") ) detail = concept or clinical or "Diagnóstico sin descripción" suffix = f". {clinical}" if clinical and concept and clinical.casefold() != concept.casefold() else "" base = f"{codigo} - {detail}{suffix}" if codigo else f"PENDIENTE - {detail}{suffix}" if not item.get("es_pop"): return base procedures = item.get("procedimientos_pop_asociados") or [item.get("procedimiento_pop_asociado")] procedure_text = "; ".join( dict.fromkeys(str(value).strip() for value in procedures if str(value).strip()) ) return f"{base} — POSTOPERATORIO {procedure_text}" if procedure_text else f"{base} — POSTOPERATORIO" def _coerce_raw_diagnostico_item(raw_item: Any) -> dict[str, Any] | None: if isinstance(raw_item, dict): return dict(raw_item) text = _normalize_whitespace(raw_item) if not text: return None return {"diagnostico": text} def _build_diagnostico_seed(raw: dict[str, Any]) -> dict[str, Any]: seed = { "diagnostico": _normalize_whitespace( raw.get("diagnostico") or raw.get("texto_original") or raw.get("descripcion") ), "descripcion": _normalize_whitespace(raw.get("descripcion")), "codigo": _normalize_cie10_code(raw.get("codigo")), "descripcion_original": _normalize_whitespace(raw.get("descripcion")), "texto_original": str( raw.get("texto_original") or raw.get("diagnostico") or raw.get("descripcion") or "" ).strip(), } for key in ( "origen", "estado_codificacion", "estado_concepto_cie10", "razones_revision", "razon_revision", "concepto_cie10_oficial", "concepto_diagnostico_presentacion", "descripcion_clinica_normalizada", "descripciones_clinicas", "catalog_description", "procedimiento_origen", "evidencia_clinica", "fuente_documental", "pagina", "seccion", "advertencias", "es_pop", "procedimiento_pop_asociado", "procedimientos_pop_asociados", "hallazgo_quirurgico", "descripcion_quirurgica", "fuente_pop", "pagina_pop", "seccion_pop", "extracto_pop", "fuente_cirugia", "pagina_cirugia", "seccion_cirugia", "extracto_cirugia", "evidencias_pop", "evidencias_cirugia", "evidencias", "evidence", "document_uid", "document_id", "document_type", "filename", "page", "section", ): if raw.get(key) not in (None, "", []): seed[key] = raw[key] return seed def _apply_codigo_asignado(seed: dict[str, str], raw: dict[str, Any]) -> dict[str, str]: updated = dict(seed) parsed_codigo_asignado = _parse_codigo_asignado(raw.get("codigo_asignado")) if not parsed_codigo_asignado: return updated if _is_technical_code(updated["codigo"]) or not updated["codigo"]: updated["codigo"] = parsed_codigo_asignado["codigo"] if not updated["descripcion"] or _is_technical_description(updated["descripcion"]): updated["descripcion"] = parsed_codigo_asignado["descripcion"] return updated def _apply_inline_cie10(seed: dict[str, str]) -> dict[str, str]: updated = dict(seed) inline = extraer_cie10_inline(updated["diagnostico"]) if not inline or _is_real_cie10_code(updated["codigo"]): return updated updated["codigo"] = inline["codigo"] if not updated["descripcion"] or _is_technical_description(updated["descripcion"]): updated["descripcion"] = inline["descripcion"] return updated def _apply_catalog_resolution(seed: dict[str, Any], retriever: Any) -> dict[str, Any]: updated = dict(seed) updated["_catalog_available"] = bool( retriever is not None and ( callable(getattr(retriever, "resolver_codigo", None)) or callable(getattr(retriever, "resolver_diagnostico", None)) or callable(getattr(retriever, "resolve_cie10", None)) or hasattr(retriever, "catalogo") ) ) if retriever is None: return updated resolver_codigo = getattr(retriever, "resolver_codigo", None) resolver_diagnostico = getattr(retriever, "resolver_diagnostico", None) resolver_cie10 = getattr(retriever, "resolve_cie10", None) catalog = getattr(retriever, "catalogo", None) resolution = None if _is_real_cie10_code(updated.get("codigo")) and callable(resolver_codigo): resolution = resolver_codigo(updated["codigo"]) elif _is_real_cie10_code(updated.get("codigo")) and callable(resolver_cie10): resolution = resolver_cie10(updated["codigo"]) elif _is_real_cie10_code(updated.get("codigo")) and catalog is not None: entry = catalog.obtener(updated["codigo"]) if callable(getattr(catalog, "obtener", None)) else None if entry is not None: resolution = { "codigo": entry.codigo, "codigo_canonico": entry.codigo_canonico, "descripcion": entry.descripcion, "es_hoja": bool(entry.es_hoja), "padre": entry.padre, "estado_codificacion": "codificado" if entry.es_hoja else "provisional_grupo", } elif not _is_real_cie10_code(updated.get("codigo")) and callable(resolver_diagnostico): try: resolution = resolver_diagnostico(updated.get("diagnostico", "")) except Exception: resolution = None if not isinstance(resolution, dict): if _is_real_cie10_code(updated.get("codigo")) and ( callable(resolver_codigo) or callable(resolver_cie10) or catalog is not None ): updated["codigo_reportado"] = updated.get("codigo", "") updated["codigo"] = "" updated["_resolution"] = { "estado_codificacion": "pendiente_validacion", "candidatos_rechazados": [ { "codigo": updated.get("codigo_reportado", ""), "razon": "código explícito ausente del catálogo local", } ], "razon_revision": "El código explícito no existe en el catálogo CIE-10 local.", } return updated if resolution.get("codigo"): updated["codigo"] = _normalize_cie10_code(resolution.get("codigo")) updated["_resolution"] = resolution return updated def _apply_retriever_cie10(seed: dict[str, Any], retriever: Any) -> dict[str, Any]: # Catalog resolution is deliberately deterministic. Legacy callers may still pass a # retriever exposing ``asignar_codigo_cie10``; it is never used to invent a concept. return dict(seed) def _finalize_normalized_diagnostico( seed: dict[str, Any], source: str, *, occurrence_order: int = 0 ) -> dict[str, Any] | None: codigo_real = _normalize_cie10_code(seed["codigo"]) if _is_real_cie10_code(seed["codigo"]) else "" descripcion = "" if _is_technical_description(seed["descripcion"]) else seed["descripcion"] diagnostico_final = seed["diagnostico"] or descripcion detalle = descripcion or diagnostico_final if not detalle and not codigo_real: return None clave_texto = _normalize_diagnostico_key_text(detalle or diagnostico_final or codigo_real) if not clave_texto and not codigo_real: return None resolution_value = seed.get("_resolution") resolution: dict[str, Any] = resolution_value if isinstance(resolution_value, dict) else {} retriever_available = bool(seed.get("_catalog_available")) estado_concepto_cie10, razones_revision = _resolution_state_and_reasons( code=codigo_real or _normalize_cie10_code(seed.get("codigo_reportado")), resolution=resolution, retriever_available=retriever_available, ) # An explicit human decision remains authoritative over the automatic state. estado_codificacion = seed.get("estado_codificacion") or ( "codificado" if estado_concepto_cie10 == "oficial" else "pendiente_validacion" ) if estado_concepto_cie10 != "oficial": estado_codificacion = ( "provisional_grupo" if resolution and resolution.get("codigo") and not resolution.get("es_hoja") else "pendiente_validacion" ) official_concept = ( normalize_clinical_diagnosis_text(resolution.get("descripcion")) if estado_concepto_cie10 == "oficial" else "" ) description_pairs = _clinical_text_candidates(seed, official_concept=official_concept) clinical_descriptions = _deduplicate_clinical_texts([normalized for _, normalized in description_pairs]) clinical_text = ". ".join(clinical_descriptions) presentation_concept = ( official_concept or clinical_text or normalize_clinical_diagnosis_text(diagnostico_final) ) if not clinical_text: clinical_text = normalize_clinical_diagnosis_text(detalle) if not presentation_concept: presentation_concept = clinical_text reason_list = list( dict.fromkeys([*razones_revision, *(_as_text_list(resolution.get("razones_revision")))]) ) if resolution.get("razon_revision"): reason_list.append(normalize_clinical_diagnosis_text(resolution["razon_revision"])) reason_list = list(dict.fromkeys(reason_list)) evidence_values = seed.get("evidencias") or seed.get("evidence") or [] source_original = str(seed.get("texto_original") or "").strip() descriptions_payload = [ { "texto_original": source_original if index == 0 and source_original else original, "texto_normalizado": normalized, "document_uid": str(seed.get("document_uid") or ""), "document_id": str(seed.get("document_id") or ""), "document_type": str(seed.get("document_type") or source or ""), "filename": str(seed.get("filename") or ""), "page": seed.get("page") or seed.get("pagina"), "section": str(seed.get("section") or seed.get("seccion") or "diagnosticos"), "orden_documental": occurrence_order, "evidencias": list(evidence_values) if isinstance(evidence_values, list) else [], } for index, (original, normalized) in enumerate(description_pairs) ] clave_canonica = f"cie10:{codigo_real}" if codigo_real else f"texto:{clave_texto}" normalized_item = { "key": clave_canonica, "codigo": codigo_real, "codigo_reportado": _normalize_cie10_code(seed.get("codigo_reportado")), "descripcion": clinical_text, "diagnostico": normalize_clinical_diagnosis_text(diagnostico_final or clinical_text), "estado_codificacion": estado_codificacion, "origen": seed.get("origen") or resolution.get("fuente") or "documento_clinico", "descripcion_original": _normalize_whitespace(seed.get("descripcion_original") or diagnostico_final), "descripcion_catalogo": official_concept, "catalog_description": official_concept, "concepto_cie10_oficial": official_concept, "concepto_diagnostico_presentacion": presentation_concept, "descripcion_clinica_normalizada": clinical_text, "descripciones_clinicas": descriptions_payload, "estado_concepto_cie10": estado_concepto_cie10, "razones_revision": reason_list, "razon_revision": "; ".join(reason_list), "codigo_canonico": _normalize_cie10_code(resolution.get("codigo_canonico") or codigo_real), "referencia_catalogo": _normalize_whitespace( resolution.get("referencia_catalogo") or resolution.get("referencia") ), "version_catalogo": _normalize_whitespace(resolution.get("version_catalogo")), "es_hoja": bool(resolution.get("es_hoja")) if resolution else bool(codigo_real), "padre": _normalize_cie10_code(resolution.get("padre")), "candidatos": list(resolution.get("candidatos") or []), "candidatos_rechazados": list(resolution.get("candidatos_rechazados") or []), "razon_seleccion": _normalize_whitespace(resolution.get("razon_seleccion")), "fuente_documental": _normalize_whitespace(seed.get("fuente_documental") or source), "pagina": seed.get("pagina"), "seccion": _normalize_whitespace(seed.get("seccion")), "evidencia_clinica": _normalize_whitespace(seed.get("evidencia_clinica") or diagnostico_final), "procedimiento_origen": _normalize_whitespace(seed.get("procedimiento_origen")), "advertencias": [ _normalize_whitespace(value) for value in seed.get("advertencias") or [] if _normalize_whitespace(value) ], "fuentes": [str(source or "desconocida")], "tipo_contexto": _coerce_tipo_contexto(diagnostico_final or detalle), "clave_canonica": clave_canonica, "clave_codigo": f"cie10:{codigo_real}" if codigo_real else "", "clave_texto": f"texto:{clave_texto}", "texto_original": str(seed.get("texto_original") or diagnostico_final or detalle).strip(), } for key in ( "procedimiento_origen", "evidencia_clinica", "fuente_documental", "pagina", "seccion", "advertencias", "es_pop", "procedimiento_pop_asociado", "procedimientos_pop_asociados", "hallazgo_quirurgico", "descripcion_quirurgica", "fuente_pop", "pagina_pop", "seccion_pop", "extracto_pop", "fuente_cirugia", "pagina_cirugia", "seccion_cirugia", "extracto_cirugia", "evidencias_pop", "evidencias_cirugia", ): if seed.get(key) not in (None, "", []): normalized_item[key] = seed[key] normalized_item["texto_presentacion"] = _build_diagnostico_texto_presentacion(normalized_item) return normalized_item def normalize_cie10_entries( entries: Any, *, retriever: Any = None, source: str = "desconocida", ) -> list[dict[str, Any]]: normalized: list[dict[str, Any]] = [] for raw_item in entries or []: raw = _coerce_raw_diagnostico_item(raw_item) if raw is None: continue seed = _build_diagnostico_seed(raw) seed = _apply_codigo_asignado(seed, raw) seed = _apply_inline_cie10(seed) seed = _apply_catalog_resolution(seed, retriever) seed = _apply_retriever_cie10(seed, retriever) normalized_item = _finalize_normalized_diagnostico(seed, source, occurrence_order=len(normalized)) if normalized_item is not None: normalized.append(normalized_item) return normalized def _build_soat_diagnostico_payload(item: dict[str, Any]) -> list[dict[str, Any]]: fallback_descripcion = item.get("descripcion") or "" candidates = [item.get("cie10_principal"), *(item.get("cie10_secundarios") or [])] payload: list[dict[str, Any]] = [] for cie in candidates: if not isinstance(cie, dict): continue payload.append( { "diagnostico": cie.get("descripcion") or fallback_descripcion, "codigo": cie.get("codigo"), "descripcion": cie.get("descripcion"), } ) return payload def _normalize_soat_diagnosticos(items: Any, *, source: str, retriever: Any = None) -> list[dict[str, Any]]: normalized: list[dict[str, Any]] = [] for item in items or []: if not isinstance(item, dict): continue item_source = str(item.get("fuente") or source).strip() or source payload = _build_soat_diagnostico_payload(item) normalized.extend(normalize_cie10_entries(payload, retriever=retriever, source=item_source)) return normalized def _merge_hallazgo( hallazgos: list[dict[str, Any]], *, tipo: str, item: dict[str, Any], detalle: str, merged_key: str = "", ) -> None: hallazgos.append( { "tipo": tipo, "detalle": detalle, "codigo": item.get("codigo", ""), "descripcion": item.get("descripcion", ""), "diagnostico": item.get("diagnostico", ""), "fuentes": list(item.get("fuentes") or []), "merged_into_key": merged_key, } ) def _prefer_incoming_diagnostico(existing: dict[str, Any], incoming: dict[str, Any]) -> bool: incoming_state = _coding_state_priority(incoming.get("estado_codificacion")) existing_state = _coding_state_priority(existing.get("estado_codificacion")) if incoming_state != existing_state: return incoming_state > existing_state incoming_priority = _source_priority((incoming.get("fuentes") or [""])[0]) existing_priority = _source_priority((existing.get("fuentes") or [""])[0]) if incoming_priority != existing_priority: return incoming_priority > existing_priority return len(str(incoming.get("descripcion") or "")) > len(str(existing.get("descripcion") or "")) def _merge_diagnostico_record(existing: dict[str, Any], incoming: dict[str, Any]) -> None: if _prefer_incoming_diagnostico(existing, incoming): existing["codigo"] = incoming.get("codigo", existing.get("codigo", "")) existing["descripcion"] = incoming.get("descripcion") or existing.get("descripcion", "") existing["diagnostico"] = incoming.get("diagnostico") or existing.get("diagnostico", "") existing["estado_codificacion"] = incoming.get( "estado_codificacion", existing.get("estado_codificacion", "pendiente_validacion") ) existing["texto_original"] = incoming.get("texto_original") or existing.get("texto_original", "") existing["descripcion_original"] = incoming.get("descripcion_original") or existing.get( "descripcion_original", "" ) existing["descripcion_catalogo"] = incoming.get("descripcion_catalogo") or existing.get( "descripcion_catalogo", "" ) for key in ( "codigo_reportado", "codigo_canonico", "es_hoja", "padre", "concepto_cie10_oficial", "catalog_description", "concepto_diagnostico_presentacion", "estado_concepto_cie10", ): if incoming.get(key) not in (None, ""): existing[key] = incoming[key] existing["tipo_contexto"] = _merge_tipo_contexto( existing.get("tipo_contexto", "no_clasificado"), incoming.get("tipo_contexto", "no_clasificado"), ) existing_sources = [str(source) for source in existing.get("fuentes") or []] for source in incoming.get("fuentes") or []: source = str(source) if source not in existing_sources: existing_sources.append(source) existing["fuentes"] = existing_sources descriptions = list(existing.get("descripciones_clinicas") or []) for description in incoming.get("descripciones_clinicas") or []: if not isinstance(description, dict): continue identity = ( _normalize_ascii(description.get("texto_normalizado")).casefold(), str(description.get("document_uid") or description.get("document_id") or ""), description.get("orden_documental", 0), ) if not any( ( _normalize_ascii(current.get("texto_normalizado")).casefold(), str(current.get("document_uid") or current.get("document_id") or ""), current.get("orden_documental", 0), ) == identity for current in descriptions if isinstance(current, dict) ): descriptions.append(description) descriptions.sort(key=lambda item: int(item.get("orden_documental") or 0)) if descriptions: existing["descripciones_clinicas"] = descriptions clinical_texts = _deduplicate_clinical_texts( [str(item.get("texto_normalizado") or "") for item in descriptions if isinstance(item, dict)] ) existing["descripcion_clinica_normalizada"] = ". ".join(clinical_texts) existing["descripcion"] = existing["descripcion_clinica_normalizada"] existing["concepto_diagnostico_presentacion"] = ( existing.get("concepto_cie10_oficial") or existing["descripcion_clinica_normalizada"] ) official = _normalize_whitespace( existing.get("concepto_cie10_oficial") or incoming.get("concepto_cie10_oficial") ) if official: existing["concepto_cie10_oficial"] = official existing["catalog_description"] = official existing["descripcion_catalogo"] = official existing["estado_concepto_cie10"] = "oficial" existing["concepto_diagnostico_presentacion"] = official reasons = list(existing.get("razones_revision") or []) for reason in incoming.get("razones_revision") or []: if reason and reason not in reasons: reasons.append(reason) if reasons: existing["razones_revision"] = reasons existing["razon_revision"] = "; ".join(reasons) for key in ( "candidatos", "candidatos_rechazados", "advertencias", "procedimientos_pop_asociados", "evidencias_pop", "evidencias_cirugia", ): current = list(existing.get(key) or []) for candidate in incoming.get(key) or []: if candidate not in current: current.append(candidate) if current: existing[key] = current for key in ( "procedimiento_origen", "evidencia_clinica", "fuente_documental", "pagina", "seccion", "razon_seleccion", "razon_revision", "origen", "es_pop", "procedimiento_pop_asociado", "procedimientos_pop_asociados", "hallazgo_quirurgico", "descripcion_quirurgica", "fuente_pop", "pagina_pop", "seccion_pop", "extracto_pop", "fuente_cirugia", "pagina_cirugia", "seccion_cirugia", "extracto_cirugia", "evidencias_pop", "evidencias_cirugia", ): if not existing.get(key) and incoming.get(key): existing[key] = incoming[key] existing["texto_presentacion"] = _build_diagnostico_texto_presentacion(existing) def _collect_diagnostico_entries( context: dict[str, Any], *, retriever: Any = None, ) -> list[dict[str, Any]]: historia = context.get("historia") or {} quirurgico = context.get("quirurgico") or {} pdf_draft = coerce_pdf_draft(context.get("pdf_draft")) entries: list[dict[str, Any]] = [] entries.extend( normalize_cie10_entries(historia.get("codigos_cie10"), retriever=retriever, source="historia_clinica") ) entries.extend( normalize_cie10_entries(quirurgico.get("codigos_cie10"), retriever=retriever, source="quirurgico") ) entries.extend( _normalize_soat_diagnosticos(context.get("auto_soat_results"), source="auto_soat", retriever=retriever) ) entries.extend( _normalize_soat_diagnosticos(context.get("qx_agent_results"), source="qx_agent", retriever=retriever) ) entries.extend( _normalize_soat_diagnosticos( context.get("manual_soat_results"), source="manual_soat", retriever=retriever ) ) if not any(context.get(key) for key in ("auto_soat_results", "qx_agent_results", "manual_soat_results")): entries.extend( _normalize_soat_diagnosticos( context.get("codigos_desde_soat"), source="soat", retriever=retriever ) ) entries.extend( normalize_cie10_entries( pdf_draft.get("manual_diagnosticos"), retriever=retriever, source="manual_pdf" ) ) for order, item in enumerate(entries): item["orden_documental"] = order return entries def _diagnostico_order_key(item: dict[str, Any]) -> tuple[int, int, int]: return ( int(item.get("orden_documental") or 0), int((item.get("descripciones_clinicas") or [{}])[0].get("orden_documental") or 0), int(item.get("ocurrencia_documental") or 0), ) def _resolve_diagnostico_target_key( item: dict[str, Any], *, by_code: dict[str, str], by_text: dict[str, str], ) -> tuple[str, str, str]: code_key = str(item.get("clave_codigo") or "").strip() text_key = str(item.get("clave_texto") or "").strip() if code_key and code_key in by_code: return by_code[code_key], code_key, text_key if text_key and text_key in by_text: return by_text[text_key], code_key, text_key return "", code_key, text_key def _index_diagnostico_key( target_key: str, *, code_key: str, text_key: str, by_code: dict[str, str], by_text: dict[str, str], ) -> None: if code_key: by_code[code_key] = target_key if text_key: by_text[text_key] = target_key def _merge_into_consolidated_entry( target_key: str, *, item: dict[str, Any], consolidated: dict[str, dict[str, Any]], hallazgos: list[dict[str, Any]], ) -> None: existing = consolidated[target_key] _merge_diagnostico_record(existing, item) _merge_hallazgo( hallazgos, tipo="duplicado_consolidado", item=item, detalle=f"Se consolidó con {existing.get('texto_presentacion', '')}.", merged_key=target_key, ) def _consolidate_diagnostico_entries( ordered_entries: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: hallazgos: list[dict[str, Any]] = [] consolidated: dict[str, dict[str, Any]] = {} by_code: dict[str, str] = {} by_text: dict[str, str] = {} for item in ordered_entries: key = str(item.get("key") or "").strip() target_key, code_key, text_key = _resolve_diagnostico_target_key( item, by_code=by_code, by_text=by_text, ) if target_key and target_key in consolidated: _merge_into_consolidated_entry( target_key, item=item, consolidated=consolidated, hallazgos=hallazgos, ) _index_diagnostico_key( target_key, code_key=code_key, text_key=text_key, by_code=by_code, by_text=by_text, ) continue consolidated[key] = dict(item) _index_diagnostico_key( key, code_key=code_key, text_key=text_key, by_code=by_code, by_text=by_text, ) diagnosticos_consolidados = list(consolidated.values()) for item in diagnosticos_consolidados: item["texto_presentacion"] = _build_diagnostico_texto_presentacion(item) return diagnosticos_consolidados, hallazgos def _build_pendientes_diagnosticos( diagnosticos_consolidados: list[dict[str, Any]], hallazgos: list[dict[str, Any]], ) -> list[dict[str, Any]]: diagnosticos_pendientes_validacion = [ dict(item) for item in diagnosticos_consolidados if item.get("estado_codificacion") in {"pendiente_validacion", "provisional_grupo", "inferido_revisable"} ] for item in diagnosticos_pendientes_validacion: _merge_hallazgo( hallazgos, tipo=str(item.get("estado_codificacion") or "pendiente_validacion"), item=item, detalle="Diagnóstico visible sin código definitivo; requiere validación clínica.", ) return diagnosticos_pendientes_validacion def build_diagnosticos_consolidados( context: dict[str, Any], *, retriever: Any = None, ) -> dict[str, list[dict[str, Any]]]: all_entries = _collect_diagnostico_entries(context, retriever=retriever) ordered_entries = sorted(all_entries, key=_diagnostico_order_key) diagnosticos_consolidados, hallazgos = _consolidate_diagnostico_entries(ordered_entries) diagnosticos_pendientes_validacion = _build_pendientes_diagnosticos( diagnosticos_consolidados, hallazgos, ) return { "diagnosticos_consolidados": diagnosticos_consolidados, "diagnosticos_pendientes_validacion": diagnosticos_pendientes_validacion, "diagnosticos_consolidacion_hallazgos": hallazgos, } def _coerce_diagnostico_catalog(values: Any) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for item in values or []: if not isinstance(item, dict): continue normalized = { "key": _normalize_whitespace(item.get("key")), "codigo": _normalize_cie10_code(item.get("codigo")), "codigo_reportado": _normalize_cie10_code(item.get("codigo_reportado")), "descripcion": _normalize_whitespace(item.get("descripcion")), "diagnostico": _normalize_whitespace(item.get("diagnostico")), "estado_codificacion": _normalize_whitespace(item.get("estado_codificacion")) or "pendiente_validacion", "fuentes": [str(source).strip() for source in (item.get("fuentes") or []) if str(source).strip()], "tipo_contexto": _normalize_whitespace(item.get("tipo_contexto")) or "no_clasificado", "texto_presentacion": _normalize_whitespace(item.get("texto_presentacion")), "origen": _normalize_whitespace(item.get("origen")) or "documento_clinico", "descripcion_original": _normalize_whitespace(item.get("descripcion_original")), "descripcion_catalogo": _normalize_whitespace(item.get("descripcion_catalogo")), "catalog_description": _normalize_whitespace( item.get("catalog_description") or item.get("concepto_cie10_oficial") ), "concepto_cie10_oficial": _normalize_whitespace(item.get("concepto_cie10_oficial")), "concepto_diagnostico_presentacion": _normalize_whitespace( item.get("concepto_diagnostico_presentacion") ), "descripcion_clinica_normalizada": _normalize_whitespace( item.get("descripcion_clinica_normalizada") ), "descripciones_clinicas": [ dict(description) for description in item.get("descripciones_clinicas") or [] if isinstance(description, dict) ], "estado_concepto_cie10": _normalize_whitespace(item.get("estado_concepto_cie10")) or ("oficial" if item.get("concepto_cie10_oficial") else "pendiente_revision"), "razones_revision": [ _normalize_whitespace(reason) for reason in item.get("razones_revision") or [] if _normalize_whitespace(reason) ], "razon_revision": _normalize_whitespace(item.get("razon_revision")), "codigo_canonico": _normalize_cie10_code(item.get("codigo_canonico")), "referencia_catalogo": _normalize_whitespace(item.get("referencia_catalogo")), "version_catalogo": _normalize_whitespace(item.get("version_catalogo")), "es_hoja": bool(item.get("es_hoja")), "padre": _normalize_cie10_code(item.get("padre")), "candidatos": [ candidate for candidate in item.get("candidatos") or [] if isinstance(candidate, dict) ], "candidatos_rechazados": [ candidate for candidate in item.get("candidatos_rechazados") or [] if isinstance(candidate, dict) ], "razon_seleccion": _normalize_whitespace(item.get("razon_seleccion")), "es_pop": bool(item.get("es_pop")), "procedimiento_pop_asociado": _normalize_whitespace(item.get("procedimiento_pop_asociado")), "procedimientos_pop_asociados": [ _normalize_whitespace(value) for value in item.get("procedimientos_pop_asociados") or [] if _normalize_whitespace(value) ], "hallazgo_quirurgico": _normalize_whitespace(item.get("hallazgo_quirurgico")), "descripcion_quirurgica": _normalize_whitespace(item.get("descripcion_quirurgica")), "fuente_pop": _normalize_whitespace(item.get("fuente_pop")), "pagina_pop": item.get("pagina_pop"), "seccion_pop": _normalize_whitespace(item.get("seccion_pop")), "extracto_pop": _normalize_whitespace(item.get("extracto_pop")), "fuente_cirugia": _normalize_whitespace(item.get("fuente_cirugia")), "pagina_cirugia": item.get("pagina_cirugia"), "seccion_cirugia": _normalize_whitespace(item.get("seccion_cirugia")), "extracto_cirugia": _normalize_whitespace(item.get("extracto_cirugia")), "evidencias_pop": [ evidence for evidence in item.get("evidencias_pop") or [] if isinstance(evidence, dict) ], "evidencias_cirugia": [ evidence for evidence in item.get("evidencias_cirugia") or [] if isinstance(evidence, dict) ], "fuente_documental": _normalize_whitespace( item.get("fuente_documental") or (item.get("fuentes") or [""])[0] ), "pagina": item.get("pagina"), "seccion": _normalize_whitespace(item.get("seccion")), "evidencia_clinica": _normalize_whitespace(item.get("evidencia_clinica")), "procedimiento_origen": _normalize_whitespace(item.get("procedimiento_origen")), "advertencias": [ _normalize_whitespace(value) for value in item.get("advertencias") or [] if _normalize_whitespace(value) ], } for key in ("procedimiento_origen", "evidencia_clinica", "fuente_documental", "seccion"): value = _normalize_whitespace(item.get(key)) if value: normalized[key] = value if item.get("pagina") not in (None, ""): normalized["pagina"] = item.get("pagina") if item.get("advertencias"): normalized["advertencias"] = [ _normalize_whitespace(value) for value in item.get("advertencias") if _normalize_whitespace(value) ] if not normalized["key"]: normalized["key"] = ( f"cie10:{normalized['codigo']}" if normalized["codigo"] else f"texto:{_normalize_diagnostico_key_text(normalized['descripcion'] or normalized['diagnostico'])}" ) if not normalized["descripciones_clinicas"]: clinical = ( normalized["descripcion_clinica_normalizada"] or normalized["descripcion"] or normalized["diagnostico"] ) if clinical: normalized["descripciones_clinicas"] = [ { "texto_original": clinical, "texto_normalizado": normalize_clinical_diagnosis_text(clinical), "document_uid": "", "document_id": "", "document_type": "", "filename": "", "page": None, "section": "diagnosticos", "orden_documental": 0, "evidencias": [], } ] if not normalized["descripcion_clinica_normalizada"]: normalized["descripcion_clinica_normalizada"] = ". ".join( _deduplicate_clinical_texts( [ cast(str, description.get("texto_normalizado", "")) for description in normalized["descripciones_clinicas"] if isinstance(description, dict) and isinstance(description.get("texto_normalizado", ""), str) ] ) ) if not normalized["concepto_cie10_oficial"] and normalized["catalog_description"]: normalized["concepto_cie10_oficial"] = normalized["catalog_description"] normalized["estado_concepto_cie10"] = "oficial" normalized["catalog_description"] = normalized["concepto_cie10_oficial"] normalized["descripcion_catalogo"] = normalized["concepto_cie10_oficial"] normalized["concepto_diagnostico_presentacion"] = ( normalized["concepto_diagnostico_presentacion"] or normalized["concepto_cie10_oficial"] or normalized["descripcion_clinica_normalizada"] or normalized["diagnostico"] ) if not normalized["texto_presentacion"]: normalized["texto_presentacion"] = _build_diagnostico_texto_presentacion(normalized) normalized["auto_seleccionable"] = normalized["estado_concepto_cie10"] == "oficial" or normalized[ "estado_codificacion" ] in {"codificado", "aprobado", "corregido", "validado_automaticamente"} result.append(normalized) return result def _coerce_diagnostico_hallazgos(values: Any) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for item in values or []: if not isinstance(item, dict): continue result.append( { "tipo": _normalize_whitespace(item.get("tipo")), "detalle": _normalize_whitespace(item.get("detalle")), "codigo": _normalize_cie10_code(item.get("codigo")), "descripcion": _normalize_whitespace(item.get("descripcion")), "diagnostico": _normalize_whitespace(item.get("diagnostico")), "fuentes": [ str(source).strip() for source in (item.get("fuentes") or []) if str(source).strip() ], "merged_into_key": _normalize_whitespace(item.get("merged_into_key")), } ) return result