from __future__ import annotations import re import unicodedata from typing import Any from app.case_epicrisis.application.cost_ordering import ( COST_ORDER_VERSION, apply_factura_cost_ordering, attach_and_sort_costs, extract_invoice_cost_lines, normalize_money, sort_publishable_procedures, ) from app.case_epicrisis.application.curation import ( curation_context_payload, curation_from_context, ) from app.case_epicrisis.application.medication_pertinence import ( MedicationInvoiceState, apply_clinical_pertinence_reviews, build_medication_glosa_findings, evaluate_medication_pertinence, ) from app.case_epicrisis.domain.models import ( IncapacidadClinica, PdfPreparationSoatGroup, PdfPreparationUserMetadata, ProcedimientoClinico, ProcedureClassification, RecomendacionMedica, RecomendacionMedicaCategoria, ) from app.services.clinical_document_projection import ( get_structured_document_model, render_document_analysis_html, serialize_analysis_document, ) from app.services.clinical_processing import ( extraer_antecedentes_historia, extraer_medicamentos_historia, extraer_procedimientos_historia, formatear_medicamento_canonico, normalizar_lista_medicamentos, ) from app.services.diagnostic_aids_html_legacy import extract_first_html_paragraph_field from app.services.factura_html_legacy import ( extract_factura_procedimientos_from_html, is_valid_factura_procedimiento, ) from app.services.historia_medicamentos import reconcile_case_medications from app.services.soat_processing import _parse_cie10_lines _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", ) _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) _AID_TYPES = {"imagen", "laboratorio", "procedimiento_diagnostico"} _AID_TEXT_TERMS = ( "rx", "rayos x", "radiografia", "radiografía", "tac", "tomografia", "tomografía", "rm", "resonancia", "ecografia", "ecografía", "ultrasonido", "laboratorio", "hemograma", "creatinina", "uroanalisis", "uroanálisis", "paraclinico", "paraclínico", "ayuda diagnostica", "ayuda diagnóstica", "estudio diagnostico", "estudio diagnóstico", ) _AID_TEXT_PATTERN = re.compile( r"\b(" r"rx|rayos\s+x|radiograf[ií]a|tac|tomograf[ií]a|rm|resonancia|" r"ecograf[ií]a|ultrasonido|laboratorio|hemograma|creatinina|" r"uroan[aá]lisis|paracl[ií]nico|ayuda\s+diagn[oó]stica|estudio\s+diagn[oó]stico" r")\b", flags=re.IGNORECASE, ) _PDF_PRIMARY_DIAGNOSTICO_SOURCES = {"historia_clinica", "quirurgico", "manual_pdf"} _PDF_PRIMARY_AYUDA_SOURCES = {"radiologia", "laboratorio", "factura", "manual_pdf"} _PDF_ANALYTICAL_AID_ALERTS = {"referencia_textual_pendiente_revision", "referencia_qx_pendiente_revision"} _AID_UNINTERPRETED_PLACEHOLDERS = { "", "no interpretado", "no interpretada", "sin interpretacion", "sin interpretación", "pendiente", } RECOMMENDATION_CATEGORY_LABELS = { RecomendacionMedicaCategoria.INCAPACIDAD.value: "Incapacidad", RecomendacionMedicaCategoria.TRATAMIENTO_CUIDADO.value: "Tratamientos y cuidados", RecomendacionMedicaCategoria.RESTRICCION.value: "Restricciones", RecomendacionMedicaCategoria.SEGUIMIENTO_REHABILITACION.value: "Seguimiento y rehabilitación", RecomendacionMedicaCategoria.SIGNO_ALARMA.value: "Signos de alarma", RecomendacionMedicaCategoria.OTRA.value: "Otras indicaciones", } def strip_html_tags(value: str) -> str: return re.sub(r"<[^>]+>", "", value or "").strip() def serialize_doc(doc: dict[str, Any] | None) -> dict[str, Any] | None: return serialize_analysis_document(doc) def build_medication_display_list(items: Any) -> list[str]: return [formatear_medicamento_canonico(item) for item in normalizar_lista_medicamentos(items)] def _format_formulated_medication(item: Any) -> str: name = _normalize_whitespace(getattr(item, "nombre", "") or "") if not name: return "" code = _normalize_whitespace(getattr(item, "codigo", "") or "") posology = _normalize_whitespace( getattr(item, "posologia", "") or " ".join( str(value).strip() for value in ( getattr(item, "via", ""), getattr(item, "frecuencia", ""), ) if str(value or "").strip() ) ) dose = _normalize_whitespace(getattr(item, "dosis", "") or "") quantity = _normalize_whitespace(getattr(item, "cantidad", "") or "") description = " - ".join(value for value in (name, posology, dose, quantity) if value) return f"{code} - {description}" if code else description def extract_recomendaciones_medicas(historia: dict[str, Any] | None) -> list[dict[str, Any]]: """Proyecta solo recomendaciones ya persistidas en la historia estructurada.""" if not isinstance(historia, dict): return [] structured = get_structured_document_model(historia) if structured is None or not hasattr(structured, "recomendaciones_medicas"): return [] recommendations: list[RecomendacionMedica] = [] for item in getattr(structured, "recomendaciones_medicas", []) or []: recommendations.append( RecomendacionMedica( categoria=getattr(item, "categoria", RecomendacionMedicaCategoria.OTRA), indicacion=getattr(item, "indicacion", "") or "", duracion=getattr(item, "periodo_duracion", "") or "", fecha=getattr(item, "fecha", "") or "", pagina=getattr(item, "pagina", None), evidencia=getattr(item, "evidencia", "") or "", ) ) for medication in getattr(structured, "medicamentos", []) or []: if str(getattr(medication, "tipo_uso", "") or "").strip().casefold() != "formulado": continue indication = _format_formulated_medication(medication) if indication: recommendations.append( RecomendacionMedica( categoria=RecomendacionMedicaCategoria.TRATAMIENTO_CUIDADO, indicacion=indication, duracion=getattr(medication, "duracion", "") or "", ) ) result: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for recommendation in recommendations: key = ( recommendation.categoria.value, recommendation.indicacion.casefold(), recommendation.duracion.casefold(), ) if key in seen: continue seen.add(key) result.append(recommendation.model_dump(mode="json", exclude_none=True)) return result def group_recomendaciones_medicas_for_view(items: Any) -> list[dict[str, Any]]: grouped: dict[str, list[dict[str, Any]]] = {category: [] for category in RECOMMENDATION_CATEGORY_LABELS} for raw in items if isinstance(items, list) else []: if not isinstance(raw, dict): continue try: item = RecomendacionMedica.model_validate(raw) except ValueError: continue if not any((item.indicacion, item.duracion, item.fecha)): continue grouped[item.categoria.value].append(item.model_dump(mode="json", exclude_none=True)) return [ {"key": category, "label": label, "items": grouped[category]} for category, label in RECOMMENDATION_CATEGORY_LABELS.items() if grouped[category] ] def normalize_factura_medicamentos(doc: dict[str, Any] | None) -> dict[str, Any] | None: if not isinstance(doc, dict): return doc factura_json = doc.get("factura_json") if not isinstance(factura_json, dict): return doc servicios = factura_json.get("servicios_procedimientos") if not isinstance(servicios, dict): return doc servicios["medicamentos"] = normalizar_lista_medicamentos(servicios.get("medicamentos"), fuente="factura") return doc def _build_factura_procedimiento( item: dict[str, Any] | Any, descripcion: Any = None ) -> dict[str, str] | None: source = item if isinstance(item, dict) else {"codigo_cups": item, "descripcion": descripcion} codigo_normalizado = str(source.get("codigo_cups") or source.get("codigo_referencia") or "").strip() descripcion_normalizada = str(source.get("descripcion") or source.get("concepto") or "").strip() if not is_valid_factura_procedimiento(codigo_normalizado, descripcion_normalizada): return None return { "codigo_cups": codigo_normalizado, "codigo_facturacion": str(source.get("codigo_facturacion") or "").strip(), "codigo_referencia": str(source.get("codigo_referencia") or codigo_normalizado).strip(), "descripcion": descripcion_normalizada, } def _extraer_procedimientos_factura_json(factura_doc: dict[str, Any]) -> list[dict[str, str]]: factura_json = factura_doc.get("factura_json") if not isinstance(factura_json, dict): return [] servicios = factura_json.get("servicios_procedimientos") or {} procedimientos: list[dict[str, str]] = [] for item in servicios.get("procedimientos_quirurgicos") or []: if not isinstance(item, dict): continue procedimiento = _build_factura_procedimiento(item) if procedimiento: procedimientos.append(procedimiento) return procedimientos def _extract_factura_table_html(factura_doc: dict[str, Any]) -> list[dict[str, str]]: factura_html = str(factura_doc.get("analisis_html") or render_document_analysis_html(factura_doc) or "") if not factura_html: return [] return extract_factura_procedimientos_from_html(factura_html) def _extraer_procedimientos_factura_html(factura_doc: dict[str, Any]) -> list[dict[str, str]]: return _extract_factura_table_html(factura_doc) def extraer_procedimientos_factura(factura_doc: dict[str, Any] | None) -> list[dict[str, str]]: if not isinstance(factura_doc, dict): return [] procedimientos_json = _extraer_procedimientos_factura_json(factura_doc) if procedimientos_json: return procedimientos_json return _extraer_procedimientos_factura_html(factura_doc) def _imagenologia_factura_display(item: Any) -> str: if not isinstance(item, dict): return str(item or "").strip() estudio = str(item.get("estudio") or item.get("prueba") or item.get("concepto") or "").strip() descripcion = str(item.get("descripcion") or "").strip() if estudio and descripcion and estudio.lower() != descripcion.lower(): return f"{estudio} - {descripcion}" return estudio or descripcion def extraer_imagenologia_factura(factura_doc: dict[str, Any] | None) -> list[str]: if not isinstance(factura_doc, dict): return [] factura_json = factura_doc.get("factura_json") if not isinstance(factura_json, dict): return [] servicios = factura_json.get("servicios_procedimientos") or {} return _clean_unique_text_list( [_imagenologia_factura_display(item) for item in servicios.get("imagenologia") or []] ) def _extract_html_field(html: str, labels: tuple[str, ...]) -> str: return extract_first_html_paragraph_field(html, labels) def _radiologia_interpretacion(doc: dict[str, Any]) -> str: structured = get_structured_document_model(doc) if structured is not None and hasattr(structured, "conclusion"): conclusion = _normalize_whitespace(getattr(structured, "conclusion", "") or "") if conclusion and conclusion.lower() not in {"...", "[...]", "no especificado", "no disponible"}: return conclusion html = str(doc.get("analisis_html") or doc.get("analisis") or render_document_analysis_html(doc) or "") conclusion = _extract_html_field( html, ( "Conclusión", "Conclusion", "Conclusión/Impresión diagnóstica", "Conclusion/Impresion diagnostica", "Impresión diagnóstica", "Impresion diagnostica", ), ) if conclusion and conclusion.lower() not in {"...", "[...]", "no especificado", "no disponible"}: return conclusion return "No interpretado" def _radiologia_display(doc: Any) -> str: if not isinstance(doc, dict): return str(doc or "").strip() structured = get_structured_document_model(doc) nombre = "" if structured is not None and hasattr(structured, "tipo_estudio"): nombre = _normalize_whitespace(getattr(structured, "tipo_estudio", "") or "") if not nombre: nombre = str(doc.get("nombre_archivo") or doc.get("filename") or doc.get("archivo") or "").strip() if not nombre: nombre = ( _extract_html_field( str(doc.get("analisis_html") or render_document_analysis_html(doc) or ""), ("Tipo de estudio",), ) or "Reporte radiológico" ) return f"{nombre} - {_radiologia_interpretacion(doc)}" def extraer_imagenes_radiologia(radiologia_docs: Any) -> list[str]: if not isinstance(radiologia_docs, list): return [] return _clean_unique_text_list([_radiologia_display(doc) for doc in radiologia_docs]) def extraer_imagenes_diagnosticas( factura_doc: dict[str, Any] | None, radiologia_docs: Any = None, ) -> list[str]: return _clean_unique_text_list( [ *extraer_imagenologia_factura(factura_doc), *extraer_imagenes_radiologia(radiologia_docs), ] ) def _normalize_bool(value: Any) -> bool: if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) normalized = _normalize_ascii(value).lower() return normalized in {"1", "si", "sí", "true", "verdadero", "yes"} def _normalize_ayuda_tipo(value: Any) -> str: normalized = _normalize_ascii(value).lower().replace("-", "_").replace(" ", "_") aliases = { "imagenes": "imagen", "imagenologia": "imagen", "radiologia": "imagen", "radiologico": "imagen", "rx": "imagen", "lab": "laboratorio", "laboratorios": "laboratorio", "procedimiento": "procedimiento_diagnostico", "procedimiento_diagnóstico": "procedimiento_diagnostico", } normalized = aliases.get(normalized, normalized) return normalized if normalized in _AID_TYPES else "imagen" def _normalize_ayuda_key_text(value: Any) -> str: text = _normalize_ascii(value).lower() text = re.sub(r"[^a-z0-9]+", "-", text) return text.strip("-") def _build_ayuda_key( *, tipo: str, nombre: str, fuente: str, documento_id: str = "", ) -> str: base = _normalize_ayuda_key_text(nombre) or "sin-nombre" scope = _normalize_ayuda_key_text(documento_id or fuente) or "sin-fuente" return f"ayuda:{tipo}:{base}:{scope}" def evaluar_ayuda_diagnostica_exportable(ayuda: dict[str, Any], excluded_keys: Any = None) -> bool: excluded = set(_clean_unique_text_list(excluded_keys)) key = _normalize_whitespace(ayuda.get("key")) if key and key in excluded: return False if _normalize_bool(ayuda.get("glosado")): return False return _normalize_ascii(ayuda.get("estado_interpretacion")).lower() != "excluido" def _build_ayuda_diagnostica( *, tipo: str, nombre: Any, concepto: Any = "", fuente: str, documento: dict[str, Any] | None = None, ordenado: bool = False, interpretado: bool = False, facturado: bool = False, glosado: bool = False, estado_interpretacion: str = "", razon_inclusion: str = "", alertas: Any = None, ) -> dict[str, Any] | None: tipo_normalizado = _normalize_ayuda_tipo(tipo) nombre_texto = _normalize_whitespace(nombre) concepto_texto = _normalize_whitespace(concepto) if not nombre_texto: return None documento = documento if isinstance(documento, dict) else {} documento_id = _normalize_whitespace(documento.get("_id") or documento.get("id")) documento_nombre = _normalize_whitespace( documento.get("nombre_archivo") or documento.get("filename") or documento.get("archivo") ) estado = _normalize_ascii(estado_interpretacion).lower() if estado_interpretacion else "" if not estado: estado = "interpretado" if interpretado else "no_interpretado" if estado not in {"interpretado", "no_interpretado", "pendiente_revision", "excluido"}: estado = "pendiente_revision" interpretado = estado == "interpretado" or bool(interpretado) if estado == "no_interpretado" and not concepto_texto: concepto_texto = "No interpretado" normalized_alerts = _clean_unique_text_list(alertas) if estado == "no_interpretado" and "ayuda_no_interpretada" not in normalized_alerts: normalized_alerts.append("ayuda_no_interpretada") if ( _normalize_whitespace(documento.get("texto_no_extraible")) and "texto_no_extraible" not in normalized_alerts ): normalized_alerts.append("texto_no_extraible") ayuda = { "key": _build_ayuda_key( tipo=tipo_normalizado, nombre=nombre_texto, fuente=fuente, documento_id=documento_id, ), "tipo": tipo_normalizado, "nombre": nombre_texto, "concepto": concepto_texto or ("No interpretado" if estado == "no_interpretado" else ""), "fuente": _normalize_whitespace(fuente) or "desconocida", "documento_id": documento_id, "documento_nombre": documento_nombre, "ordenado": bool(ordenado), "interpretado": bool(interpretado), "facturado": bool(facturado), "glosado": bool(glosado), "estado_interpretacion": estado, "razon_inclusion": _normalize_whitespace(razon_inclusion), "alertas": normalized_alerts, } ayuda["exportable"] = evaluar_ayuda_diagnostica_exportable(ayuda) return ayuda def _is_uninterpreted_placeholder(value: Any) -> bool: return _normalize_ascii(value).lower() in _AID_UNINTERPRETED_PLACEHOLDERS def _format_factura_ayuda_nombre(nombre: Any, descripcion: Any) -> str: nombre_texto = _normalize_whitespace(nombre) descripcion_texto = _normalize_whitespace(descripcion) if not nombre_texto: return descripcion_texto if ( descripcion_texto and descripcion_texto.lower() != nombre_texto.lower() and descripcion_texto.lower() not in nombre_texto.lower() ): return f"{nombre_texto} - {descripcion_texto}" return nombre_texto def format_ayuda_diagnostica_nombre_exportable(item: Any) -> str: if not isinstance(item, dict): return _normalize_whitespace(item) nombre = _normalize_whitespace(item.get("nombre")) concepto = _normalize_whitespace(item.get("concepto") or item.get("interpretacion")) fuente = _normalize_ascii(item.get("fuente")).lower() if ( fuente == "factura" and not _normalize_bool(item.get("interpretado")) and concepto and not _is_uninterpreted_placeholder(concepto) and concepto.lower() != nombre.lower() and concepto.lower() not in nombre.lower() ): return f"{nombre} - {concepto}" if nombre else concepto return nombre def format_ayuda_diagnostica_interpretacion_exportable(item: Any) -> str: if not isinstance(item, dict) or not _normalize_bool(item.get("interpretado")): return "" interpretacion = _normalize_whitespace(item.get("concepto") or item.get("interpretacion")) return "" if _is_uninterpreted_placeholder(interpretacion) else interpretacion def _merge_ayuda(existing: dict[str, Any], incoming: dict[str, Any]) -> None: for field in ("ordenado", "interpretado", "facturado", "glosado"): existing[field] = bool(existing.get(field)) or bool(incoming.get(field)) if incoming.get("concepto") and ( not existing.get("concepto") or existing.get("concepto") == "No interpretado" ): existing["concepto"] = incoming.get("concepto", "") if existing.get("estado_interpretacion") != "interpretado": existing["estado_interpretacion"] = incoming.get( "estado_interpretacion", existing.get("estado_interpretacion") ) for field in ("fuente", "documento_id", "documento_nombre", "razon_inclusion"): if not existing.get(field) and incoming.get(field): existing[field] = incoming[field] existing_alerts = _clean_unique_text_list(existing.get("alertas")) for alerta in _clean_unique_text_list(incoming.get("alertas")): if alerta not in existing_alerts: existing_alerts.append(alerta) existing["alertas"] = existing_alerts existing["exportable"] = evaluar_ayuda_diagnostica_exportable(existing) def _append_ayuda(target: list[dict[str, Any]], ayuda: dict[str, Any] | None) -> None: if ayuda is None: return for existing in target: if existing.get("key") == ayuda.get("key"): _merge_ayuda(existing, ayuda) return target.append(ayuda) def _factura_ayudas_diagnosticas(factura_doc: dict[str, Any] | None) -> list[dict[str, Any]]: if not isinstance(factura_doc, dict): return [] servicios = _servicios_factura(factura_doc) ayudas: list[dict[str, Any]] = [] for item in servicios.get("imagenologia") or []: nombre = ( _normalize_whitespace(item.get("estudio") or item.get("prueba") or item.get("concepto")) if isinstance(item, dict) else _imagenologia_factura_display(item) ) descripcion = _normalize_whitespace(item.get("descripcion")) if isinstance(item, dict) else "" _append_ayuda( ayudas, _build_ayuda_diagnostica( tipo="imagen", nombre=_format_factura_ayuda_nombre(nombre, descripcion), concepto="", fuente="factura", documento=factura_doc, ordenado=True, facturado=True, estado_interpretacion="no_interpretado", razon_inclusion="Detectada en imagenología facturada.", ), ) for item in servicios.get("examenes_laboratorio") or []: nombre = ( _normalize_whitespace(item.get("estudio") or item.get("prueba") or item.get("concepto")) if isinstance(item, dict) else _imagenologia_factura_display(item) ) descripcion = _normalize_whitespace(item.get("descripcion")) if isinstance(item, dict) else "" _append_ayuda( ayudas, _build_ayuda_diagnostica( tipo="laboratorio", nombre=_format_factura_ayuda_nombre(nombre, descripcion), concepto="", fuente="factura", documento=factura_doc, ordenado=True, facturado=True, estado_interpretacion="no_interpretado", razon_inclusion="Detectada en exámenes de laboratorio facturados.", ), ) return ayudas def _radiologia_ayuda_diagnostica(doc: dict[str, Any]) -> dict[str, Any] | None: structured = get_structured_document_model(doc) html = str(doc.get("analisis_html") or doc.get("analisis") or render_document_analysis_html(doc) or "") nombre = "" if structured is not None and hasattr(structured, "tipo_estudio"): nombre = _normalize_whitespace(getattr(structured, "tipo_estudio", "") or "") if not nombre: nombre = _normalize_whitespace(doc.get("nombre_archivo") or doc.get("filename")) if not nombre: nombre = _extract_html_field(html, ("Tipo de estudio", "Estudio", "Examen")) or "Reporte radiológico" concepto = _radiologia_interpretacion(doc) interpretado = concepto != "No interpretado" return _build_ayuda_diagnostica( tipo="imagen", nombre=nombre, concepto=concepto, fuente="radiologia", documento=doc, ordenado=True, interpretado=interpretado, estado_interpretacion="interpretado" if interpretado else "no_interpretado", razon_inclusion="Documento radiológico asociado al caso.", ) def _laboratorio_ayuda_diagnostica(doc: dict[str, Any]) -> dict[str, Any] | None: structured = get_structured_document_model(doc) html = str(doc.get("analisis_html") or doc.get("analisis") or render_document_analysis_html(doc) or "") nombre = "" concepto = "" if structured is not None and hasattr(structured, "tipo_examen"): nombre = _normalize_whitespace(getattr(structured, "tipo_examen", "") or "") concepto = _normalize_whitespace( getattr(structured, "interpretacion", "") or getattr(structured, "observaciones", "") or "" ) if not nombre: nombre = ( _extract_html_field(html, ("Tipo de examen", "Examen", "Prueba", "Nombre del examen")) or _normalize_whitespace(doc.get("nombre_archivo") or doc.get("filename")) or "Reporte de laboratorio" ) if not concepto: concepto = _extract_html_field( html, ( "Interpretación", "Interpretacion", "Conclusión", "Conclusion", "Observaciones", "Resumen", ), ) interpretado = bool(concepto) return _build_ayuda_diagnostica( tipo="laboratorio", nombre=nombre, concepto=concepto or "No interpretado", fuente="laboratorio", documento=doc, ordenado=True, interpretado=interpretado, estado_interpretacion="interpretado" if interpretado else "no_interpretado", razon_inclusion="Documento de laboratorio asociado al caso.", ) def _tipo_ayuda_desde_texto(texto: str) -> str: normalized = _normalize_ascii(texto).lower() if any(term in normalized for term in ("laboratorio", "hemograma", "creatinina", "uroanalisis")): return "laboratorio" if any(term in normalized for term in ("endoscopia", "electrocardiograma", "ecg")): return "procedimiento_diagnostico" return "imagen" def _sentencias_con_ayudas(texto: Any) -> list[str]: plain = _normalize_whitespace(texto) if not plain: return [] sentences = re.split(r"(?<=[.;:\n])\s+", plain) matches: list[str] = [] for sentence in sentences: sentence = sentence.strip(" -.;:") if not sentence or len(sentence) < 4: continue if _AID_TEXT_PATTERN.search(sentence): matches.append(sentence[:220]) return _clean_unique_text_list(matches) def _referencias_ayudas_desde_documento(doc: Any, fuente: str) -> list[dict[str, Any]]: if not isinstance(doc, dict): return [] structured = get_structured_document_model(doc) if structured is not None and hasattr(structured, "referencias_diagnosticas"): ayudas: list[dict[str, Any]] = [] for item in getattr(structured, "referencias_diagnosticas", []) or []: _append_ayuda( ayudas, _build_ayuda_diagnostica( tipo=_normalize_ayuda_tipo(getattr(item, "tipo", "imagen")), nombre=_normalize_whitespace(getattr(item, "nombre", "")), concepto=_normalize_whitespace(getattr(item, "concepto", "")), fuente=fuente, documento=doc, ordenado=True, interpretado=bool(getattr(item, "concepto", "")), estado_interpretacion="interpretado" if getattr(item, "concepto", "") else "no_interpretado", razon_inclusion="Referencia diagnóstica estructurada asociada al documento.", ), ) if ayudas: return ayudas text = "\n".join( filter( None, [ str(doc.get("descripcion") or ""), strip_html_tags(str(doc.get("analisis_html") or render_document_analysis_html(doc) or "")), ], ) ) ayudas: list[dict[str, Any]] = [] if not _normalize_whitespace(text): _append_ayuda( ayudas, _build_ayuda_diagnostica( tipo="procedimiento_diagnostico", nombre=_normalize_whitespace(doc.get("nombre_archivo")) or "Documento sin texto extraíble", concepto="No interpretado", fuente=fuente, documento={**doc, "texto_no_extraible": True}, estado_interpretacion="pendiente_revision", razon_inclusion="El documento no contiene texto extraíble por ReadPdf.read.", alertas=["texto_no_extraible"], ), ) return ayudas for sentence in _sentencias_con_ayudas(text): _append_ayuda( ayudas, _build_ayuda_diagnostica( tipo=_tipo_ayuda_desde_texto(sentence), nombre=sentence, concepto="Referencia textual sin interpretación estructurada.", fuente=fuente, documento=doc, ordenado=True, estado_interpretacion="pendiente_revision", razon_inclusion="Referencia textual conservadora en documento clínico.", alertas=["referencia_textual_pendiente_revision"], ), ) return ayudas def _referencias_ayudas_desde_qx(context: dict[str, Any]) -> list[dict[str, Any]]: text = "\n".join( [ str(context.get("hallazgos_quirurgicos") or ""), str(context.get("descripcion_procedimiento") or ""), ] ) ayudas: list[dict[str, Any]] = [] for sentence in _sentencias_con_ayudas(text): _append_ayuda( ayudas, _build_ayuda_diagnostica( tipo=_tipo_ayuda_desde_texto(sentence), nombre=sentence, concepto="Referencia desde hallazgos/descripción QX pendiente de revisión.", fuente="agente_hallazgos_qx", estado_interpretacion="pendiente_revision", razon_inclusion="Referencia diagnóstica mencionada en contexto quirúrgico.", alertas=["referencia_qx_pendiente_revision"], ), ) return ayudas def _coerce_manual_ayuda_item(item: Any) -> dict[str, Any] | None: if isinstance(item, dict): nombre = item.get("nombre") or item.get("texto") or item.get("descripcion") ayuda = _build_ayuda_diagnostica( tipo=item.get("tipo") or "imagen", nombre=nombre, concepto=item.get("concepto") or item.get("interpretacion") or "", fuente=item.get("fuente") or "manual_pdf", documento=None, ordenado=_normalize_bool(item.get("ordenado")), interpretado=_normalize_bool(item.get("interpretado")), facturado=_normalize_bool(item.get("facturado")), glosado=_normalize_bool(item.get("glosado")), estado_interpretacion=item.get("estado_interpretacion") or "pendiente_revision", razon_inclusion=item.get("razon_inclusion") or "Agregada manualmente para exportación.", alertas=item.get("alertas"), ) if ayuda and _normalize_whitespace(item.get("key")): ayuda["key"] = _normalize_whitespace(item.get("key")) return ayuda text = _normalize_whitespace(item) if not text: return None parts = [part.strip() for part in text.split("|")] if len(parts) >= 2: tipo = parts[0] nombre = parts[1] concepto = " | ".join(parts[2:]).strip() else: tipo = _tipo_ayuda_desde_texto(text) nombre = text concepto = "" return _build_ayuda_diagnostica( tipo=tipo, nombre=nombre, concepto=concepto or "Agregada manualmente.", fuente="manual_pdf", estado_interpretacion="pendiente_revision", razon_inclusion="Agregada manualmente para exportación.", ) def coerce_ayudas_diagnosticas(values: Any) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for item in values or []: if not isinstance(item, dict): ayuda = _coerce_manual_ayuda_item(item) else: ayuda = _build_ayuda_diagnostica( tipo=item.get("tipo") or "imagen", nombre=item.get("nombre") or item.get("texto") or item.get("descripcion"), concepto=item.get("concepto") or item.get("interpretacion") or "", fuente=item.get("fuente") or "desconocida", documento={ "_id": item.get("documento_id"), "nombre_archivo": item.get("documento_nombre"), }, ordenado=_normalize_bool(item.get("ordenado")), interpretado=_normalize_bool(item.get("interpretado")), facturado=_normalize_bool(item.get("facturado")), glosado=_normalize_bool(item.get("glosado")), estado_interpretacion=item.get("estado_interpretacion") or "", razon_inclusion=item.get("razon_inclusion") or "", alertas=item.get("alertas"), ) if ayuda and _normalize_whitespace(item.get("key")): ayuda["key"] = _normalize_whitespace(item.get("key")) if ayuda and isinstance(item.get("costo_facturado"), dict): ayuda["costo_facturado"] = dict(item["costo_facturado"]) _append_ayuda(result, ayuda) return result def format_ayuda_diagnostica_presentacion(item: Any) -> str: if not isinstance(item, dict): return _normalize_whitespace(item) nombre = format_ayuda_diagnostica_nombre_exportable(item) concepto = _normalize_whitespace(item.get("concepto")) if ( concepto and not _is_uninterpreted_placeholder(concepto) and concepto.lower() != nombre.lower() and concepto.lower() not in nombre.lower() ): return f"{nombre} - {concepto}" return nombre def build_ayudas_diagnosticas(context: dict[str, Any]) -> list[dict[str, Any]]: context = context or {} ayudas: list[dict[str, Any]] = [] for ayuda in _factura_ayudas_diagnosticas(context.get("factura")): _append_ayuda(ayudas, ayuda) for doc in context.get("radiologia") or []: _append_ayuda(ayudas, _radiologia_ayuda_diagnostica(doc)) for doc in context.get("laboratorio") or []: _append_ayuda(ayudas, _laboratorio_ayuda_diagnostica(doc)) for doc_key, source in ( ("historia", "historia_clinica"), ("quirurgico", "quirurgico"), ): for ayuda in _referencias_ayudas_desde_documento(context.get(doc_key), source): _append_ayuda(ayudas, ayuda) for doc in context.get("generico") or []: for ayuda in _referencias_ayudas_desde_documento(doc, "generico"): _append_ayuda(ayudas, ayuda) for ayuda in _referencias_ayudas_desde_qx(context): _append_ayuda(ayudas, ayuda) for ayuda in coerce_ayudas_diagnosticas( (context.get("pdf_draft") or {}).get("manual_ayudas_diagnosticas") ): _append_ayuda(ayudas, ayuda) supported_names = { (_normalize_ayuda_tipo(item.get("tipo")), _normalize_ayuda_key_text(item.get("nombre"))) for item in ayudas if item.get("fuente") != "factura" } for item in ayudas: signature = (_normalize_ayuda_tipo(item.get("tipo")), _normalize_ayuda_key_text(item.get("nombre"))) if item.get("fuente") == "factura" and signature not in supported_names: alerts = _clean_unique_text_list(item.get("alertas")) if "factura_sin_soporte_clinico_asociado" not in alerts: alerts.append("factura_sin_soporte_clinico_asociado") item["alertas"] = alerts excluded = (context.get("pdf_draft") or {}).get("excluded_ayudas_diagnosticas_keys") for item in ayudas: item["exportable"] = evaluar_ayuda_diagnostica_exportable(item, excluded) return ayudas def merge_soat_results( existing: list[dict[str, Any]] | None, nuevos: list[dict[str, Any]] | None ) -> list[dict[str, Any]]: merged: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() for item in (existing or []) + (nuevos or []): if not isinstance(item, dict): continue key = ( str(item.get("codigo_soat") or "").strip().lower(), str(item.get("descripcion") or "").strip().lower(), ) if key in seen: continue seen.add(key) merged.append(dict(item)) return merged def default_pdf_draft() -> dict[str, Any]: return { "excluded_diagnosticos": [], "excluded_diagnostico_keys": [], "manual_diagnosticos": [], "manual_diagnostico_entries": [], "excluded_procedimientos": [], "manual_procedimientos": [], "manual_procedure_entries": [], "procedure_classification_overrides": {}, "excluded_medicamentos": [], "excluded_medicamento_keys": [], "manual_medicamentos": [], "manual_medicamento_entries": [], "excluded_ayudas_diagnosticas_keys": [], "manual_ayudas_diagnosticas": [], "excluded_imagenes_diagnosticas": [], "manual_imagenes_diagnosticas": [], "hallazgos_extra": "", "descripcion_procedimiento_extra": "", } def _clean_unique_text_list(values: Any) -> list[str]: if not isinstance(values, list): return [] result: list[str] = [] seen: set[str] = set() for item in values: text = str(item or "").strip() if not text: continue key = text.lower() if key in seen: continue seen.add(key) result.append(text) return result def _normalize_manual_soat_results(values: Any) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] for item in values or []: if not isinstance(item, dict): continue descripcion = str(item.get("descripcion") or "").strip() codigo_soat = str(item.get("codigo_soat") or "").strip() if not descripcion: continue results.append({**item, "descripcion": descripcion, "codigo_soat": codigo_soat}) return merge_soat_results([], results) def _normalize_manual_procedure_entries(values: Any, legacy_values: Any = None) -> list[dict[str, str]]: entries: list[dict[str, str]] = [] seen_keys: set[str] = set() for item in values if isinstance(values, list) else []: if not isinstance(item, dict): continue text = str(item.get("value") or item.get("texto") or item.get("descripcion") or "").strip() key = str(item.get("key") or "").strip() if not text or not key or key in seen_keys: continue seen_keys.add(key) entries.append({"key": key, "value": text}) for text in _clean_unique_text_list(legacy_values): if any(entry["value"].casefold() == text.casefold() for entry in entries): continue key = f"manual:{_procedure_key(text)}" if key in seen_keys: continue seen_keys.add(key) entries.append({"key": key, "value": text}) return entries def _normalize_manual_text_entries(values: Any, legacy_values: Any, *, prefix: str) -> list[dict[str, Any]]: entries: list[dict[str, Any]] = [] seen_keys: set[str] = set() for item in values if isinstance(values, list) else []: if not isinstance(item, dict): continue value = str(item.get("value") or item.get("texto") or "").strip() key = str(item.get("key") or "").strip() if not value or not key or key in seen_keys: continue seen_keys.add(key) entries.append({**item, "key": key, "value": value}) for value in _clean_unique_text_list(legacy_values): if any(str(entry["value"]).casefold() == value.casefold() for entry in entries): continue normalized = re.sub(r"[^a-z0-9]+", "-", _normalize_ascii(value).casefold()).strip("-") key = f"manual:{prefix}:{normalized or len(entries) + 1}" if key in seen_keys: continue seen_keys.add(key) entries.append({"key": key, "value": value}) return entries def coerce_pdf_preparation_user_metadata(values: Any) -> dict[str, Any]: if not isinstance(values, dict): return PdfPreparationUserMetadata().model_dump(mode="json") try: return PdfPreparationUserMetadata.model_validate(values).model_dump(mode="json") except ValueError: valid = [] for item in values.get("procedure_soat_groups") or []: try: valid.append(PdfPreparationSoatGroup.model_validate(item)) except (TypeError, ValueError): continue return PdfPreparationUserMetadata(procedure_soat_groups=valid).model_dump(mode="json") def _normalize_procedure_classification_overrides(values: Any) -> dict[str, str]: if not isinstance(values, dict): return {} valid = { ProcedureClassification.QUIRURGICO.value, ProcedureClassification.NO_QUIRURGICO.value, } return { str(key).strip(): str(value).strip() for key, value in values.items() if str(key).strip() and str(value).strip() in valid } def coerce_pdf_draft(values: Any) -> dict[str, Any]: draft = default_pdf_draft() if isinstance(values, dict): manual_procedimientos = _clean_unique_text_list(values.get("manual_procedimientos")) manual_procedure_entries = _normalize_manual_procedure_entries( values.get("manual_procedure_entries"), manual_procedimientos ) if manual_procedure_entries: manual_procedimientos = [item["value"] for item in manual_procedure_entries] manual_diagnostico_entries = _normalize_manual_text_entries( values.get("manual_diagnostico_entries"), values.get("manual_diagnosticos"), prefix="diagnostico" ) manual_medicamento_entries = _normalize_manual_text_entries( values.get("manual_medicamento_entries"), values.get("manual_medicamentos"), prefix="medicamento" ) draft.update( { "excluded_diagnosticos": _clean_unique_text_list(values.get("excluded_diagnosticos")), "excluded_diagnostico_keys": _clean_unique_text_list(values.get("excluded_diagnostico_keys")), "manual_diagnosticos": [item["value"] for item in manual_diagnostico_entries], "manual_diagnostico_entries": manual_diagnostico_entries, "excluded_procedimientos": _clean_unique_text_list(values.get("excluded_procedimientos")), "manual_procedimientos": manual_procedimientos, "manual_procedure_entries": manual_procedure_entries, "procedure_classification_overrides": ( _normalize_procedure_classification_overrides( values.get("procedure_classification_overrides") ) ), "excluded_medicamentos": _clean_unique_text_list(values.get("excluded_medicamentos")), "excluded_medicamento_keys": _clean_unique_text_list(values.get("excluded_medicamento_keys")), "manual_medicamentos": [item["value"] for item in manual_medicamento_entries], "manual_medicamento_entries": manual_medicamento_entries, "excluded_ayudas_diagnosticas_keys": _clean_unique_text_list( values.get("excluded_ayudas_diagnosticas_keys") ), "manual_ayudas_diagnosticas": coerce_ayudas_diagnosticas( values.get("manual_ayudas_diagnosticas") ), "excluded_imagenes_diagnosticas": _clean_unique_text_list( values.get("excluded_imagenes_diagnosticas") ), "manual_imagenes_diagnosticas": _clean_unique_text_list( values.get("manual_imagenes_diagnosticas") ), "hallazgos_extra": str(values.get("hallazgos_extra") or "").strip(), "descripcion_procedimiento_extra": str( values.get("descripcion_procedimiento_extra") or "" ).strip(), } ) if not draft["manual_ayudas_diagnosticas"] and draft["manual_imagenes_diagnosticas"]: draft["manual_ayudas_diagnosticas"] = coerce_ayudas_diagnosticas( draft["manual_imagenes_diagnosticas"] ) return draft def _normalize_whitespace(value: Any) -> str: return re.sub(r"\s+", " ", strip_html_tags(str(value or ""))).strip() def _normalize_ascii(value: Any) -> str: text = _normalize_whitespace(value) normalized = unicodedata.normalize("NFKD", text) return "".join(char for char in normalized if not unicodedata.combining(char)) def _is_real_cie10_code(value: Any) -> bool: return bool(_CIE10_CODE_PATTERN.match(str(value or "").strip().upper())) def _normalize_cie10_code(value: Any) -> str: return str(value or "").strip().upper() def _is_technical_code(value: Any) -> bool: return _normalize_cie10_code(value).lower() in _TECHNICAL_CODE_VALUES 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 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")) descripcion = _normalize_whitespace( item.get("descripcion_original") or item.get("diagnostico") or item.get("descripcion") ) diagnostico = _normalize_whitespace(item.get("diagnostico")) detalle = descripcion or diagnostico or "Diagnóstico sin descripción" if codigo: return f"{codigo} - {detalle}" return f"PENDIENTE - {detalle}" 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")), } for key in ( "origen", "estado_codificacion", "procedimiento_origen", "evidencia_clinica", "fuente_documental", "pagina", "seccion", "advertencias", ): 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) if retriever is None: return updated resolver_codigo = getattr(retriever, "resolver_codigo", None) resolver_diagnostico = getattr(retriever, "resolver_diagnostico", None) resolution = None if _is_real_cie10_code(updated.get("codigo")) and callable(resolver_codigo): resolution = resolver_codigo(updated["codigo"]) 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): 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")) if resolution.get("descripcion") and not updated.get("descripcion"): updated["descripcion"] = _normalize_whitespace(resolution.get("descripcion")) updated["_resolution"] = resolution return updated def _apply_retriever_cie10(seed: dict[str, Any], retriever: Any) -> dict[str, Any]: updated = dict(seed) if ( retriever is None or not updated["diagnostico"] or _is_real_cie10_code(updated["codigo"]) or updated.get("_resolution") ): return updated try: texto_cie10 = retriever.asignar_codigo_cie10(updated["diagnostico"], temperature=0.0, k=30) codigos_parseados = _parse_cie10_lines(texto_cie10) except Exception: return updated if not codigos_parseados: return updated updated["codigo"] = _normalize_cie10_code(codigos_parseados[0].get("codigo")) updated["descripcion"] = _normalize_whitespace(codigos_parseados[0].get("descripcion")) updated["origen"] = "llm_legacy" return updated def _finalize_normalized_diagnostico(seed: dict[str, Any], source: str) -> 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 {} estado_codificacion = seed.get("estado_codificacion") or resolution.get("estado_codificacion") or ( "codificado" if codigo_real else "pendiente_validacion" ) 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": detalle, "diagnostico": diagnostico_final, "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": _normalize_whitespace(resolution.get("descripcion")), "codigo_canonico": _normalize_cie10_code(resolution.get("codigo_canonico") or codigo_real), "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")), "razon_revision": _normalize_whitespace(resolution.get("razon_revision")), "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": diagnostico_final or detalle, } for key in ( "procedimiento_origen", "evidencia_clinica", "fuente_documental", "pagina", "seccion", "advertencias", ): 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) 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) -> 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, 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", "" ) 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 for key in ( "candidatos", "candidatos_rechazados", "advertencias", ): 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", ): 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")) entries.extend(_normalize_soat_diagnosticos(context.get("qx_agent_results"), source="qx_agent")) entries.extend(_normalize_soat_diagnosticos(context.get("manual_soat_results"), source="manual_soat")) 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")) entries.extend( normalize_cie10_entries( pdf_draft.get("manual_diagnosticos"), retriever=retriever, source="manual_pdf" ) ) return entries def _diagnostico_order_key(item: dict[str, Any]) -> tuple[int, int, int]: return ( _coding_state_priority(item.get("estado_codificacion")), _source_priority((item.get("fuentes") or [""])[0]), len(str(item.get("descripcion") or "")), ) 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, reverse=True) 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")), "codigo_canonico": _normalize_cie10_code(item.get("codigo_canonico")), "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")), "razon_revision": _normalize_whitespace(item.get("razon_revision")), "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["texto_presentacion"]: normalized["texto_presentacion"] = _build_diagnostico_texto_presentacion(normalized) 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 def refresh_diagnostico_context( context: dict[str, Any], *, retriever: Any = None, ) -> dict[str, Any]: refreshed = dict(context or {}) for doc_key, source in (("historia", "historia_clinica"), ("quirurgico", "quirurgico")): doc = refreshed.get(doc_key) if not isinstance(doc, dict): continue doc_copy = dict(doc) doc_copy["codigos_cie10"] = normalize_cie10_entries( doc_copy.get("codigos_cie10"), retriever=retriever, source=source, ) refreshed[doc_key] = doc_copy refreshed["pdf_draft"] = coerce_pdf_draft(refreshed.get("pdf_draft")) refreshed.update(build_diagnosticos_consolidados(refreshed, retriever=retriever)) refreshed["ayudas_diagnosticas"] = build_ayudas_diagnosticas(refreshed) refreshed["imagenes_diagnosticas"] = [ format_ayuda_diagnostica_presentacion(item) for item in refreshed["ayudas_diagnosticas"] if item.get("tipo") == "imagen" ] refreshed["pdf_draft"] = reconcile_pdf_draft(refreshed.get("pdf_draft"), refreshed) excluded_ayudas = refreshed["pdf_draft"].get("excluded_ayudas_diagnosticas_keys") for ayuda in refreshed["ayudas_diagnosticas"]: ayuda["exportable"] = evaluar_ayuda_diagnostica_exportable(ayuda, excluded_ayudas) refreshed.update(build_pdf_export_blocks(refreshed)) return refreshed def _normalize_optional_doc(value: Any) -> dict[str, Any] | None: return dict(value) if isinstance(value, dict) else None def _normalize_dict_list(value: Any) -> list[dict[str, Any]]: return [item for item in (value or []) if isinstance(item, dict)] def _normalize_text_list(value: Any) -> list[str]: return _clean_unique_text_list(value) def _normalize_recomendaciones_medicas(value: Any) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for raw in value if isinstance(value, list) else []: if not isinstance(raw, dict): continue try: item = RecomendacionMedica.model_validate(raw) except ValueError: continue key = (item.categoria.value, item.indicacion.casefold(), item.duracion.casefold()) if key in seen: continue seen.add(key) result.append(item.model_dump(mode="json", exclude_none=True)) return result def _normalize_historia_clinica_base(normalized: dict[str, Any]) -> None: historia = normalized.get("historia") historia_html = str( (historia or {}).get("analisis_html") or render_document_analysis_html(historia) or "" ).strip() antecedentes_hc = normalized.get("antecedentes_hc") if isinstance(antecedentes_hc, list): normalized["antecedentes_hc"] = _normalize_text_list(antecedentes_hc) else: normalized["antecedentes_hc"] = [] if historia_html and not normalized["antecedentes_hc"]: normalized["antecedentes_hc"] = _normalize_text_list( extraer_antecedentes_historia(historia or historia_html) ) if normalized["antecedentes_hc"]: normalized["antecedentes"] = normalized["antecedentes_hc"] antecedentes_estructurados = normalized.get("antecedentes_hc_estructurados") normalized["antecedentes_hc_estructurados"] = ( [dict(item) for item in antecedentes_estructurados if isinstance(item, dict)] if isinstance(antecedentes_estructurados, list) else [] ) procedimientos_hc = normalized.get("procedimientos_hc") if isinstance(procedimientos_hc, list): normalized["procedimientos_hc"] = _normalize_text_list(procedimientos_hc) else: normalized["procedimientos_hc"] = [] if historia_html and not normalized["procedimientos_hc"]: normalized["procedimientos_hc"] = _normalize_text_list( extraer_procedimientos_historia(historia or historia_html) ) medicamentos_hc = normalized.get("medicamentos_hc") if isinstance(medicamentos_hc, list): normalized["medicamentos_hc"] = normalizar_lista_medicamentos( medicamentos_hc, fuente="historia_clinica", ) else: normalized["medicamentos_hc"] = [] if historia and not normalized["medicamentos_hc"]: normalized["medicamentos_hc"] = normalizar_lista_medicamentos( extraer_medicamentos_historia(historia or historia_html), fuente="historia_clinica", ) medicamentos_hc_display = normalized.get("medicamentos_hc_display") if isinstance(medicamentos_hc_display, list): normalized["medicamentos_hc_display"] = _normalize_text_list(medicamentos_hc_display) else: normalized["medicamentos_hc_display"] = [] if normalized["medicamentos_hc"] and not normalized["medicamentos_hc_display"]: normalized["medicamentos_hc_display"] = build_medication_display_list(normalized["medicamentos_hc"]) def _normalize_factura_base(normalized: dict[str, Any]) -> None: normalized["factura"] = normalize_factura_medicamentos(normalized.get("factura")) normalized["procedimientos_factura"] = [ item for item in (normalized.get("procedimientos_factura") or []) if isinstance(item, dict) and is_valid_factura_procedimiento( item.get("codigo_cups") or item.get("codigo_referencia"), item.get("descripcion") or item.get("concepto"), ) ] if isinstance(normalized.get("factura"), dict) and not normalized["procedimientos_factura"]: normalized["procedimientos_factura"] = extraer_procedimientos_factura(normalized["factura"]) def _medication_invoice_state(factura: Any) -> MedicationInvoiceState: if not isinstance(factura, dict): return MedicationInvoiceState.AUSENTE factura_json = factura.get("factura_json") if not isinstance(factura_json, dict): return MedicationInvoiceState.INSUFICIENTE servicios = factura_json.get("servicios_procedimientos") if not isinstance(servicios, dict): return MedicationInvoiceState.INSUFICIENTE return MedicationInvoiceState.VALIDA def _medication_support_documents(context: dict[str, Any]) -> list[dict[str, str]]: support: list[dict[str, str]] = [] groups = ( ("historia_clinica", [context.get("historia")]), ("quirurgico", [context.get("quirurgico")]), ("generico", context.get("generico") or []), ("radiologia", context.get("radiologia") or []), ("laboratorio", context.get("laboratorio") or []), ) for document_type, documents in groups: for document in documents: if not isinstance(document, dict): continue text = _normalize_whitespace( document.get("descripcion") or document.get("texto_extraido") or document.get("texto") or strip_html_tags( document.get("analisis_html") or render_document_analysis_html(document) or "" ) ) if not text: continue filename = _normalize_whitespace(document.get("nombre_archivo")) support.append( { "fuente": f"{document_type}: {filename}" if filename else document_type, "texto": text, } ) return support def _previous_clinical_medication_reviews(catalog: Any) -> list[dict[str, Any]]: reviews: list[dict[str, Any]] = [] for item in catalog if isinstance(catalog, list) else []: if not isinstance(item, dict) or not isinstance(item.get("pertinencia"), dict): continue assessment = item["pertinencia"] if "llm" not in str(assessment.get("origen") or ""): continue reviews.append( { "key": str(item.get("key") or ""), "estado": str(assessment.get("estado") or ""), "justificacion": str(assessment.get("justificacion_clinica") or ""), "evidence_ids": list(assessment.get("evidencias_clinicas_ids") or []), } ) return reviews def normalize_context_payload(context: dict[str, Any]) -> dict[str, Any]: normalized = dict(context or {}) previous_medication_catalog = normalized.get("medicamentos_caso") normalized["historia"] = _normalize_optional_doc(normalized.get("historia")) normalized["quirurgico"] = _normalize_optional_doc(normalized.get("quirurgico")) normalized["factura"] = _normalize_optional_doc(normalized.get("factura")) normalized["prescripcion"] = _normalize_dict_list(normalized.get("prescripcion")) raw_integrated_summary = normalized.get("resumen_clinico_integrado") integrated_summary = dict(raw_integrated_summary) if isinstance(raw_integrated_summary, dict) else {} summary_text = str( integrated_summary.get("texto") or (normalized.get("metadatos_hc") or {}).get("resumen") or "" ).strip() paragraphs = _normalize_text_list(integrated_summary.get("parrafos")) if not paragraphs and summary_text: paragraphs = [summary_text] integrated_summary.update( { "estado": str(integrated_summary.get("estado") or "solo_historia").strip(), "texto": summary_text, "parrafos": paragraphs, "fuentes": _normalize_dict_list(integrated_summary.get("fuentes")), "advertencia": str(integrated_summary.get("advertencia") or "").strip() or None, } ) normalized["resumen_clinico_integrado"] = integrated_summary normalized["metadatos_hc"] = dict(normalized.get("metadatos_hc") or {}) normalized["metadatos_hc"]["resumen"] = summary_text normalized["manual_soat_results"] = _normalize_manual_soat_results(normalized.get("manual_soat_results")) normalized["auto_soat_results"] = merge_soat_results([], normalized.get("auto_soat_results")) normalized["qx_agent_results"] = merge_soat_results([], normalized.get("qx_agent_results")) normalized["codigos_desde_soat"] = merge_soat_results([], normalized.get("codigos_desde_soat")) normalized["pdf_draft"] = coerce_pdf_draft(normalized.get("pdf_draft")) normalized["pdf_preparation_user_metadata"] = coerce_pdf_preparation_user_metadata( normalized.get("pdf_preparation_user_metadata") ) normalized["recomendaciones_medicas"] = _normalize_recomendaciones_medicas( normalized.get("recomendaciones_medicas") ) _normalize_historia_clinica_base(normalized) _normalize_factura_base(normalized) factura_services = _servicios_factura(normalized.get("factura") or {}) normalized["medicamentos_prescripcion"] = normalizar_lista_medicamentos( normalized.get("medicamentos_prescripcion"), fuente="prescripcion", ) medication_catalog = reconcile_case_medications( [ *(normalized.get("medicamentos_hc") or []), *(normalized.get("medicamentos_prescripcion") or []), ], factura_services.get("medicamentos") or [], ) normalized["medicamentos_caso"] = evaluate_medication_pertinence( medication_catalog, invoice_state=_medication_invoice_state(normalized.get("factura")), support_documents=_medication_support_documents(normalized), ) normalized["medicamentos_caso"] = apply_clinical_pertinence_reviews( normalized["medicamentos_caso"], _previous_clinical_medication_reviews(previous_medication_catalog), ) normalized["medicamentos_glosa"] = build_medication_glosa_findings(normalized["medicamentos_caso"]) normalized["medicamentos_hc_display"] = build_medication_display_list(normalized["medicamentos_caso"]) normalized["radiologia"] = _normalize_dict_list(normalized.get("radiologia")) normalized["laboratorio"] = _normalize_dict_list(normalized.get("laboratorio")) normalized["generico"] = _normalize_dict_list(normalized.get("generico")) ayudas_existentes = coerce_ayudas_diagnosticas(normalized.get("ayudas_diagnosticas")) normalized["ayudas_diagnosticas"] = ayudas_existentes or build_ayudas_diagnosticas(normalized) excluded_ayudas = normalized["pdf_draft"].get("excluded_ayudas_diagnosticas_keys") for ayuda in normalized["ayudas_diagnosticas"]: ayuda["exportable"] = evaluar_ayuda_diagnostica_exportable(ayuda, excluded_ayudas) apply_factura_cost_ordering(normalized) normalized["medicamentos_hc_display"] = build_medication_display_list(normalized["medicamentos_caso"]) normalized["imagenes_diagnosticas"] = _normalize_text_list(normalized.get("imagenes_diagnosticas")) or [ format_ayuda_diagnostica_presentacion(item) for item in normalized["ayudas_diagnosticas"] if item.get("tipo") == "imagen" ] normalized["diagnosticos_consolidados"] = _coerce_diagnostico_catalog( normalized.get("diagnosticos_consolidados") ) normalized["diagnosticos_pendientes_validacion"] = _coerce_diagnostico_catalog( normalized.get("diagnosticos_pendientes_validacion") ) normalized["diagnosticos_consolidacion_hallazgos"] = _coerce_diagnostico_hallazgos( normalized.get("diagnosticos_consolidacion_hallazgos") ) if not normalized["diagnosticos_consolidados"]: normalized.update(build_diagnosticos_consolidados(normalized)) normalized["pdf_draft"] = reconcile_pdf_draft(normalized.get("pdf_draft"), normalized) normalized.update(build_pdf_export_blocks(normalized)) if curation := curation_from_context(normalized): normalized.update(curation_context_payload(curation)) apply_factura_cost_ordering(normalized) normalized["pdf_preparation_procedimientos"] = build_pdf_preparation_procedimientos(normalized) return normalized def is_compatible_cached_context(context: dict[str, Any] | None) -> bool: if not isinstance(context, dict): return False if context.get("orden_costo_version") != COST_ORDER_VERSION: return False required_keys = { "historia", "quirurgico", "factura", "prescripcion", "auto_soat_results", "qx_agent_results", "manual_soat_results", "pdf_draft", "procedimientos_hc", "medicamentos_hc", "medicamentos_hc_display", "medicamentos_caso", "medicamentos_prescripcion", "medicamentos_glosa", "procedimientos_factura", "imagenes_diagnosticas", "diagnosticos_consolidados", "diagnosticos_pendientes_validacion", "diagnosticos_consolidacion_hallazgos", "ayudas_diagnosticas", } if not required_keys.issubset(set(context.keys())): return False if any( context.get(key) is not None and not isinstance(context.get(key), dict) for key in ("historia", "quirurgico", "factura") ): return False medication_catalog = context.get("medicamentos_caso") if isinstance(medication_catalog, list) and any( not isinstance(item, dict) or not isinstance(item.get("pertinencia"), dict) or item["pertinencia"].get("version") != "v2" for item in medication_catalog ): return False return all( isinstance(context.get(key), list) for key in ( "procedimientos_hc", "medicamentos_hc", "medicamentos_hc_display", "medicamentos_caso", "medicamentos_prescripcion", "medicamentos_glosa", "prescripcion", "procedimientos_factura", "imagenes_diagnosticas", ) ) def _diagnostico_display(item: Any) -> str: if not isinstance(item, dict): return str(item or "").strip() codigo = str(item.get("codigo") or item.get("codigo_asignado") or "").strip().upper() descripcion = str(item.get("descripcion") or "").strip() diagnostico = str(item.get("diagnostico") or "").strip() detalle = descripcion or diagnostico if codigo and detalle: return f"{codigo} - {detalle}" return codigo or detalle def _procedimiento_display(item: Any) -> str: if not isinstance(item, dict): return str(item or "").strip() codigo = str(item.get("codigo") or item.get("codigo_cups") or "").strip() descripcion = str(item.get("descripcion") or item.get("procedimiento") or "").strip() if codigo and descripcion: return f"{codigo} - {descripcion}" return codigo or descripcion def _add_text_value(values: set[str], text: Any) -> None: normalized = str(text or "").strip() if normalized: values.add(normalized) def _extend_with_display_values( values: set[str], items: Any, formatter, ) -> None: for item in items or []: _add_text_value(values, formatter(item)) def _iter_soat_diagnosticos(items: Any) -> list[Any]: diagnosticos: list[Any] = [] for item in items or []: if not isinstance(item, dict): continue principal = item.get("cie10_principal") if isinstance(principal, dict): diagnosticos.append(principal) diagnosticos.extend(item.get("cie10_secundarios") or []) return diagnosticos def _iter_soat_procedimientos(items: Any) -> list[Any]: procedimientos: list[Any] = [] for item in items or []: if not isinstance(item, dict): continue principal = item.get("cups_principal") if isinstance(principal, dict): procedimientos.append(principal) procedimientos.extend(item.get("cups_alternativos") or []) return procedimientos def _iter_factura_procedimientos(items: Any) -> list[dict[str, str]]: procedimientos: list[dict[str, str]] = [] for item in items or []: if not isinstance(item, dict): continue procedimientos.append( { "codigo": item.get("codigo_cups") or item.get("codigo_referencia"), "descripcion": item.get("descripcion"), } ) return procedimientos def _servicios_factura(factura: Any) -> dict[str, Any]: if not isinstance(factura, dict): return {} factura_json = factura.get("factura_json") if not isinstance(factura_json, dict): return {} servicios = factura_json.get("servicios_procedimientos") return servicios if isinstance(servicios, dict) else {} def _primary_diagnostico_sources(item: dict[str, Any]) -> set[str]: return {str(source).strip() for source in (item.get("fuentes") or []) if str(source).strip()} def _has_meaningful_diagnostico_text(item: dict[str, Any]) -> bool: diagnostico = _normalize_whitespace(item.get("diagnostico")) descripcion = _normalize_whitespace(item.get("descripcion")) return any( clinical_text and not _is_technical_description(clinical_text) for clinical_text in (diagnostico, descripcion) ) def _is_pdf_primary_diagnostico(item: dict[str, Any]) -> bool: fuentes = _primary_diagnostico_sources(item) if fuentes: has_clinical_source = bool(fuentes & _PDF_PRIMARY_DIAGNOSTICO_SOURCES) else: tipo_contexto = str(item.get("tipo_contexto") or "").strip().lower() has_clinical_source = tipo_contexto not in {"no_clasificado", "procedimiento_soat"} if not has_clinical_source or not _has_meaningful_diagnostico_text(item): return False estado = str(item.get("estado_codificacion") or "").strip() if estado == "codificado": return _is_real_cie10_code(item.get("codigo")) return estado in {"pendiente_validacion", "provisional_grupo", "inferido_revisable"} def build_pdf_primary_diagnosticos(context: dict[str, Any]) -> list[dict[str, Any]]: return [ dict(item) for item in _coerce_diagnostico_catalog(context.get("diagnosticos_consolidados")) if _is_pdf_primary_diagnostico(item) ] def _procedure_key(description: Any, cups: Any = "") -> str: code = re.sub(r"\D", "", str(cups or "")) text = _normalize_ascii(description).casefold() text = re.sub(r"[^a-z0-9]+", " ", text).strip() return f"{code}|" if code else f"|{text}" def _classify_cups(code: Any) -> ProcedureClassification: # La clasificación automática solo se acepta cuando ya fue resuelta por el catálogo CUPS. # Un código aislado no contiene el número de capítulo en sus primeros dígitos. return ProcedureClassification.SIN_CLASIFICAR def build_procedimientos_clinicos(context: dict[str, Any]) -> list[dict[str, Any]]: """Consolida fuentes clínicas/administrativas sin perder trazabilidad interna.""" if context.get("curation_schema_version"): publishable = {"validado_automaticamente", "aprobado", "corregido"} resolved_states = {"realizado_y_facturado", "no_quirurgico_facturado"} curated = [] for raw in context.get("procedimientos_curados") or []: if not isinstance(raw, dict) or raw.get("status") not in publishable: continue classification = str(raw.get("classification") or "sin_clasificar") if classification not in {"quirurgico", "no_quirurgico"}: continue if raw.get("reconciliation_status") not in resolved_states and raw.get("status") not in { "aprobado", "corregido", }: continue curated.append( { "key": raw.get("item_id") or _procedure_key(raw.get("description"), raw.get("codigo_cups")), "codigo_cups": str(raw.get("codigo_cups") or ""), "descripcion": str(raw.get("description") or ""), "clasificacion": classification, "clasificacion_automatica": classification, "clasificacion_manual": False, "soporte_clinico": "realizado" in (raw.get("states") or []), "fuentes": [ evidence.get("document_type") for evidence in raw.get("evidence") or [] if isinstance(evidence, dict) and evidence.get("document_type") ], "estado_documental": list(raw.get("states") or []), "evidencias": [ evidence for evidence in raw.get("evidence") or [] if isinstance(evidence, dict) ], "advertencias": list(raw.get("warnings") or []), "estado_conciliacion": raw.get("reconciliation_status"), "razon_conciliacion": str(raw.get("reconciliation_reason") or ""), } ) ordered = sort_publishable_procedures( curated, lines=extract_invoice_cost_lines(context.get("factura")), ) return [ { key: value for key, value in item.items() if key not in {"codigo_soat", "grupo_quirurgico_soat", "costo_facturado"} } for item in ordered if isinstance(item, dict) ] merged: dict[str, ProcedimientoClinico] = {} draft = coerce_pdf_draft(context.get("pdf_draft")) overrides = draft.get("procedure_classification_overrides") or {} def add( raw: Any, source: str, *, clinical: bool, surgical_evidence: bool = False, official_classification: str = "", ) -> None: if isinstance(raw, dict): description = str( raw.get("descripcion") or raw.get("procedimiento") or raw.get("concepto") or "" ).strip() cups = str(raw.get("codigo_cups") or raw.get("codigo") or "").strip() soat = str(raw.get("codigo_soat") or "").strip() else: description = str(raw or "").strip() match = re.match(r"^\s*(\d{6})\s*(?:[-:]|\s)\s*(.+)$", description) cups = match.group(1) if match else "" description = match.group(2).strip() if match else description soat = "" if not description: return base_key = _procedure_key(description, cups) key = base_key existing = merged.get(key) classification = ( ProcedureClassification(official_classification) if official_classification in {"quirurgico", "no_quirurgico"} else _classify_cups(cups) ) if surgical_evidence and classification == ProcedureClassification.SIN_CLASIFICAR: classification = ProcedureClassification.QUIRURGICO if existing is None: existing = ProcedimientoClinico( descripcion=description, codigo_cups=cups, codigo_soat=soat, clasificacion=classification, fuentes=[source], evidencia=description if clinical else "", soporte_clinico=clinical, ) else: if source == "factura": existing.descripcion = description existing.codigo_cups = existing.codigo_cups or cups existing.codigo_soat = existing.codigo_soat or soat existing.fuentes = list(dict.fromkeys([*existing.fuentes, source])) existing.soporte_clinico = existing.soporte_clinico or clinical if existing.clasificacion == ProcedureClassification.SIN_CLASIFICAR: existing.clasificacion = classification override = overrides.get(key) or overrides.get(base_key) if override: existing.clasificacion = ProcedureClassification(override) existing.clasificacion_manual = True merged[key] = existing for item in context.get("procedimientos_hc") or []: add(item, "historia_clinica", clinical=True) quirurgico = context.get("quirurgico") or {} for item in quirurgico.get("codigos_cups") or []: add(item, "quirurgico", clinical=True, surgical_evidence=True) invoice_services = _servicios_factura(context.get("factura") or {}) if invoice_services: for item in invoice_services.get("procedimientos_quirurgicos") or []: add(item, "factura", clinical=False, official_classification="quirurgico") for item in invoice_services.get("procedimientos_no_quirurgicos") or []: add(item, "factura", clinical=False, official_classification="no_quirurgico") else: for item in context.get("procedimientos_factura") or []: add(item, "factura", clinical=False) for result in context.get("codigos_desde_soat") or []: if not isinstance(result, dict): continue source = "qx_agent" if result in (context.get("qx_agent_results") or []) else "soat" surgical = source == "qx_agent" principal = result.get("cups_principal") if isinstance(principal, dict): add( {**principal, "codigo_soat": result.get("codigo_soat")}, source, clinical=surgical, surgical_evidence=surgical, ) for alternative in result.get("cups_alternativos") or []: add(alternative, source, clinical=surgical, surgical_evidence=surgical) procedures = [] for key, item in merged.items(): automatic_classification = item.clasificacion if automatic_classification == ProcedureClassification.SIN_CLASIFICAR and { "quirurgico", "qx_agent", } & set(item.fuentes): automatic_classification = ProcedureClassification.QUIRURGICO procedures.append( item.model_dump(mode="json") | { "key": key, "clasificacion_automatica": automatic_classification.value, } ) ordered = attach_and_sort_costs( procedures, category="procedimiento", lines=extract_invoice_cost_lines(context.get("factura")), ) return [ { key: value for key, value in item.items() if key not in {"codigo_soat", "grupo_quirurgico_soat", "costo_facturado"} } if isinstance(item, dict) else item for item in ordered ] def build_pdf_primary_procedimientos(context: dict[str, Any]) -> list[str]: """Contrato legado: la vista nueva consume ``procedimientos_clinicos``.""" procedimientos: list[str] = [] for item in build_procedimientos_clinicos(context): if not isinstance(item, dict): continue if not context.get("curation_schema_version") and not ( {"historia_clinica", "quirurgico"} & set(item.get("fuentes") or []) ): continue description = str(item.get("descripcion") or "").strip() code = str(item.get("codigo_cups") or "").strip() display = " - ".join(part for part in (code, description) if part) if display: procedimientos.append(display) return procedimientos def _parse_procedure_display(value: Any) -> tuple[str, str]: text = str(value or "").strip() match = re.match(r"^\s*(\d{6})\s*(?:[-:]|\s)\s*(.+)$", text) if match: return match.group(1), match.group(2).strip() return "", text def build_pdf_selectable_procedimientos(context: dict[str, Any]) -> list[dict[str, Any]]: """Enriquece la selección legado sin cambiar sus fuentes ni su orden visible.""" clinical = build_procedimientos_clinicos(context) by_display = { " - ".join( part for part in ( str(item.get("codigo_cups") or "").strip(), str(item.get("descripcion") or "").strip(), ) if part ): item for item in clinical if isinstance(item, dict) } selectable: list[dict[str, Any]] = [] for display in build_pdf_primary_procedimientos(context): item = by_display.get(display) if item is not None: selectable.append(dict(item)) continue code, description = _parse_procedure_display(display) automatic = _classify_cups(code).value selectable.append( { "key": _procedure_key(description, code), "codigo_cups": code, "descripcion": description, "clasificacion": automatic, "clasificacion_automatica": automatic, "clasificacion_manual": False, "soporte_clinico": True, "fuentes": [], } ) draft = coerce_pdf_draft(context.get("pdf_draft")) overrides = draft.get("procedure_classification_overrides") or {} for entry in draft.get("manual_procedure_entries") or []: key = str(entry.get("key") or "").strip() text = str(entry.get("value") or "").strip() if not key or not text: continue code, description = _parse_procedure_display(text) classification = str(overrides.get(key) or ProcedureClassification.SIN_CLASIFICAR.value) selectable.append( { "key": key, "codigo_cups": code, "descripcion": description, "texto_manual": text, "clasificacion": classification, "clasificacion_automatica": ProcedureClassification.SIN_CLASIFICAR.value, "clasificacion_manual": classification in { ProcedureClassification.QUIRURGICO.value, ProcedureClassification.NO_QUIRURGICO.value, }, "soporte_clinico": True, "fuentes": ["manual_pdf"], "manual": True, } ) return selectable def build_pdf_preparation_procedimientos(context: dict[str, Any]) -> list[dict[str, Any]]: """Proyección exclusiva de UI; costos y fórmulas nunca cruzan este borde.""" selectable = [ item for item in build_pdf_selectable_procedimientos(context) if isinstance(item, dict) and item.get("clasificacion") in {"quirurgico", "no_quirurgico"} ] enriched = attach_and_sort_costs( selectable, category="procedimiento", lines=extract_invoice_cost_lines(context.get("factura")), ) metadata = coerce_pdf_preparation_user_metadata(context.get("pdf_preparation_user_metadata")) manual_groups = { str(item.get("procedure_key") or ""): int(item["group"]) for item in metadata["procedure_soat_groups"] } decorated: list[tuple[dict[str, Any], int | None, Any, int, str]] = [] for canonical_order, raw in enumerate(enriched): if not isinstance(raw, dict): continue item = dict(raw) key = str(item.get("key") or "").strip() classification = str(item.get("clasificacion") or "") raw_official_group = item.get("grupo_quirurgico_soat") official_group: int | None = None if raw_official_group is not None: try: parsed_official_group = int(raw_official_group) except (TypeError, ValueError): pass else: official_group = parsed_official_group if 2 <= parsed_official_group <= 23 else None manual_group = manual_groups.get(key) if classification == "quirurgico" else None effective_group = official_group if official_group is not None else manual_group origin = "oficial" if official_group is not None else "manual_visual" if manual_group else "" cost = item.get("costo_facturado") cost_value = normalize_money(cost.get("valor_total")) if isinstance(cost, dict) else None decorated.append((item, effective_group, cost_value, canonical_order, origin)) decorated.sort( key=lambda entry: ( entry[1] is None, -(entry[1] or 0), entry[2] is None, -(entry[2] or 0), entry[3], ) ) result = [] for visual_order, (item, group, _cost, canonical_order, origin) in enumerate(decorated): projection = { key: value for key, value in item.items() if key not in {"codigo_soat", "grupo_quirurgico_soat", "costo_facturado", "valoracion_soat"} } projection.update( { "grupo_soat_preparacion": group, "grupo_soat_origen": origin, "orden_visual": visual_order, "orden_canonico": canonical_order, } ) if origin == "manual_visual": projection["grupo_soat_advertencia"] = ( "Asignación manual para organización visual; sin efecto tarifario" ) result.append(projection) return result def build_selected_pdf_procedimientos(context: dict[str, Any]) -> list[dict[str, Any]]: """Aplica la selección persistida y descarta categorías no publicables en servidor.""" draft = coerce_pdf_draft(context.get("pdf_draft")) excluded = { str(value).strip() for value in draft.get("excluded_procedimientos") or [] if str(value).strip() } selected = [] for item in build_pdf_selectable_procedimientos(context): classification = str(item.get("clasificacion") or "") if classification not in {"quirurgico", "no_quirurgico"}: continue display = " - ".join( value for value in ( str(item.get("codigo_cups") or "").strip(), str(item.get("descripcion") or "").strip(), ) if value ) if display in excluded or str(item.get("descripcion") or "").strip() in excluded: continue selected.append( { "codigo_cups": str(item.get("codigo_cups") or ""), "descripcion": str(item.get("descripcion") or ""), "clasificacion": classification, } ) return selected def extract_incapacidad_clinica(context: dict[str, Any]) -> dict[str, Any]: historia = context.get("historia") or {} structured = historia.get("analysis_structured") or historia.get("historia_json") or {} candidate = structured.get("incapacidad") if isinstance(structured, dict) else None if isinstance(candidate, dict): ordered = candidate.get("ordenada") is True or str(candidate.get("ordenada") or "").casefold() in { "si", "sí", "true", } return IncapacidadClinica( ordenada=ordered, fecha_inicio=str(candidate.get("fecha_inicio") or ""), fecha_fin=str(candidate.get("fecha_fin") or ""), duracion=str(candidate.get("duracion") or candidate.get("dias") or ""), unidad=str(candidate.get("unidad") or ("días" if candidate.get("dias") else "")), motivo=str(candidate.get("motivo") or ""), recomendacion=str(candidate.get("recomendacion") or ""), evidencia=str(candidate.get("evidencia") or ""), ).model_dump(mode="json") raw = _normalize_whitespace(historia.get("texto_extraido") or historia.get("texto") or "") match = re.search( r"(?i)\b(?:se\s+)?(?:ordena|prescribe|otorga)\s+incapacidad(?:\s+(?:m[eé]dica\s+)?)?" r"(?:por\s+)?(?P\d+)\s*(?Pd[ií]as?|horas?)\b[^.\n]{0,160}", raw, ) if not match: return IncapacidadClinica().model_dump(mode="json") return IncapacidadClinica( ordenada=True, duracion=match.group("duration"), unidad=match.group("unit"), evidencia=match.group(0), ).model_dump(mode="json") def _is_meaningful_pdf_medicamento(item: dict[str, Any]) -> bool: nombre = str(item.get("medicamento") or "").strip() dosis = str(item.get("dosis") or "").strip() cantidad = str(item.get("cantidad") or "").strip() texto_original = str(item.get("texto_original") or "").strip() if nombre.lower() == "no especificado" and not dosis and not cantidad: return False if nombre and nombre.lower() != "no especificado": return True return bool(dosis or cantidad or (texto_original and texto_original.lower() != "no especificado")) def build_pdf_primary_medicamentos(context: dict[str, Any]) -> list[dict[str, Any]]: catalog = context.get("medicamentos_caso") if isinstance(catalog, list): return [ dict(item) for item in catalog if isinstance(item, dict) and "administrado" in {str(status or "").strip().casefold() for status in (item.get("estados") or [])} and _is_meaningful_pdf_medicamento(item) ] factura = context.get("factura") or {} servicios_factura = _servicios_factura(factura) medications = reconcile_case_medications( normalizar_lista_medicamentos(context.get("medicamentos_hc"), fuente="historia_clinica"), servicios_factura.get("medicamentos") or [], ) return [ item for item in medications if "administrado" in {str(status or "").strip().casefold() for status in (item.get("estados") or [])} and _is_meaningful_pdf_medicamento(item) ] def _is_primary_pdf_ayuda(ayuda: dict[str, Any]) -> bool: if ayuda.get("exportable") is False or ayuda.get("glosado"): return False alertas = {str(alerta).strip() for alerta in (ayuda.get("alertas") or []) if str(alerta).strip()} if alertas & _PDF_ANALYTICAL_AID_ALERTS: return False fuente = str(ayuda.get("fuente") or "").strip() if fuente in _PDF_PRIMARY_AYUDA_SOURCES: return True return bool(ayuda.get("interpretado") or ayuda.get("facturado")) def build_pdf_primary_ayudas_diagnosticas(context: dict[str, Any]) -> list[dict[str, Any]]: ayudas = coerce_ayudas_diagnosticas(context.get("ayudas_diagnosticas")) or build_ayudas_diagnosticas( context ) return [dict(ayuda) for ayuda in ayudas if _is_primary_pdf_ayuda(ayuda)] def build_pdf_export_blocks(context: dict[str, Any]) -> dict[str, Any]: return { "pdf_primary_diagnosticos": build_pdf_primary_diagnosticos(context), "pdf_primary_procedimientos": build_pdf_primary_procedimientos(context), "pdf_selectable_procedimientos": build_pdf_selectable_procedimientos(context), "pdf_primary_medicamentos": build_pdf_primary_medicamentos(context), "pdf_primary_ayudas_diagnosticas": build_pdf_primary_ayudas_diagnosticas(context), "procedimientos_clinicos": build_procedimientos_clinicos(context), "incapacidad_clinica": extract_incapacidad_clinica(context), } def _build_diagnosticos_base(context: dict[str, Any]) -> tuple[set[str], set[str]]: diagnosticos_catalog = _coerce_diagnostico_catalog(context.get("diagnosticos_consolidados")) if diagnosticos_catalog: return ( { item.get("texto_presentacion", "") for item in diagnosticos_catalog if item.get("texto_presentacion") }, {item.get("key", "") for item in diagnosticos_catalog if item.get("key")}, ) historia = context.get("historia") or {} quirurgico = context.get("quirurgico") or {} codigos_desde_soat = context.get("codigos_desde_soat") diagnosticos: set[str] = set() _extend_with_display_values( diagnosticos, (historia.get("codigos_cie10") or []) + (quirurgico.get("codigos_cie10") or []), _diagnostico_display, ) _extend_with_display_values( diagnosticos, _iter_soat_diagnosticos(codigos_desde_soat), _diagnostico_display, ) return diagnosticos, set() def _build_procedimientos_base( quirurgico: dict[str, Any], procedimientos_hc: Any, procedimientos_factura: Any, codigos_desde_soat: Any, ) -> set[str]: procedimientos: set[str] = set() _extend_with_display_values(procedimientos, procedimientos_hc, lambda item: str(item or "").strip()) _extend_with_display_values(procedimientos, quirurgico.get("codigos_cups"), _procedimiento_display) _extend_with_display_values( procedimientos, _iter_factura_procedimientos(procedimientos_factura), _procedimiento_display, ) _extend_with_display_values( procedimientos, _iter_soat_procedimientos(codigos_desde_soat), _procedimiento_display, ) return procedimientos def _build_medicamentos_base(context: dict[str, Any], factura: dict[str, Any]) -> tuple[set[str], set[str]]: medicamentos: set[str] = set() keys: set[str] = set() catalog = context.get("medicamentos_caso") if isinstance(catalog, list): for item in catalog: if not isinstance(item, dict): continue display = formatear_medicamento_canonico(item) if display: medicamentos.add(display) if item.get("key"): keys.add(str(item["key"])) return medicamentos, keys servicios_factura = _servicios_factura(factura) medicamentos_hc = normalizar_lista_medicamentos( context.get("medicamentos_hc"), fuente="historia_clinica", ) medicamentos_factura = normalizar_lista_medicamentos( servicios_factura.get("medicamentos"), fuente="factura", ) _extend_with_display_values(medicamentos, medicamentos_hc, formatear_medicamento_canonico) _extend_with_display_values(medicamentos, medicamentos_factura, formatear_medicamento_canonico) return medicamentos, keys def _build_imagenes_diagnosticas_base(context: dict[str, Any], factura: dict[str, Any]) -> set[str]: imagenes = set(_clean_unique_text_list(context.get("imagenes_diagnosticas"))) if imagenes: return imagenes return set(extraer_imagenes_diagnosticas(factura, context.get("radiologia"))) def _build_ayudas_diagnosticas_base(context: dict[str, Any]) -> tuple[set[str], set[str]]: ayudas = coerce_ayudas_diagnosticas(context.get("ayudas_diagnosticas")) or build_ayudas_diagnosticas( context ) return ( {format_ayuda_diagnostica_presentacion(item) for item in ayudas if item.get("nombre")}, {item.get("key", "") for item in ayudas if item.get("key")}, ) def build_selection_base(context: dict[str, Any]) -> dict[str, set[str]]: quirurgico = context.get("quirurgico") or {} factura = context.get("factura") or {} codigos_desde_soat = context.get("codigos_desde_soat") diagnosticos_textos, diagnosticos_keys = _build_diagnosticos_base(context) ayudas_textos, ayudas_keys = _build_ayudas_diagnosticas_base(context) medicamentos_textos, medicamento_keys = _build_medicamentos_base(context, factura) return { "diagnosticos": diagnosticos_textos, "diagnostico_keys": diagnosticos_keys, "procedimientos": _build_procedimientos_base( quirurgico, context.get("procedimientos_hc"), context.get("procedimientos_factura"), codigos_desde_soat, ), "medicamentos": medicamentos_textos, "medicamento_keys": medicamento_keys, "ayudas_diagnosticas": ayudas_textos, "ayudas_diagnosticas_keys": ayudas_keys, "imagenes_diagnosticas": _build_imagenes_diagnosticas_base(context, factura), } def reconcile_pdf_draft(pdf_draft: Any, context: dict[str, Any]) -> dict[str, Any]: draft = coerce_pdf_draft(pdf_draft) base = build_selection_base(context) diagnosticos_catalog = { item.get("key", ""): item.get("texto_presentacion", "") for item in _coerce_diagnostico_catalog(context.get("diagnosticos_consolidados")) } legacy_keys = [ key for key, text in diagnosticos_catalog.items() if text in draft["excluded_diagnosticos"] ] draft["excluded_diagnostico_keys"] = [ item for item in _clean_unique_text_list(draft["excluded_diagnostico_keys"] + legacy_keys) if item in base["diagnostico_keys"] ] draft["excluded_diagnosticos"] = [ diagnosticos_catalog[key] for key in draft["excluded_diagnostico_keys"] if key in diagnosticos_catalog ] draft["excluded_diagnosticos"] = [ item for item in draft["excluded_diagnosticos"] if item in base["diagnosticos"] ] selectable_procedure_texts = set(base["procedimientos"]) | set(draft["manual_procedimientos"]) draft["excluded_procedimientos"] = [ item for item in draft["excluded_procedimientos"] if item in selectable_procedure_texts ] draft["excluded_medicamentos"] = [ item for item in draft["excluded_medicamentos"] if item in base["medicamentos"] ] medication_catalog = { str(item.get("key") or ""): formatear_medicamento_canonico(item) for item in (context.get("medicamentos_caso") or []) if isinstance(item, dict) and item.get("key") } legacy_medication_keys = [ key for key, text in medication_catalog.items() if text in draft["excluded_medicamentos"] ] draft["excluded_medicamento_keys"] = [ item for item in _clean_unique_text_list(draft["excluded_medicamento_keys"] + legacy_medication_keys) if item in base["medicamento_keys"] ] draft["excluded_medicamentos"] = [ medication_catalog[key] for key in draft["excluded_medicamento_keys"] if key in medication_catalog ] legacy_ayuda_keys = [ item.get("key", "") for item in coerce_ayudas_diagnosticas(context.get("ayudas_diagnosticas")) if format_ayuda_diagnostica_presentacion(item) in draft["excluded_imagenes_diagnosticas"] ] draft["excluded_ayudas_diagnosticas_keys"] = [ item for item in _clean_unique_text_list(draft["excluded_ayudas_diagnosticas_keys"] + legacy_ayuda_keys) if item in base["ayudas_diagnosticas_keys"] ] draft["excluded_imagenes_diagnosticas"] = [ item for item in draft["excluded_imagenes_diagnosticas"] if item in base["imagenes_diagnosticas"] ] draft["manual_diagnosticos"] = _clean_unique_text_list(draft["manual_diagnosticos"]) draft["manual_diagnostico_entries"] = _normalize_manual_text_entries( draft["manual_diagnostico_entries"], draft["manual_diagnosticos"], prefix="diagnostico" ) draft["manual_diagnosticos"] = [item["value"] for item in draft["manual_diagnostico_entries"]] draft["manual_procedimientos"] = _clean_unique_text_list(draft["manual_procedimientos"]) draft["manual_procedure_entries"] = _normalize_manual_procedure_entries( draft["manual_procedure_entries"], draft["manual_procedimientos"] ) draft["manual_procedimientos"] = [item["value"] for item in draft["manual_procedure_entries"]] draft["manual_medicamentos"] = _clean_unique_text_list(draft["manual_medicamentos"]) draft["manual_medicamento_entries"] = _normalize_manual_text_entries( draft["manual_medicamento_entries"], draft["manual_medicamentos"], prefix="medicamento" ) draft["manual_medicamentos"] = [item["value"] for item in draft["manual_medicamento_entries"]] draft["manual_imagenes_diagnosticas"] = _clean_unique_text_list(draft["manual_imagenes_diagnosticas"]) draft["manual_ayudas_diagnosticas"] = coerce_ayudas_diagnosticas(draft["manual_ayudas_diagnosticas"]) return draft