from __future__ import annotations import hashlib import json import re from typing import Any, Final from app.rda.domain.models import RdaArtifactType, RdaFieldStatus, RdaSectionTrace RDA_SCHEMA_VERSION: Final[int] = 1 RDA_MAPPER_VERSION: Final[str] = "2026-05-rda-v1" PATIENT_ID_METADATA_SOURCE: Final[str] = "metadatos_hc.datos_identificacion_paciente" PATIENT_ID_INVOICE_SOURCE: Final[str] = "factura.factura_json.informacion_paciente.numero_identificacion" _DOCUMENT_RE: Final[re.Pattern[str]] = re.compile( r"^(?P[A-Z]{2,4})\s+(?P.+)$", flags=re.IGNORECASE ) def build_source_context_fingerprint( context: dict[str, Any], *, artifact_type: str | RdaArtifactType, schema_version: int = RDA_SCHEMA_VERSION, mapper_version: str = RDA_MAPPER_VERSION, ) -> str: normalized_artifact_type = ( artifact_type if isinstance(artifact_type, RdaArtifactType) else RdaArtifactType(str(artifact_type).strip().lower()) ) canonical = { "artifact_type": normalized_artifact_type.value, "schema_version": schema_version, "mapper_version": mapper_version, "context": context, } serialized = json.dumps(canonical, ensure_ascii=False, sort_keys=True, default=str) return hashlib.sha256(serialized.encode("utf-8")).hexdigest() def build_rda_patient_summary( context: dict[str, Any], ) -> tuple[dict[str, Any], list[RdaSectionTrace], list[str]]: factura_json = _resolve_factura_json(context) metadatos = _resolve_dict(context.get("metadatos_hc")) patient_doc = _resolve_patient_document(context, metadatos, factura_json) traces = [ _trace_for_candidates( "patient.name", [ ("context.nombre_paciente", context.get("nombre_paciente"), RdaFieldStatus.DOCUMENTAL), ( "historia.nombre_paciente", _resolve_dict(context.get("historia")).get("nombre_paciente"), RdaFieldStatus.DOCUMENTAL, ), ( "factura.nombre_paciente", _resolve_dict(context.get("factura")).get("nombre_paciente"), RdaFieldStatus.DOCUMENTAL, ), ], ), _trace_for_candidates( "patient.document.type", [ (PATIENT_ID_METADATA_SOURCE, patient_doc.get("type"), RdaFieldStatus.INFERIDO), (PATIENT_ID_INVOICE_SOURCE, patient_doc.get("invoice_type"), RdaFieldStatus.INFERIDO), ], ), _trace_for_candidates( "patient.document.number", [ (PATIENT_ID_METADATA_SOURCE, patient_doc.get("number"), RdaFieldStatus.INFERIDO), (PATIENT_ID_INVOICE_SOURCE, patient_doc.get("invoice_number"), RdaFieldStatus.INFERIDO), ], ), _trace_for_candidates( "patient.birth_date", [("metadatos_hc.fecha_nacimiento", metadatos.get("fecha_nacimiento"), RdaFieldStatus.DOCUMENTAL)], ), _trace_for_candidates( "patient.sex", [("metadatos_hc.sexo", metadatos.get("sexo"), RdaFieldStatus.DOCUMENTAL)], ), _trace_for_candidates( "patient.provider.name", [ ( "factura.factura_json.proveedor.nombre", _resolve_dict(factura_json.get("proveedor")).get("nombre"), RdaFieldStatus.DOCUMENTAL, ), ( "metadatos_hc.prestador_servicio", metadatos.get("prestador_servicio"), RdaFieldStatus.DOCUMENTAL, ), ], ), _trace_for_candidates( "patient.provider.nit", [ ( "factura.factura_json.proveedor.nit", _resolve_dict(factura_json.get("proveedor")).get("nit"), RdaFieldStatus.DOCUMENTAL, ) ], ), _trace_for_candidates( "patient.payer.name", [ ( "factura.factura_json.pagador.aseguradora_eps", _resolve_dict(factura_json.get("pagador")).get("aseguradora_eps"), RdaFieldStatus.DOCUMENTAL, ) ], ), _trace_for_candidates( "patient.payer.nit", [ ( "factura.factura_json.pagador.nit_pagador", _resolve_dict(factura_json.get("pagador")).get("nit_pagador"), RdaFieldStatus.DOCUMENTAL, ) ], ), _trace_for_candidates( "patient.clinical_background", [ ("metadatos_hc.antecedentes", metadatos.get("antecedentes"), RdaFieldStatus.DOCUMENTAL), ("context.antecedentes", context.get("antecedentes"), RdaFieldStatus.DOCUMENTAL), ], ), _trace_for_candidates( "patient.diagnoses", [("diagnosticos_consolidados", _build_diagnosis_list(context), RdaFieldStatus.INFERIDO)], ), ] payload = { "summary_type": RdaArtifactType.PATIENT.value, "patient": { "name": traces[0].value, "document": {"type": traces[1].value, "number": traces[2].value}, "birth_date": traces[3].value, "sex": traces[4].value, }, "organization": { "provider": {"name": traces[5].value, "nit": traces[6].value}, "payer": {"name": traces[7].value, "nit": traces[8].value}, }, "clinical_background": traces[9].value, "diagnoses": traces[10].value or [], } return payload, traces, _missing_fields(traces) def build_rda_emergency_summary( context: dict[str, Any], ) -> tuple[dict[str, Any], list[RdaSectionTrace], list[str]]: factura_json = _resolve_factura_json(context) metadatos = _resolve_dict(context.get("metadatos_hc")) patient_doc = _resolve_patient_document(context, metadatos, factura_json) procedimientos = _build_procedure_list(context, factura_json) medications = _build_medication_list(context, factura_json) diagnostic_support = _build_diagnostic_support_list(context) traces = [ _trace_for_candidates( "encounter.case_key", [("context.case_key", context.get("case_key"), RdaFieldStatus.DOCUMENTAL)], ), _trace_for_candidates( "encounter.case_number", [("context.case_number", context.get("case_number"), RdaFieldStatus.DOCUMENTAL)], ), _trace_for_candidates( "encounter.admission_at", [ ( "factura.factura_json.informacion_paciente.fecha_ingreso", _resolve_dict(factura_json.get("informacion_paciente")).get("fecha_ingreso"), RdaFieldStatus.DOCUMENTAL, ), ("metadatos_hc.fecha_ingreso", metadatos.get("fecha_ingreso"), RdaFieldStatus.DOCUMENTAL), ], ), _trace_for_candidates( "encounter.reason", [ ("metadatos_hc.motivo_consulta", metadatos.get("motivo_consulta"), RdaFieldStatus.DOCUMENTAL), ( "metadatos_hc.causa_motivo_atencion", metadatos.get("causa_motivo_atencion"), RdaFieldStatus.DOCUMENTAL, ), ], ), _trace_for_candidates( "encounter.triage", [("metadatos_hc.triage", metadatos.get("triage"), RdaFieldStatus.DOCUMENTAL)], ), _trace_for_candidates( "encounter.responsible_professional", [ ( "metadatos_hc.profesional_tratante", metadatos.get("profesional_tratante"), RdaFieldStatus.DOCUMENTAL, ), ( "metadatos_hc.medico_responsable", metadatos.get("medico_responsable"), RdaFieldStatus.DOCUMENTAL, ), ], ), _trace_for_candidates( "patient.document.type", [ (PATIENT_ID_METADATA_SOURCE, patient_doc.get("type"), RdaFieldStatus.INFERIDO), (PATIENT_ID_INVOICE_SOURCE, patient_doc.get("invoice_type"), RdaFieldStatus.INFERIDO), ], ), _trace_for_candidates( "patient.document.number", [ (PATIENT_ID_METADATA_SOURCE, patient_doc.get("number"), RdaFieldStatus.INFERIDO), (PATIENT_ID_INVOICE_SOURCE, patient_doc.get("invoice_number"), RdaFieldStatus.INFERIDO), ], ), _trace_for_candidates( "patient.name", [ ("context.nombre_paciente", context.get("nombre_paciente"), RdaFieldStatus.DOCUMENTAL), ( "historia.nombre_paciente", _resolve_dict(context.get("historia")).get("nombre_paciente"), RdaFieldStatus.DOCUMENTAL, ), ], ), _trace_for_candidates( "clinical.diagnoses", [("diagnosticos_consolidados", _build_diagnosis_list(context), RdaFieldStatus.INFERIDO)], ), _trace_for_candidates( "clinical.procedures", [("procedimientos_factura|procedimientos_hc", procedimientos, RdaFieldStatus.INFERIDO)], ), _trace_for_candidates( "clinical.medications", [ ( "medicamentos_hc_display|factura.servicios_procedimientos.medicamentos", medications, RdaFieldStatus.INFERIDO, ) ], ), _trace_for_candidates( "clinical.diagnostic_support", [("ayudas_diagnosticas|imagenes_diagnosticas", diagnostic_support, RdaFieldStatus.INFERIDO)], ), _trace_for_candidates( "encounter.provider.name", [ ( "factura.factura_json.proveedor.nombre", _resolve_dict(factura_json.get("proveedor")).get("nombre"), RdaFieldStatus.DOCUMENTAL, ), ( "metadatos_hc.prestador_servicio", metadatos.get("prestador_servicio"), RdaFieldStatus.DOCUMENTAL, ), ], ), _trace_for_candidates( "encounter.provider.nit", [ ( "factura.factura_json.proveedor.nit", _resolve_dict(factura_json.get("proveedor")).get("nit"), RdaFieldStatus.DOCUMENTAL, ) ], ), _trace_for_candidates( "encounter.payer.name", [ ( "factura.factura_json.pagador.aseguradora_eps", _resolve_dict(factura_json.get("pagador")).get("aseguradora_eps"), RdaFieldStatus.DOCUMENTAL, ) ], ), _trace_for_candidates( "encounter.payer.nit", [ ( "factura.factura_json.pagador.nit_pagador", _resolve_dict(factura_json.get("pagador")).get("nit_pagador"), RdaFieldStatus.DOCUMENTAL, ) ], ), ] payload = { "summary_type": RdaArtifactType.EMERGENCY.value, "patient": { "name": traces[8].value, "document": {"type": traces[6].value, "number": traces[7].value}, }, "encounter": { "case_key": traces[0].value, "case_number": traces[1].value, "admission_at": traces[2].value, "reason": traces[3].value, "triage": traces[4].value, "responsible_professional": traces[5].value, "provider": {"name": traces[13].value, "nit": traces[14].value}, "payer": {"name": traces[15].value, "nit": traces[16].value}, }, "clinical": { "diagnoses": traces[9].value or [], "procedures": traces[10].value or [], "medications": traces[11].value or [], "diagnostic_support": traces[12].value or [], }, } return payload, traces, _missing_fields(traces) def _resolve_factura_json(context: dict[str, Any]) -> dict[str, Any]: factura = _resolve_dict(context.get("factura")) return _resolve_dict(factura.get("factura_json")) def _resolve_dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def _clean_text(value: Any) -> str: if value is None: return "" if isinstance(value, list): return ", ".join(item for item in [_clean_text(item) for item in value] if item) return str(value).strip() def _has_value(value: Any) -> bool: if value is None: return False if isinstance(value, str): return bool(value.strip()) if isinstance(value, (list, dict, tuple, set)): return bool(value) return True def _normalize_compare(value: Any) -> str: if isinstance(value, list): return "|".join(_normalize_compare(item) for item in value if _has_value(item)) return _clean_text(value).casefold() def _parse_document(value: Any) -> tuple[str, str]: normalized = _clean_text(value) if not normalized: return ("", "") match = _DOCUMENT_RE.match(normalized) if not match: return ("", normalized) return (match.group("tipo").upper(), match.group("numero").strip()) def _resolve_patient_document( context: dict[str, Any], metadatos: dict[str, Any], factura_json: dict[str, Any], ) -> dict[str, str]: metadata_type, metadata_number = _parse_document(metadatos.get("datos_identificacion_paciente")) invoice_type, invoice_number = _parse_document( _resolve_dict(factura_json.get("informacion_paciente")).get("numero_identificacion") ) history_type, history_number = _parse_document(_resolve_dict(context.get("historia")).get("patient_id")) return { "type": metadata_type or history_type or invoice_type, "number": metadata_number or history_number or invoice_number, "invoice_type": invoice_type, "invoice_number": invoice_number, } def _trace_for_candidates( field_path: str, candidates: list[tuple[str, Any, RdaFieldStatus]], ) -> RdaSectionTrace: non_empty = [(source, value, status) for source, value, status in candidates if _has_value(value)] if not non_empty: return RdaSectionTrace( field_path=field_path, value=None, status=RdaFieldStatus.FALTANTE, primary_source="missing", ) primary_source, primary_value, primary_status = non_empty[0] alternate_sources: list[str] = [] distinct_values: list[str] = [] for source, value, _status in non_empty[1:]: alternate_sources.append(source) normalized = _normalize_compare(value) if normalized and normalized != _normalize_compare(primary_value): distinct_values.append(source) conflict_note = None if distinct_values: conflict_note = f"Se priorizó {primary_source} sobre {', '.join(distinct_values)}." return RdaSectionTrace( field_path=field_path, value=primary_value, status=primary_status, primary_source=primary_source, alternate_sources=alternate_sources, conflict_note=conflict_note, ) def _missing_fields(traces: list[RdaSectionTrace]) -> list[str]: return [trace.field_path for trace in traces if trace.status == RdaFieldStatus.FALTANTE] def _build_diagnosis_list(context: dict[str, Any]) -> list[dict[str, str]]: diagnoses = [] for item in context.get("diagnosticos_consolidados") or []: if not isinstance(item, dict): continue code = _clean_text(item.get("codigo")) description = _clean_text(item.get("descripcion") or item.get("diagnostico")) if not code and not description: continue diagnoses.append({"code": code, "description": description}) return diagnoses def _build_procedure_list(context: dict[str, Any], factura_json: dict[str, Any]) -> list[dict[str, str]]: procedures: list[dict[str, str]] = [] for item in context.get("procedimientos_factura") or []: if not isinstance(item, dict): continue code = _clean_text(item.get("codigo") or item.get("codigo_cups")) description = _clean_text(item.get("descripcion") or item.get("nombre") or item.get("procedimiento")) if code or description: procedures.append({"code": code, "description": description}) if procedures: return procedures servicios = _resolve_dict(factura_json.get("servicios_procedimientos")) for item in servicios.get("procedimientos_quirurgicos") or []: if not isinstance(item, dict): continue code = _clean_text(item.get("codigo_cups")) description = _clean_text(item.get("descripcion")) if code or description: procedures.append({"code": code, "description": description}) return procedures def _build_medication_list(context: dict[str, Any], factura_json: dict[str, Any]) -> list[str]: display = [ _clean_text(item) for item in context.get("medicamentos_hc_display") or [] if _clean_text(item) ] if display: return display meds: list[str] = [] servicios = _resolve_dict(factura_json.get("servicios_procedimientos")) for item in servicios.get("medicamentos") or []: if not isinstance(item, dict): continue value = _clean_text(item.get("medicamento") or item.get("nombre") or item.get("texto_original")) if value: meds.append(value) return meds def _build_diagnostic_support_list(context: dict[str, Any]) -> list[str]: ayudas = [] for item in context.get("ayudas_diagnosticas") or []: if not isinstance(item, dict): continue value = _clean_text(item.get("nombre") or item.get("texto_presentacion") or item.get("tipo")) if value: ayudas.append(value) if ayudas: return ayudas return [_clean_text(item) for item in context.get("imagenes_diagnosticas") or [] if _clean_text(item)]