from __future__ import annotations import inspect import unittest from datetime import UTC from types import MethodType, SimpleNamespace from typing import Any, cast from unittest.mock import patch from app.batch_processing.infrastructure.mongo_repositories import MongoBatchCaseRepository from app.case_epicrisis.application.utils import extraer_procedimientos_factura from app.config import config from app.llm import GeminiAdapter, GroqAdapter, LLMOutputKind, LLMStructuredRequest, LLMTask from app.llm.schemas import ( DiagnosticoItem, DocumentoQuirurgicoStructured, FacturaStructured, HistoriaClinicaStructured, LaboratorioStructured, MedicamentoItem, PrescripcionStructured, ProcedimientoItem, RadiologiaStructured, normalize_clinical_date_value, ) from app.services.clinical_document_projection import serialize_analysis_document from app.services.clinical_document_service import ( CaseIdentityResolution, ClinicalDocumentRequest, ClinicalDocumentService, ) from app.services.clinical_extraction_quality import assess_clinical_extraction from app.services.clinical_processing import ( extraer_antecedentes_historia, extraer_diagnosticos_quirurgicos, extraer_factura_json, extraer_medicamentos_historia, extraer_metadatos_historia, extraer_nombre_y_resumen_historia, extraer_procedimientos_historia, extraer_procedimientos_quirurgicos, extraer_secciones_quirurgicas, procesar_documento_generico, procesar_documento_quirurgico, procesar_factura, procesar_laboratorio, procesar_prescripcion, procesar_radiologia, ) from app.services.clinical_structured_extraction import ClinicalStructuredExtractionService from app.services.soat_processing import _extraer_seccion_documento _FACTURA_HTML_LEGACY = """

Nombre del paciente

Paciente Legacy

1. INFORMACIÓN DEL PROVEEDOR

Nombre de la institución médicaClinica Legacy
NIT900123

2. INFORMACIÓN DEL LA FACTURA

Número de facturaFAC-LEG-1
Número de casoCASE-LEG-1

3. INFORMACIÓN DEL PAGADOR

Aseguradora/EPSEPS Demo

4. INFORMACIÓN DEL PACIENTE

Nombre completoPaciente Legacy
Número de identificación (CC)CC 999

5. SERVICIOS Y PROCEDIMIENTOS MÉDICOS

Procedimientos quirúrgicos

CIRUGIA555555Proc factura legacy110001000

Exámenes de laboratorio

HemogramaLeucocitosis1100100

Imagenología

RX de tóraxSin lesión aguda1200200

Medicamentos

Dipirona1 g IV1 ampolla5050

6. ANÁLISIS FINANCIERO

Valor total de la factura$1350

7. OBSERVACIONES IMPORTANTES

""" _HISTORIA_HTML_LEGACY = """

Nombre del paciente

Paciente Legacy HC

Prestador de servicio

Clinica Legacy HC

Número de caso

CASE-HC-1

Tipo y número de documento

CC 12345

Sexo

F

Edad

45

Fecha de ingreso

2026-05-01

Fecha de nacimiento

1980-01-01

Motivo de consulta

dolor abdominal intenso

Resumen

Resumen legacy de la historia clínica.

Procedimientos

  1. 123456 Lavado quirúrgico
  2. 654321 Curación avanzada

Medicamentos administrados

""" _QUIRURGICO_HTML_LEGACY = """

Nombre del paciente

Paciente Legacy QX

3. DIAGNÓSTICOS

  1. K81.0 - Colecistitis aguda
  2. K80.2 - Colelitiasis

4. PROCEDIMIENTOS REALIZADOS

  1. 765432 Colecistectomia laparoscopica
  2. 123456 Colangiografia intraoperatoria

5. HALLAZGOS QUIRÚRGICOS

Vesicula distendida con proceso inflamatorio agudo.

6. DESCRIPCIÓN DEL PROCEDIMIENTO

Se realiza diseccion del triangulo de Calot y retiro de vesicula.

""" _QUIRURGICO_HTML_LEGACY_ALT = """

DIAGNÓSTICOS

  1. S82.2 - Fractura de tibia

PROCEDIMIENTOS REALIZADOS

  1. 654321 Osteosintesis de tibia

HALLAZGOS QUIRÚRGICOS

Fractura diafisaria desplazada.

DESCRIPCIÓN DEL PROCEDIMIENTO

Reduccion abierta y fijacion interna.

""" _QUIRURGICO_TEXT_LEGACY = """ Diagnósticos prequirúrgico Fractura expuesta de tibia derecha procedimientos realizados Lavado quirurgico y osteosintesis Hallazgos quirúrgicos Compromiso de tejidos blandos con contaminacion moderada Descripción del procedimiento Se realiza lavado, desbridamiento y fijacion interna Justificación del procedimiento Control de daño y estabilizacion osea Diagnósticos postquirúrgico Fractura de tibia estabilizada """ class GeminiStructuredOutputTest(unittest.TestCase): def test_generates_structured_output_with_json_text(self) -> None: payload = HistoriaClinicaStructured.model_validate( { "patient_name": "PACIENTE UNO", "resumen_clinico": "Resumen clínico", "diagnosticos": [], "procedimientos": [], "medicamentos": [], } ).model_dump_json(by_alias=True, exclude_none=True, exclude_defaults=True) models = SimpleNamespace(generate_content=lambda **kwargs: SimpleNamespace(text=payload)) adapter = GeminiAdapter( api_key="secret", default_model="gemini-2.5-flash-lite", client=SimpleNamespace(models=models), ) result = adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=HistoriaClinicaStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(result.provider, "gemini") self.assertEqual(result.content["dt"], "historia_clinica") self.assertEqual(result.content["pn"], "PACIENTE UNO") class GroqStructuredOutputTest(unittest.TestCase): def test_generates_structured_output_with_json_text(self) -> None: payload = LaboratorioStructured.model_validate( { "patient_name": "PACIENTE DOS", "tipo_examen": "Hemograma", "resultados": [], "valores_alterados": [], "interpretacion": "Leucocitosis leve", } ).model_dump_json(by_alias=True, exclude_none=True, exclude_defaults=True) fake_response = SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content=payload))]) fake_client = SimpleNamespace( chat=SimpleNamespace(completions=SimpleNamespace(create=lambda **kwargs: fake_response)) ) adapter = GroqAdapter(api_key="secret", default_model="openai/gpt-oss-120b", client=fake_client) result = adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=LaboratorioStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(result.provider, "groq") self.assertEqual(result.content["dt"], "laboratorio") self.assertEqual(result.content["te"], "Hemograma") class AnalysisProjectionTest(unittest.TestCase): def test_serialize_analysis_document_derives_html_from_structured_payload(self) -> None: serialized = serialize_analysis_document( { "_id": "doc-1", "tipo_documento": "laboratorio", "analysis_structured": LaboratorioStructured.model_validate( { "patient_name": "PACIENTE DOS", "tipo_examen": "Hemograma", "interpretacion": "Leucocitosis leve", "resultados": [], "valores_alterados": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), } ) self.assertIsNotNone(serialized) assert serialized is not None self.assertIn("Hemograma", serialized["analisis_html"]) self.assertIn("PACIENTE DOS", serialized["analisis_html"]) def test_clinical_dates_normalize_to_bogota_and_render_as_machine_values(self) -> None: self.assertEqual( normalize_clinical_date_value("2024-05-07T14:38Z", allow_time=True), "2024-05-07T09:38-05:00", ) self.assertEqual( normalize_clinical_date_value("2024-05-07T10:38-04:00", allow_time=True), "2024-05-07T09:38-05:00", ) self.assertEqual(normalize_clinical_date_value("1972-05-29"), "1972-05-29") self.assertIsNone(normalize_clinical_date_value("fecha desconocida", allow_time=True)) payloads = ( ( "laboratorio", LaboratorioStructured.model_validate({"pn": "Paciente", "fe": "2024-05-07T14:38Z"}), "2024-05-07T09:38-05:00", ), ( "radiologia", RadiologiaStructured.model_validate({"pn": "Paciente", "fe": "07/05/2024 09:38"}), "2024-05-07T09:38-05:00", ), ( "prescripcion", PrescripcionStructured.model_validate({"pn": "Paciente", "fe": "2024-05-07"}), "2024-05-07", ), ( "quirurgico", DocumentoQuirurgicoStructured.model_validate( {"pn": "Paciente", "fp": "2024-05-07T09:38-05:00"} ), "2024-05-07T09:38-05:00", ), ) for document_type, model, expected_date in payloads: with self.subTest(document_type=document_type): serialized = serialize_analysis_document( { "tipo_documento": document_type, "analysis_structured": model.model_dump( by_alias=True, exclude_none=True, exclude_defaults=True, ), } ) assert serialized is not None self.assertIn('data-local-date="true"', serialized["analisis_html"]) self.assertIn(f'datetime="{expected_date}"', serialized["analisis_html"]) def test_historia_render_marks_birth_and_admission_dates_for_localization(self) -> None: serialized = serialize_analysis_document( { "tipo_documento": "historia_clinica", "analysis_structured": HistoriaClinicaStructured.model_validate( { "pn": "Paciente", "fn": "1972-05-29", "fi": "2024-05-07T14:38Z", } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), } ) assert serialized is not None self.assertIn('datetime="1972-05-29"', serialized["analisis_html"]) self.assertIn('datetime="2024-05-07T09:38-05:00"', serialized["analisis_html"]) class ClinicalProcessingFacadeTest(unittest.TestCase): def test_active_processing_facades_no_longer_embed_html_prompt_instructions(self) -> None: active_facades = ( procesar_laboratorio, procesar_radiologia, procesar_prescripcion, procesar_documento_quirurgico, procesar_documento_generico, procesar_factura, ) for facade in active_facades: source = inspect.getsource(facade) self.assertNotIn("Formato de salida en HTML", source) self.assertNotIn("HTML válido", source) self.assertNotIn("DEVUELVE EXCLUSIVAMENTE HTML", source) class ClinicalStructuredExtractorsTest(unittest.TestCase): def test_historia_extractors_use_analysis_structured_before_html(self) -> None: document = { "tipo_documento": "historia_clinica", "analysis_structured": HistoriaClinicaStructured.model_validate( { "patient_name": "PACIENTE ESTRUCTURADO", "prestador_servicio": "Clinica Central", "numero_caso": "CASE-42", "identificacion_paciente": "CC 123", "sexo": "f", "edad": "45", "fecha_nacimiento": "1980-01-01", "fecha_ingreso": "2026-05-01", "motivo_consulta": "dolor abdominal", "resumen_clinico": "Resumen estructurado", "procedimientos": [{"codigo": "123456", "descripcion": "Lavado quirúrgico"}], "medicamentos": [ { "codigo": "7701", "nombre": "Dipirona", "dosis": "1 g", "via": "IV", "frecuencia": "Cada 8 horas", "cantidad": "1 ampolla", } ], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), "analisis_html": "

Resumen

NO USAR

", } metadatos = extraer_metadatos_historia(document) nombre, resumen = extraer_nombre_y_resumen_historia(document) procedimientos = extraer_procedimientos_historia(document) medicamentos = extraer_medicamentos_historia(document) self.assertEqual(metadatos["nombre_paciente"], "PACIENTE ESTRUCTURADO") self.assertEqual(metadatos["prestador_servicio"], "Clinica Central") self.assertEqual(metadatos["motivo_consulta"], "DOLOR ABDOMINAL") self.assertEqual(nombre, "PACIENTE ESTRUCTURADO") self.assertEqual(resumen, "Resumen estructurado") self.assertEqual(procedimientos, ["123456 Lavado quirúrgico"]) self.assertEqual(medicamentos[0]["medicamento"], "Dipirona") self.assertEqual(medicamentos[0]["codigo"], "7701") def test_historia_extractors_repair_fields_and_separate_antecedents(self) -> None: document = { "tipo_documento": "historia_clinica", "descripcion": ( "ERP: EPS SURA\n" "Edad: 47 años\n" "Fecha de ingreso: 07/05/2024 09:38\n" "Motivo de consulta: dolor abdominal intenso posterior a trauma\n" "Antecedentes quirúrgicos:\n" "Apendicectomia hace 10 años\n" "Medicamentos administrados:\n" "Dipirona 1 g IV cada 8 horas\n" ), "analysis_structured": HistoriaClinicaStructured.model_validate( { "patient_name": "PACIENTE ESTRUCTURADO", "prestador_servicio": None, "edad": None, "fecha_ingreso": None, "motivo_consulta": "CASADO", "resumen_clinico": ( "Paciente consulta por dolor abdominal posterior a trauma. " "Se administra Dipirona 1 g IV. " "Se documenta manejo inicial y seguimiento clínico." ), "diagnosticos": [], "procedimientos": [{"descripcion": "Lavado quirúrgico actual"}], "procedimientos_antecedentes": [{"descripcion": "Colecistectomia previa"}], "antecedentes": [], "medicamentos": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), } metadatos = extraer_metadatos_historia(document) procedimientos = extraer_procedimientos_historia(document) antecedentes = extraer_antecedentes_historia(document) medicamentos = extraer_medicamentos_historia(document) self.assertEqual(metadatos["prestador_servicio"], "EPS SURA") self.assertEqual(metadatos["edad"], "47 años") self.assertIn("2024-05-07T", metadatos["fecha_ingreso"]) self.assertEqual(metadatos["motivo_consulta"], "DOLOR ABDOMINAL INTENSO POSTERIOR A TRAUMA") self.assertEqual(procedimientos, ["Lavado quirúrgico actual"]) self.assertIn("Apendicectomia hace 10 años", antecedentes) self.assertNotIn("Colecistectomia previa", antecedentes) self.assertTrue(any(item["medicamento"] == "Dipirona" for item in medicamentos)) def test_historia_extractors_repair_valle_salud_abbreviated_dates(self) -> None: document = { "tipo_documento": "historia_clinica", "descripcion": ( "Edad: 51 AÑOS\nFec. Nacim. : 29/05/1972 Fecha Ing.: 07/05/2024 Hora Ing.: 09:38\n" ), "analysis_structured": { "dt": "historia_clinica", "pn": "PACIENTE PRUEBA", "fn": "NOMBRE INCORRECTAMENTE EXTRAIDO", "fi": "No especificado", "rs": "Paciente con información clínica disponible.", }, } metadatos = extraer_metadatos_historia(document) self.assertEqual(metadatos["fecha_nacimiento"], "1972-05-29") self.assertEqual(metadatos["fecha_ingreso"], "2024-05-07T09:38-05:00") self.assertEqual(metadatos["edad"], "51 AÑOS") def test_historia_structured_model_strips_html_from_llm_fields(self) -> None: model = HistoriaClinicaStructured.model_validate( { "pn": "PACIENTE HTML", "rs": "

Paciente con manejo clínico y evolución estable.

", "dx": [{"d": "Fractura de radio"}], "pr": [], "md": [{"n": "Dipirona", "tu": "administrado"}], } ) self.assertEqual(model.patient_name, "PACIENTE HTML") self.assertEqual(model.resumen_clinico, "Paciente con manejo clínico y evolución estable.") self.assertEqual(model.diagnosticos[0].descripcion, "Fractura de radio") self.assertEqual(model.medicamentos[0].nombre, "Dipirona") def test_historia_extractors_reject_administrative_structured_summary_and_recover_raw_summary( self, ) -> None: document = { "tipo_documento": "historia_clinica", "descripcion": ( "EPS: SEGUROS COMERCIALES BOLIVAR S.A.\n" "Resumen\n" "Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.\n" ), "analysis_structured": HistoriaClinicaStructured.model_validate( { "patient_name": "PACIENTE ESTRUCTURADO", "resumen_clinico": "SEGUROS COMERCIALES BOLIVAR S.A.", "diagnosticos": [], "procedimientos": [], "medicamentos": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), } metadatos = extraer_metadatos_historia(document) nombre, resumen = extraer_nombre_y_resumen_historia(document) self.assertEqual(nombre, "PACIENTE ESTRUCTURADO") self.assertEqual( metadatos["resumen"], "Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.", ) self.assertEqual( resumen, "Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.", ) def test_quirurgico_extractors_use_analysis_structured_before_html(self) -> None: document = { "tipo_documento": "quirurgico", "analysis_structured": DocumentoQuirurgicoStructured.model_validate( { "patient_name": "PACIENTE QX", "procedimiento_principal": "Colecistectomia", "descripcion_procedimiento": "Descripcion estructurada", "hallazgos": "Hallazgos estructurados", "diagnosticos": [ DiagnosticoItem.model_validate( {"codigo": "K81.0", "descripcion": "Colecistitis aguda"} ) ], "procedimientos": [ ProcedimientoItem.model_validate( { "codigo": "765432", "descripcion": "Colecistectomia laparoscopica", } ) ], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), "analisis_html": "

5. HALLAZGOS QUIRÚRGICOS

NO USAR

", } procedimientos = extraer_procedimientos_quirurgicos(document) diagnosticos = extraer_diagnosticos_quirurgicos(document) hallazgos, descripcion = extraer_secciones_quirurgicas(document) self.assertEqual(procedimientos, ["765432 Colecistectomia laparoscopica"]) self.assertEqual(diagnosticos, ["K81.0 - Colecistitis aguda"]) self.assertEqual(hallazgos, "Hallazgos estructurados") self.assertEqual(descripcion, "Descripcion estructurada") def test_quirurgico_extractors_parse_legacy_html_without_structure(self) -> None: document = {"analisis_html": _QUIRURGICO_HTML_LEGACY} procedimientos = extraer_procedimientos_quirurgicos(document) diagnosticos = extraer_diagnosticos_quirurgicos(document) hallazgos, descripcion = extraer_secciones_quirurgicas(document) self.assertEqual( procedimientos, ["765432 Colecistectomia laparoscopica", "123456 Colangiografia intraoperatoria"], ) self.assertEqual(diagnosticos, ["K81.0 - Colecistitis aguda", "K80.2 - Colelitiasis"]) self.assertEqual(hallazgos, "Vesicula distendida con proceso inflamatorio agudo.") self.assertEqual( descripcion, "Se realiza diseccion del triangulo de Calot y retiro de vesicula.", ) def test_quirurgico_extractors_parse_legacy_html_without_numbered_headings(self) -> None: document = {"analisis_html": _QUIRURGICO_HTML_LEGACY_ALT} procedimientos = extraer_procedimientos_quirurgicos(document) diagnosticos = extraer_diagnosticos_quirurgicos(document) hallazgos, descripcion = extraer_secciones_quirurgicas(document) self.assertEqual(procedimientos, ["654321 Osteosintesis de tibia"]) self.assertEqual(diagnosticos, ["S82.2 - Fractura de tibia"]) self.assertEqual(hallazgos, "Fractura diafisaria desplazada.") self.assertEqual(descripcion, "Reduccion abierta y fijacion interna.") def test_historia_extractors_parse_legacy_html_without_structure(self) -> None: document = {"analisis_html": _HISTORIA_HTML_LEGACY} metadatos = extraer_metadatos_historia(document) nombre, resumen = extraer_nombre_y_resumen_historia(document) procedimientos = extraer_procedimientos_historia(document) medicamentos = extraer_medicamentos_historia(document) self.assertEqual(metadatos["nombre_paciente"], "Paciente Legacy HC") self.assertEqual(metadatos["prestador_servicio"], "Clinica Legacy HC") self.assertEqual(metadatos["caso"], "CASE-HC-1") self.assertEqual(metadatos["datos_identificacion_paciente"], "CC 12345") self.assertEqual(metadatos["sexo"], "F") self.assertEqual(metadatos["edad"], "45") self.assertEqual(metadatos["fecha_ingreso"], "2026-05-01") self.assertEqual(metadatos["fecha_nacimiento"], "1980-01-01") self.assertEqual(metadatos["motivo_consulta"], "DOLOR ABDOMINAL INTENSO") self.assertEqual(metadatos["resumen"], "Resumen legacy de la historia clínica.") self.assertEqual(nombre, "Paciente Legacy HC") self.assertEqual(resumen, "Resumen legacy de la historia clínica.") self.assertEqual(procedimientos, ["123456 Lavado quirúrgico", "654321 Curación avanzada"]) self.assertEqual( {item["medicamento"] for item in medicamentos}, {"Dipirona", "Acetaminofen"}, ) def test_historia_extractors_fallback_to_legacy_when_structured_metadata_is_low_quality(self) -> None: document = { "tipo_documento": "historia_clinica", "analysis_structured": HistoriaClinicaStructured.model_validate( { "pn": "PACIENTE ESTRUCTURADO", "ps": "IPS", "mc": "Motivo de consulta", "rs": "Resumen estructurado", "pr": [], "md": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), "analisis_html": _HISTORIA_HTML_LEGACY, } metadatos = extraer_metadatos_historia(document) self.assertEqual(metadatos["nombre_paciente"], "PACIENTE ESTRUCTURADO") self.assertEqual(metadatos["prestador_servicio"], "Clinica Legacy HC") self.assertEqual(metadatos["motivo_consulta"], "DOLOR ABDOMINAL INTENSO") def test_historia_extractors_merge_structured_and_legacy_medicamentos(self) -> None: document = { "tipo_documento": "historia_clinica", "analysis_structured": HistoriaClinicaStructured.model_validate( { "pn": "PACIENTE ESTRUCTURADO", "rs": "Resumen estructurado", "pr": [], "md": [ MedicamentoItem.model_validate( {"c": "2801", "n": "No especificado", "ds": "", "q": ""} ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) ], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), "analisis_html": _HISTORIA_HTML_LEGACY, } medicamentos = extraer_medicamentos_historia(document) self.assertTrue(any(item["medicamento"] == "Dipirona" for item in medicamentos)) self.assertTrue(any(item["medicamento"] == "Acetaminofen" for item in medicamentos)) self.assertFalse( any( item["medicamento"] == "No especificado" and not item.get("dosis") and not item.get("cantidad") for item in medicamentos ) ) def test_quirurgico_extractors_fallback_to_legacy_sections_when_structured_is_generic(self) -> None: document = { "tipo_documento": "quirurgico", "analysis_structured": DocumentoQuirurgicoStructured.model_validate( { "pn": "PACIENTE QX", "pp": "Colecistectomia", "dp": "Descripcion estructurada", "hl": "Hallazgos estructurados", "dx": [], "pr": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), "analisis_html": _QUIRURGICO_HTML_LEGACY, } hallazgos, descripcion = extraer_secciones_quirurgicas(document) self.assertEqual(hallazgos, "Vesicula distendida con proceso inflamatorio agudo.") self.assertEqual( descripcion, "Se realiza diseccion del triangulo de Calot y retiro de vesicula.", ) def test_factura_extractor_uses_analysis_structured_before_html(self) -> None: payload = { "dt": "factura", "pn": "PACIENTE FACTURA", "pv": {"ni": "Clinica Central"}, "df": {"nf": "FAC-001"}, "sp": { "pq": [{"cc": "555555", "d": "Proc factura estructurada"}], "im": [{"es": "RX de tórax", "d": "Sin lesión aguda"}], "me": [{"md": "Dipirona", "q": "1 ampolla", "do": "1 g IV"}], }, "rf": {"vt": "$1000", "ob": "Observación estructurada"}, } document = { "tipo_documento": "factura", "analysis_structured": FacturaStructured.model_validate(payload).model_dump( by_alias=True, exclude_none=True, exclude_defaults=True, ), "analisis_html": "

NO USAR

", } factura_json = extraer_factura_json(document) self.assertEqual(factura_json["nombre_paciente"], "PACIENTE FACTURA") self.assertEqual( factura_json["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["descripcion"], "Proc factura estructurada", ) self.assertEqual( factura_json["servicios_procedimientos"]["medicamentos"][0]["medicamento"], "Dipirona" ) self.assertEqual(factura_json["observaciones"], ["Observación estructurada"]) def test_factura_extractor_prefers_persisted_factura_json_before_html(self) -> None: document = { "nombre_paciente": "PACIENTE PERSISTIDO", "factura_json": { "proveedor": {"nombre": "Clinica Persistida"}, "informacion_factura": {"numero_factura": "FAC-PERSISTIDA"}, "pagador": {"aseguradora_eps": "EPS Persistida"}, "informacion_paciente": {"nombre_completo": "PACIENTE PERSISTIDO"}, "servicios_procedimientos": { "procedimientos_quirurgicos": [ {"codigo_cups": "111111", "descripcion": "Proc persistido"} ], "medicamentos": [{"medicamento": "Dipirona"}], }, "analisis_financiero": {"valor_total_factura": "$2000"}, "observaciones": ["Observación persistida"], }, "analisis_html": _FACTURA_HTML_LEGACY, } factura_json = extraer_factura_json(document) procedimientos = extraer_procedimientos_factura(document) self.assertEqual(factura_json["nombre_paciente"], "PACIENTE PERSISTIDO") self.assertEqual( factura_json["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["descripcion"], "Proc persistido", ) self.assertEqual(factura_json["observaciones"], ["Observación persistida"]) self.assertEqual( procedimientos, [ { "codigo_cups": "111111", "codigo_facturacion": "", "codigo_referencia": "111111", "descripcion": "Proc persistido", } ], ) def test_factura_extractors_parse_legacy_html_without_structure(self) -> None: document = {"analisis_html": _FACTURA_HTML_LEGACY} factura_json = extraer_factura_json(document) procedimientos = extraer_procedimientos_factura(document) self.assertEqual(factura_json["nombre_paciente"], "Paciente Legacy") self.assertEqual(factura_json["proveedor"]["nombre_institucion"], "Clinica Legacy") self.assertEqual(factura_json["informacion_factura"]["numero_factura"], "FAC-LEG-1") self.assertEqual( factura_json["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["descripcion"], "Proc factura legacy", ) self.assertEqual( factura_json["servicios_procedimientos"]["medicamentos"][0]["medicamento"], "Dipirona", ) self.assertEqual(factura_json["observaciones"], ["Observación legacy"]) self.assertEqual( procedimientos, [{"codigo_soat": "555555", "descripcion": "Proc factura legacy"}], ) class ClinicalStructuredExtractionReliabilityTest(unittest.TestCase): class _EscalatingStructuredRouter: def __init__(self, *, first_payload, second_payload) -> None: self.first_payload = first_payload self.second_payload = second_payload self.calls: list[dict[str, object]] = [] def _resolve_route(self, task, metadata=None): risk_level = str((metadata or {}).get("risk_level") or "low") model = ( config.GEMINI_MODEL_EXTRACT_HIGH_RISK if risk_level == "high" else config.GEMINI_MODEL_EXTRACT_LOW_RISK ) return SimpleNamespace(task=task, provider="gemini", model=model, metadata=dict(metadata or {})) def generate_structured(self, request): risk_level = str((request.metadata or {}).get("risk_level") or "low") model = ( config.GEMINI_MODEL_EXTRACT_HIGH_RISK if risk_level == "high" else config.GEMINI_MODEL_EXTRACT_LOW_RISK ) self.calls.append({"model": model, **dict(request.metadata or {})}) payload = ( self.second_payload if model == config.GEMINI_MODEL_EXTRACT_HIGH_RISK else self.first_payload ) return SimpleNamespace( content=payload, provider="gemini", model=model, metrics={}, output_kind=request.output_kind, ) def test_historia_extraction_escalates_to_flash_when_validation_fails(self) -> None: first_payload = HistoriaClinicaStructured.model_validate( { "patient_name": "JUAN DEMO", "resumen_clinico": "SEGUROS COMERCIALES BOLIVAR S.A.", "diagnosticos": [], "procedimientos": [], "medicamentos": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) second_payload = HistoriaClinicaStructured.model_validate( { "patient_name": "JUAN DEMO", "resumen_clinico": "Paciente con trauma torácico, manejo analgésico y vigilancia clínica.", "diagnosticos": [{"codigo": "S299", "descripcion": "Traumatismo del tórax"}], "procedimientos": [{"codigo": "123456", "descripcion": "Observación clínica"}], "medicamentos": [{"codigo": "7701", "nombre": "Dipirona", "dosis": "1 g"}], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) router = self._EscalatingStructuredRouter(first_payload=first_payload, second_payload=second_payload) result = ClinicalStructuredExtractionService(llm_router=router).extract( raw_text=( "Nombre del paciente: JUAN DEMO\n" "Motivo de consulta: trauma torácico.\n" "Diagnósticos: traumatismo del tórax.\n" "Procedimientos: observación clínica.\n" "Medicamentos: dipirona 1 g.\n" ), document_type="historia_clinica", ) self.assertEqual(len(router.calls), 2) assert result.route_metadata is not None assert result.quality_metadata is not None self.assertTrue(result.route_metadata["escalated"]) self.assertEqual(result.route_metadata["selected_model"], config.GEMINI_MODEL_EXTRACT_HIGH_RISK) self.assertEqual(result.quality_metadata["status"], "ok") self.assertEqual(result.analysis_structured["rs"], second_payload["rs"]) def test_quirurgico_extraction_escalates_to_flash_when_sections_are_sparse(self) -> None: first_payload = DocumentoQuirurgicoStructured.model_validate( { "patient_name": "JUAN PEREZ", "procedimiento_principal": "", "descripcion_procedimiento": "", "hallazgos": "", "diagnosticos": [], "procedimientos": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) second_payload = DocumentoQuirurgicoStructured.model_validate( { "patient_name": "JUAN PEREZ", "resumen_clinico": ( "Se realizó osteosíntesis por fractura desplazada de clavícula, con fijación y sin " "complicaciones documentadas." ), "procedimiento_principal": "Osteosíntesis de clavícula", "descripcion_procedimiento": "Se realiza osteosíntesis con material de fijación.", "hallazgos": "Fractura desplazada de clavícula.", "diagnosticos": [{"codigo": "S42.0", "descripcion": "Fractura de clavícula"}], "procedimientos": [{"codigo": "235401", "descripcion": "Osteosíntesis de clavícula"}], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) router = self._EscalatingStructuredRouter(first_payload=first_payload, second_payload=second_payload) result = ClinicalStructuredExtractionService(llm_router=router).extract( raw_text=( "Juan Perez\nProcedimiento: osteosíntesis de clavícula.\nHallazgos: fractura desplazada.\n" ), document_type="quirurgico", ) self.assertEqual(len(router.calls), 2) assert result.route_metadata is not None assert result.quality_metadata is not None self.assertTrue(result.route_metadata["escalated"]) self.assertEqual(result.route_metadata["selected_model"], config.GEMINI_MODEL_EXTRACT_HIGH_RISK) self.assertEqual(result.quality_metadata["status"], "ok") self.assertEqual(result.analysis_structured["pp"], "Osteosíntesis de clavícula") self.assertIn("rs", result.analysis_structured) self.assertLess( result.rendered_html.index("Resumen"), result.rendered_html.index("FECHA DEL PROCEDIMIENTO"), ) def test_quirurgico_missing_summary_alone_triggers_quality_retry(self) -> None: common = { "patient_name": "JUAN PEREZ", "procedimiento_principal": "Osteosíntesis", "descripcion_procedimiento": "Se realiza fijación interna.", "hallazgos": "Fractura desplazada.", "diagnosticos": [{"codigo": "S42.0", "descripcion": "Fractura"}], "procedimientos": [{"codigo": "235401", "descripcion": "Osteosíntesis"}], } first_payload = DocumentoQuirurgicoStructured.model_validate(common).model_dump( by_alias=True, exclude_none=True, exclude_defaults=True ) second_payload = DocumentoQuirurgicoStructured.model_validate( {**common, "resumen_clinico": "Se realizó osteosíntesis por fractura desplazada."} ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) router = self._EscalatingStructuredRouter( first_payload=first_payload, second_payload=second_payload, ) service = ClinicalStructuredExtractionService(llm_router=router) self.assertEqual(service._get_prompt_version("quirurgico"), "v3") result = service.extract( raw_text="Paciente Juan Perez con cirugía realizada.", document_type="quirurgico", ) self.assertEqual(len(router.calls), 2) assert result.quality_metadata is not None assert result.route_metadata is not None self.assertEqual(result.analysis_structured["rs"], second_payload["rs"]) self.assertEqual(result.quality_metadata["status"], "ok") self.assertIn("summary_missing", result.route_metadata["attempts"][0]["quality_reasons"]) def test_pop_prompt_is_context_only_and_quality_flags_missing_procedure(self) -> None: service = ClinicalStructuredExtractionService(llm_router=SimpleNamespace()) prompt = service._build_prompt( # noqa: SLF001 document_type="historia_clinica", raw_text="Evolución: POP sin procedimiento explícito.", summary_text=None, model_cls=HistoriaClinicaStructured, schema_name="historia_clinica", ) self.assertIn("POP nunca es por sí mismo un procedimiento", prompt) self.assertIn("no inventes un procedimiento", prompt) quality = assess_clinical_extraction( document_type="historia_clinica", raw_text="Evolución\nPOP sin procedimiento explícito.", analysis_model=HistoriaClinicaStructured.model_validate( {"patient_name": "JUAN DEMO", "resumen_clinico": "Resumen"} ), ) self.assertIn("postoperatorio_procedure_missing", quality.reason_codes) self.assertTrue(quality.should_retry) class SoatSectionExtractionTest(unittest.TestCase): def test_extracts_known_sections_with_shared_normalization(self) -> None: self.assertEqual( _extraer_seccion_documento(_QUIRURGICO_TEXT_LEGACY, "HALLAZGOS QUIRÚRGICOS"), "COMPROMISO DE TEJIDOS BLANDOS CON CONTAMINACION MODERADA", ) self.assertEqual( _extraer_seccion_documento(_QUIRURGICO_TEXT_LEGACY, "DESCRIPCIÓN DEL PROCEDIMIENTO"), "SE REALIZA LAVADO DESBRIDAMIENTO Y FIJACION INTERNA", ) def test_returns_empty_string_when_section_is_missing(self) -> None: self.assertEqual(_extraer_seccion_documento(_QUIRURGICO_TEXT_LEGACY, "SECCIÓN INEXISTENTE"), "") def test_tolerates_case_and_whitespace_variations(self) -> None: document = " diagnosticos prequirurgico \n Trauma severo \n\n procedimientos realizados\n Curacion avanzada " self.assertEqual( _extraer_seccion_documento(document, "DIAGNÓSTICOS PREQUIRÚRGICO"), "TRAUMA SEVERO", ) self.assertEqual( _extraer_seccion_documento(document, "PROCEDIMIENTOS REALIZADOS"), "CURACION AVANZADA", ) class ClinicalDocumentServiceStructuredPersistenceTest(unittest.TestCase): class FakeInsertResult: inserted_id = "analysis-1" class FakeCollection: def __init__(self) -> None: self.inserted: list[dict] = [] self.deleted: list[dict] = [] self.find_one_result = None def insert_one(self, payload): self.inserted.append(dict(payload)) return ClinicalDocumentServiceStructuredPersistenceTest.FakeInsertResult() def find_one(self, *args, **kwargs): result = self.find_one_result return dict(result) if isinstance(result, dict) else result def delete_many(self, query): self.deleted.append(dict(query)) return SimpleNamespace(deleted_count=0) class FakeLegacyStorageCollection: def find_one(self, *args, **kwargs): return None class FakeRecordsDataFrame: def __init__(self, records): self._records = list(records) def to_dict(self, orient="records"): assert orient == "records" return list(self._records) class FakeRepairCursor: def __init__(self, items): self.items = list(items) def sort(self, _fields): return self def limit(self, value): self.items = self.items[: max(1, int(value or 1))] return self def __iter__(self): return iter(self.items) class FakeRepairCollection: def __init__(self, items): self.items = [dict(item) for item in items] self.update_calls: list[tuple[dict, dict]] = [] def find(self, _query): return ClinicalDocumentServiceStructuredPersistenceTest.FakeRepairCursor(self.items) def update_one(self, query, update): self.update_calls.append((dict(query), dict(update))) target_id = query.get("_id") for item in self.items: if item.get("_id") == target_id: item.update(dict(update.get("$set") or {})) break def test_process_and_persist_stores_analysis_structured_without_analisis_html(self) -> None: collection = self.FakeCollection() llm_router = SimpleNamespace( generate_structured=lambda request: SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content=LaboratorioStructured.model_validate( { "patient_name": "PACIENTE TRES", "tipo_examen": "Hemograma", "interpretacion": "Normal", "resultados": [], "valores_alterados": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), ) ) service = ClinicalDocumentService( mongo_storage=SimpleNamespace(guardar_analisis=lambda **kwargs: "legacy-1"), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=llm_router, ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace(get_user_case=lambda username, case_key: None), ) result = service.process_and_persist( ClinicalDocumentRequest( raw_text="Paciente: Paciente Tres\nExamen: Hemograma", detected_type="laboratorio", username="tester", original_name="lab.pdf", ) ) self.assertEqual(result["nombre_paciente"], "PACIENTE TRES") self.assertIn("analysis_structured", result) self.assertIn("analisis_html", result) self.assertIn("analysis_route", result) self.assertIn("analysis_quality", result) self.assertTrue(collection.inserted) inserted = collection.inserted[0] self.assertIn("analysis_structured", inserted) self.assertIn("analysis_route", inserted) self.assertIn("analysis_quality", inserted) self.assertNotIn("analisis_html", inserted) def test_process_and_persist_uses_effective_document_type_for_support_mismatches(self) -> None: collection = self.FakeCollection() upsert_calls: list[tuple[str, str, dict[str, object]]] = [] llm_router = SimpleNamespace( generate_structured=lambda request: SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content=RadiologiaStructured.model_validate( { "patient_name": "PACIENTE RX", "tipo_estudio": "RX de tórax", "conclusion": "Sin consolidaciones.", } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), ) ) service = ClinicalDocumentService( mongo_storage=SimpleNamespace(guardar_analisis=lambda **kwargs: "legacy-1"), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=llm_router, ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace( get_user_case=lambda username, case_key: None, upsert_user_case=lambda username, case_key, payload: upsert_calls.append( (username, case_key, dict(payload)) ), ), ) result = service.process_and_persist( ClinicalDocumentRequest( raw_text="Paciente: Paciente RX\nEstudio: RX de tórax\nConclusión: Sin consolidaciones.", detected_type="radiologia", username="tester", original_name="soporte.pdf", case_key="CASE-RX-1", case_number="HC-RX-1", patient_id="111222333", provided_patient_name="PACIENTE RX", selected_document_type="factura", detected_document_type="radiologia", effective_document_type="radiologia", ) ) self.assertEqual(result["tipo_documento"], "radiologia") self.assertEqual(result["selected_document_type"], "factura") self.assertEqual(result["detected_document_type"], "radiologia") self.assertEqual(result["effective_document_type"], "radiologia") inserted = collection.inserted[0] self.assertEqual(inserted["tipo_documento"], "radiologia") self.assertEqual(inserted["document_type"], "radiologia") self.assertEqual(inserted["selected_document_type"], "factura") self.assertEqual(inserted["detected_document_type"], "radiologia") self.assertEqual(inserted["effective_document_type"], "radiologia") self.assertIn("ayudas_diagnosticas", inserted) self.assertNotIn("factura_json", inserted) self.assertEqual( collection.deleted[0], {"usuario": "tester", "tipo_documento": "epicrisis_case_cache", "case_key": "CASE-RX-1"}, ) self.assertEqual(upsert_calls[0][1], "CASE-RX-1") self.assertEqual(upsert_calls[0][2]["epicrisis_status"], "pendiente") self.assertEqual(upsert_calls[0][2]["case_number"], "HC-RX-1") self.assertEqual(upsert_calls[0][2]["patient_id"], "111222333") self.assertEqual(upsert_calls[0][2]["patient_name"], "PACIENTE RX") def test_historia_fallback_uses_structured_extraction_not_legacy_html_generator(self) -> None: collection = self.FakeCollection() service = ClinicalDocumentService( mongo_storage=SimpleNamespace( guardar_analisis=lambda **kwargs: "legacy-1", collection=self.FakeLegacyStorageCollection(), ), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=SimpleNamespace( asignar_codigos_batch=lambda html: self.FakeRecordsDataFrame( [{"codigo": "A00", "descripcion": "Diagnóstico fallback"}] ) ), cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace(get_user_case=lambda username, case_key: None), ) service_for_test = cast(Any, service) service_for_test.structured_extraction_service = SimpleNamespace( extract=lambda **kwargs: SimpleNamespace( analysis_structured=HistoriaClinicaStructured.model_validate( { "patient_name": "PACIENTE FALLBACK", "resumen_clinico": "Resumen fallback", "diagnosticos": [], "procedimientos": [], "medicamentos": [], } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), analysis_model=HistoriaClinicaStructured.model_validate( { "patient_name": "PACIENTE FALLBACK", "resumen_clinico": "Resumen fallback", "diagnosticos": [], "procedimientos": [], "medicamentos": [], } ), rendered_html="

fallback estructurado

", ) ) historia_request = SimpleNamespace( extraer_historia_estructurada=lambda raw_text, llm_router=None: (_ for _ in ()).throw( RuntimeError("fallo historia principal") ) ) with patch( "app.services.clinical_document_service._load_historia_clinica_request_class", return_value=historia_request, ): result = service.process_and_persist( ClinicalDocumentRequest( raw_text="Historia clínica de fallback", detected_type="historia_clinica", username="tester", original_name="historia.pdf", ) ) self.assertEqual(result["nombre_paciente"], "PACIENTE FALLBACK") self.assertIn("analysis_structured", result) self.assertIsNone(result["error_analisis"]) def test_process_and_persist_normalizes_historia_structured_metadata_before_storage(self) -> None: collection = self.FakeCollection() service = ClinicalDocumentService( mongo_storage=SimpleNamespace( guardar_analisis=lambda **kwargs: "legacy-1", collection=self.FakeLegacyStorageCollection(), ), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace(get_user_case=lambda username, case_key: None), ) structured_model = HistoriaClinicaStructured.model_validate( { "patient_name": "Paciente extraído", "numero_caso": "93387170", "resumen_clinico": "Resumen estructurado", "diagnosticos": [ {"codigo": "S299", "descripcion": "Traumatismo del torax"}, {"descripcion": "Dolor"}, ], "procedimientos": [], "medicamentos": [], } ) historia_request = SimpleNamespace( extraer_historia_estructurada=lambda raw_text, **kwargs: SimpleNamespace( analysis_structured={ **structured_model.model_dump( by_alias=True, exclude_none=True, exclude_defaults=True, ), "fn": "NOMBRE INCORRECTAMENTE EXTRAIDO", "fi": "No especificado", }, analysis_model=structured_model, rendered_html="

HTML previo incorrecto

", ) ) with patch( "app.services.clinical_document_service._load_historia_clinica_request_class", return_value=historia_request, ): result = service.process_and_persist( ClinicalDocumentRequest( raw_text=( "Historia clínica estructurada\n" "Fec. Nacim. : 29/05/1972 Fecha Ing.: 07/05/2024 Hora Ing.: 09:38" ), detected_type="historia_clinica", username="tester", original_name="historia.pdf", case_number="178474", patient_id="93387170", provided_patient_name="JULIO CESAR ACOSTA GUEVARA", ) ) self.assertEqual(result["analysis_structured"]["nc"], "178474") self.assertEqual(result["analysis_structured"]["ip"], "93387170") self.assertEqual(result["analysis_structured"]["pn"], "JULIO CESAR ACOSTA GUEVARA") self.assertEqual(result["analysis_structured"]["fn"], "1972-05-29") self.assertEqual(result["analysis_structured"]["fi"], "2024-05-07T09:38-05:00") self.assertIn("178474", result["analisis_html"]) self.assertIn("93387170", result["analisis_html"]) self.assertEqual(len(result["codigos_cie10"]), 1) self.assertEqual(result["codigos_cie10"][0]["codigo"], "S299") inserted = collection.inserted[0] self.assertNotIn("analisis_html", inserted) self.assertEqual(inserted["analysis_structured"]["nc"], "178474") self.assertEqual(inserted["analysis_structured"]["ip"], "93387170") self.assertEqual(inserted["analysis_structured"]["pn"], "JULIO CESAR ACOSTA GUEVARA") self.assertEqual(inserted["analysis_structured"]["fn"], "1972-05-29") self.assertEqual(inserted["analysis_structured"]["fi"], "2024-05-07T09:38-05:00") def test_process_and_persist_uses_same_canonical_antecedents_everywhere(self) -> None: collection = self.FakeCollection() legacy_payloads: list[dict[str, Any]] = [] service = ClinicalDocumentService( mongo_storage=SimpleNamespace( guardar_analisis=lambda **kwargs: legacy_payloads.append(kwargs) or "legacy-1", collection=self.FakeLegacyStorageCollection(), ), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace(get_user_case=lambda username, case_key: None), ) contaminated = ( "Motivo de consulta: dolor torácico. Examen físico: paciente estable. " "Diagnóstico actual: trauma. Tratamiento: analgesia y observación." ) structured_model = HistoriaClinicaStructured.model_validate( { "pn": "JUAN PEREZ", "rs": "Paciente consulta por trauma y recibe valoración clínica.", "an": [contaminated], } ) historia_request = SimpleNamespace( extraer_historia_estructurada=lambda raw_text, **kwargs: SimpleNamespace( analysis_structured=structured_model.model_dump( by_alias=True, exclude_none=True, ), analysis_model=structured_model, rendered_html="

HTML previo contaminado

", quality_metadata={"status": "ok", "reason_codes": []}, ) ) with patch( "app.services.clinical_document_service._load_historia_clinica_request_class", return_value=historia_request, ): result = service.process_and_persist( ClinicalDocumentRequest( raw_text=( "Nombre del paciente: JUAN PEREZ\n" "ANTECEDENTES\n" "PATOLÓGICOS: HIPERTENSIÓN ARTERIAL\n" "EXAMEN FÍSICO\n" "Paciente estable." ), detected_type="historia_clinica", username="tester", original_name="historia.pdf", ) ) expected_flat = ["HIPERTENSIÓN ARTERIAL"] self.assertEqual(result["analysis_structured"]["an"], expected_flat) self.assertEqual(collection.inserted[0]["analysis_structured"]["an"], expected_flat) self.assertEqual(legacy_payloads[0]["analysis_structured"]["an"], expected_flat) self.assertEqual( collection.inserted[0]["analysis_structured"]["ae"], legacy_payloads[0]["analysis_structured"]["ae"], ) self.assertNotIn(contaminated, result["analisis_html"]) self.assertTrue(result["review_required"]) def test_process_and_persist_preserves_complete_multicolumn_antecedents(self) -> None: collection = self.FakeCollection() legacy_payloads: list[dict[str, Any]] = [] service = ClinicalDocumentService( mongo_storage=SimpleNamespace( guardar_analisis=lambda **kwargs: legacy_payloads.append(kwargs) or "legacy-1", collection=self.FakeLegacyStorageCollection(), ), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=SimpleNamespace(asignar_codigos_batch=lambda html: self.FakeRecordsDataFrame([])), cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace(get_user_case=lambda username, case_key: None), ) raw_text = ( "ANTECEDENTES\n" "ALERGICOS : ************************* ALERGICO A " "QUIRURGICOS: APENDICECTOMIA, SAFENECTOMIA\n" "LA NITROFURAZONA *********** DERECHA.\n" "REVISIÓN POR SISTEMAS\n" "\f" "ANTECEDENTES\n" "- QUIRURGICOS: APENDICECTOMIA, SAFENECTOMIA DERECHA, " "CIRUGIA OCULAR EN OJO IZQUIERDO\n" "- ALERGICOS: NIEGA\n" "EXAMEN FISICO\n" "Se registra antecedente farmacológico de Tamsulosina para hiperplasia prostática.\n" ) structured_model = HistoriaClinicaStructured.model_validate( { "rs": "Resumen clínico independiente de los antecedentes.", "ae": [ { "ct": "alergico", "d": "NITROFURAZONA", "es": "presente", "ev": "ALERGICO A LA NITROFURAZONA", "og": "explicito", "dc": True, }, { "ct": "quirurgico", "d": "Apendicectomía", "es": "presente", "ev": "APENDICECTOMIA", "og": "explicito", }, { "ct": "quirurgico", "d": "Safenectomía derecha", "es": "presente", "ev": "SAFENECTOMIA DERECHA", "og": "explicito", }, { "ct": "quirurgico", "d": "Cirugía ocular en ojo izquierdo", "es": "presente", "ev": "CIRUGIA OCULAR EN OJO IZQUIERDO", "og": "explicito", }, { "ct": "farmacologico", "d": "Tamsulosina", "cx": "para hiperplasia prostática", "es": "presente", "ev": ("antecedente farmacológico de Tamsulosina para hiperplasia prostática"), "og": "explicito", "cf": "alta", }, ], "pa": [ {"d": "Apendicectomía"}, {"d": "Safenectomía derecha"}, {"d": "Cirugía ocular en ojo izquierdo"}, ], } ) historia_request = SimpleNamespace( extraer_historia_estructurada=lambda raw_text, **kwargs: SimpleNamespace( analysis_structured=structured_model.model_dump( by_alias=True, exclude_none=True, ), analysis_model=structured_model, rendered_html="

HTML estructurado

", quality_metadata={"status": "ok", "reason_codes": []}, route_metadata={ "historia_processing": { "antecedentes": { "accepted": 5, "quality_retries": 0, "reason_codes": [], "policy_version": "v3", } } }, ) ) with patch( "app.services.clinical_document_service._load_historia_clinica_request_class", return_value=historia_request, ): result = service.process_and_persist( ClinicalDocumentRequest( raw_text=raw_text, detected_type="historia_clinica", username="tester", original_name="historia-columnas.pdf", case_number="CASO-ANT-1", patient_id="PACIENTE-ANT-1", provided_patient_name="MARIA LOPEZ", ) ) expected = { "alergia: nitrofurazona", "apendicectomía", "safenectomía derecha", "cirugía ocular en ojo izquierdo", "farmacológico: tamsulosina — para hiperplasia prostática", } self.assertEqual({item.casefold() for item in result["analysis_structured"]["an"]}, expected) self.assertEqual( result["analysis_structured"]["an"], collection.inserted[0]["analysis_structured"]["an"], ) self.assertEqual( result["analysis_structured"]["an"], legacy_payloads[0]["analysis_structured"]["an"], ) self.assertEqual( result["analysis_structured"]["ae"], collection.inserted[0]["analysis_structured"]["ae"], ) self.assertIn( "FARMACOLÓGICO: Tamsulosina — para hiperplasia prostática", result["analisis_html"], ) self.assertEqual( result["analysis_structured"]["ae"], legacy_payloads[0]["analysis_structured"]["ae"], ) self.assertNotIn("SAFENECTOMIA LA NITROFURAZONA", result["analisis_html"].upper()) self.assertEqual(result["analysis_quality"]["status"], "ok") self.assertFalse(result["review_required"]) metrics = result["historia_processing"]["antecedentes"] self.assertEqual(metrics["accepted"], 5) self.assertEqual(metrics["pharmacological_contexts_accepted"], 1) self.assertEqual(metrics["reason_codes"], []) self.assertEqual(metrics["policy_version"], "v3") def test_process_and_persist_replaces_administrative_historia_summary_before_storage(self) -> None: collection = self.FakeCollection() service = ClinicalDocumentService( mongo_storage=SimpleNamespace( guardar_analisis=lambda **kwargs: "legacy-1", collection=self.FakeLegacyStorageCollection(), ), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace(get_user_case=lambda username, case_key: None), ) structured_model = HistoriaClinicaStructured.model_validate( { "patient_name": "Paciente extraído", "resumen_clinico": "SEGUROS COMERCIALES BOLIVAR S.A.", "diagnosticos": [], "procedimientos": [], "medicamentos": [], } ) historia_request = SimpleNamespace( extraer_historia_estructurada=lambda raw_text, **kwargs: SimpleNamespace( analysis_structured=structured_model.model_dump( by_alias=True, exclude_none=True, exclude_defaults=True ), analysis_model=structured_model, rendered_html="

HTML previo incorrecto

", ) ) with patch( "app.services.clinical_document_service._load_historia_clinica_request_class", return_value=historia_request, ): result = service.process_and_persist( ClinicalDocumentRequest( raw_text=( "EPS: SEGUROS COMERCIALES BOLIVAR S.A.\n" "Resumen\n" "Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.\n" ), detected_type="historia_clinica", username="tester", original_name="historia.pdf", ) ) self.assertEqual( result["analysis_structured"]["rs"], "Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.", ) inserted = collection.inserted[0] self.assertEqual( inserted["analysis_structured"]["rs"], "Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.", ) def test_repair_existing_documents_normalizes_historia_structured_without_touching_case_fields( self, ) -> None: repair_collection = self.FakeRepairCollection( [ { "_id": "analysis-1", "usuario": "tester", "tipo_documento": "historia_clinica", "nombre_archivo": "historia.pdf", "nombre_paciente": "JULIO CESAR ACOSTA GUEVARA", "descripcion": "Historia clínica previa", "case_key": "93387170-178474-julio-cesar-acosta-guevara", "case_number": "178474", "patient_id": "93387170", "case_resolution_status": "confirmed", "case_resolution_evidence": ["deterministic_case_number"], "review_required": False, "review_messages": [], "legacy_historia_id": "legacy-1", "analysis_structured": { "sv": "v1", "dt": "historia_clinica", "pn": "Julio César Acosta Guevara", "nc": "93387170", "rs": "Resumen previo", "dx": [], "pr": [], "md": [], }, } ] ) service = ClinicalDocumentService( mongo_storage=SimpleNamespace(collection=self.FakeLegacyStorageCollection()), mongo_analyses=SimpleNamespace(collection=repair_collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) def fake_resolve_case_identity( _self: ClinicalDocumentService, _request: ClinicalDocumentRequest, ) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="93387170-178474-julio-cesar-acosta-guevara", case_number="178474", patient_id="93387170", patient_name="JULIO CESAR ACOSTA GUEVARA", case_resolution_status="confirmed", case_resolution_evidence=["deterministic_case_number"], review_required=False, review_messages=[], ) service_for_test = cast(Any, service) service_for_test.resolve_case_identity = MethodType(fake_resolve_case_identity, service) def fake_resolve_patient_name_from_sources( _self: ClinicalDocumentService, *, provided_patient_name: str, case_key: str, username: str, analisis_html: str, analysis_structured: dict[str, object] | None, raw_text: str, deterministic_signals: object | None = None, ) -> str: return "JULIO CESAR ACOSTA GUEVARA" service_for_test._resolve_patient_name_from_sources = MethodType( fake_resolve_patient_name_from_sources, service, ) result = service.repair_existing_documents_for_user("tester") self.assertEqual(result["repaired_case_metadata"], 0) self.assertEqual(result["repaired_patient_names"], 0) self.assertEqual(len(repair_collection.update_calls), 1) update_query, update_payload = repair_collection.update_calls[0] self.assertEqual(update_query, {"_id": "analysis-1"}) self.assertEqual(update_payload["$set"]["analysis_structured"]["nc"], "178474") self.assertEqual(update_payload["$set"]["analysis_structured"]["ip"], "93387170") self.assertEqual(update_payload["$set"]["analysis_structured"]["pn"], "JULIO CESAR ACOSTA GUEVARA") self.assertNotIn("case_number", update_payload["$set"]) self.assertNotIn("patient_id", update_payload["$set"]) def test_repair_existing_documents_preserves_manual_case_key(self) -> None: repair_collection = self.FakeRepairCollection( [ { "_id": "analysis-1", "usuario": "tester", "tipo_documento": "factura", "nombre_archivo": "factura.pdf", "nombre_paciente": "Paciente Manual", "descripcion": "Factura manual", "case_key": "CASE-MANUAL-1", "case_number": "HC-001", "patient_id": "111222333", "ingestion_source": "manual", "case_resolution_status": "confirmed", "case_resolution_evidence": ["provided_case_key"], "review_required": False, "review_messages": [], } ] ) service = ClinicalDocumentService( mongo_storage=SimpleNamespace(collection=self.FakeLegacyStorageCollection()), mongo_analyses=SimpleNamespace(collection=repair_collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) def fake_resolve_case_identity( _self: ClinicalDocumentService, _request: ClinicalDocumentRequest, ) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="CASE-RECOMPUTED", case_number="HC-001", patient_id="111222333", patient_name="Paciente Manual", case_resolution_status="confirmed", case_resolution_evidence=["recomputed"], review_required=False, review_messages=[], ) service_for_test = cast(Any, service) service_for_test.resolve_case_identity = MethodType(fake_resolve_case_identity, service) def fake_resolve_patient_name_from_sources( _self: ClinicalDocumentService, *, provided_patient_name: str, case_key: str, username: str, analisis_html: str, analysis_structured: dict[str, object] | None, raw_text: str, deterministic_signals: object | None = None, ) -> str: return "Paciente Manual" service_for_test._resolve_patient_name_from_sources = MethodType( fake_resolve_patient_name_from_sources, service, ) result = service.repair_existing_documents_for_user("tester") self.assertEqual(result["repaired_case_metadata"], 1) update_query, update_payload = repair_collection.update_calls[0] self.assertEqual(update_query, {"_id": "analysis-1"}) self.assertNotIn("case_key", update_payload["$set"]) self.assertEqual(repair_collection.items[0]["case_key"], "CASE-MANUAL-1") def test_get_user_case_context_merges_runtime_and_document_identity(self) -> None: collection = self.FakeCollection() collection.find_one_result = { "_id": "analysis-2", "case_key": "CASE-MANUAL-1", "case_number": "HC-001", "patient_id": "111222333", "nombre_paciente": "Paciente Uno", } service = ClinicalDocumentService( mongo_storage=SimpleNamespace(collection=self.FakeLegacyStorageCollection()), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace( get_user_case=lambda username, case_key: { "case_key": case_key, "epicrisis_status": "pendiente", "case_number": "", "patient_id": "", "patient_name": "", } ), ) context = service.get_user_case_context("tester", "CASE-MANUAL-1") self.assertEqual(context["case_key"], "CASE-MANUAL-1") self.assertEqual(context["case_number"], "HC-001") self.assertEqual(context["patient_id"], "111222333") self.assertEqual(context["patient_name"], "Paciente Uno") self.assertEqual(context["lookup_source"], "batch_case_repository+historias_analizadas") self.assertEqual(context["document_id"], "analysis-2") def test_process_and_persist_manual_historia_then_factura_keeps_same_case_key(self) -> None: collection = self.FakeCollection() runtime_cases: dict[str, dict[str, object]] = {} service = ClinicalDocumentService( mongo_storage=SimpleNamespace(guardar_analisis=lambda **kwargs: "legacy-1"), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=SimpleNamespace(), ) def get_user_case(_username: str, case_key: str): current = runtime_cases.get(case_key) return dict(current) if current else None def upsert_user_case(_username: str, case_key: str, payload: dict[str, object]): current = dict(runtime_cases.get(case_key) or {}) current.update(payload) current["case_key"] = case_key runtime_cases[case_key] = current return dict(current) service.batch_case_repository = cast( MongoBatchCaseRepository, SimpleNamespace( get_user_case=get_user_case, upsert_user_case=upsert_user_case, ), ) def fake_resolve_case_identity( _self: ClinicalDocumentService, request: ClinicalDocumentRequest, ) -> CaseIdentityResolution: return CaseIdentityResolution( case_key=str(request.case_key or "CASE-MANUAL-1"), case_number="HC-001", patient_id="111222333", patient_name="Paciente Uno", case_resolution_status="confirmed", case_resolution_evidence=["provided_case_key"], review_required=False, review_messages=[], ) service_for_test = cast(Any, service) service_for_test.resolve_case_identity = MethodType(fake_resolve_case_identity, service) def fake_generate_analysis( _self: ClinicalDocumentService, raw_text: str, detected_type: str, *, username: str = "", ) -> SimpleNamespace: return SimpleNamespace( analysis_structured={}, analysis_schema=None, analysis_schema_version=None, analysis_render_version=None, rendered_html=f"

{detected_type}

", error_analisis=None, ) def fake_resolve_patient_name_from_sources( _self: ClinicalDocumentService, *, provided_patient_name: str, case_key: str, username: str, analisis_html: str, analysis_structured: dict[str, object] | None, raw_text: str, deterministic_signals: object | None = None, ) -> str: return str(provided_patient_name or "").strip() or "Paciente Uno" def fake_enrich_historia( _self: ClinicalDocumentService, payload: dict[str, object], *, deterministic_signals: object | None = None, ) -> None: return None def fake_enrich_factura( _self: ClinicalDocumentService, payload: dict[str, object], ) -> None: payload.setdefault("factura_json", {}) def fake_normalize_factura_structured_payload( _self: ClinicalDocumentService, payload: dict[str, object], ) -> None: return None def fake_normalize_historia_structured_payload( _self: ClinicalDocumentService, payload: dict[str, object], ) -> None: return None def fake_ensure_legacy_history( _self: ClinicalDocumentService, payload: dict[str, object], ) -> str: return "legacy-1" service_for_test._generate_analysis = MethodType(fake_generate_analysis, service) service_for_test._resolve_patient_name_from_sources = MethodType( fake_resolve_patient_name_from_sources, service, ) service_for_test._enrich_historia = MethodType(fake_enrich_historia, service) service_for_test._enrich_factura = MethodType(fake_enrich_factura, service) service_for_test._normalize_factura_structured_payload = MethodType( fake_normalize_factura_structured_payload, service, ) service_for_test._normalize_historia_structured_payload = MethodType( fake_normalize_historia_structured_payload, service, ) service_for_test._ensure_legacy_history = MethodType(fake_ensure_legacy_history, service) historia = service.process_and_persist( ClinicalDocumentRequest( raw_text="HC manual", detected_type="historia_clinica", username="tester", original_name="historia.pdf", case_key="CASE-MANUAL-1", case_number="HC-001", patient_id="111222333", provided_patient_name="Paciente Uno", ingestion_source="individual", batch_file_id="individual:1", ) ) runtime_cases["CASE-MANUAL-1"].update( { "ready_for_epicrisis": False, "epicrisis_status": "fallido", "epicrisis_rule_status": "blocked", "epicrisis_rule_findings": [{"rule_code": "EPICRISIS-HC-001"}], "epicrisis_blocking_reason": "No se puede generar la epicrisis sin historia clínica.", "epicrisis_missing_documents": ["historia_clinica"], "epicrisis_last_rule_evaluation_at": "2026-06-22T10:00:00+00:00", } ) factura = service.process_and_persist( ClinicalDocumentRequest( raw_text="Factura manual", detected_type="factura", username="tester", original_name="factura.pdf", case_key="CASE-MANUAL-1", ingestion_source="individual", batch_file_id="individual:2", ) ) self.assertEqual(historia["case_key"], "CASE-MANUAL-1") self.assertEqual(factura["case_key"], "CASE-MANUAL-1") self.assertEqual(factura["case_number"], "HC-001") self.assertEqual(factura["patient_id"], "111222333") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["case_key"], "CASE-MANUAL-1") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["case_number"], "HC-001") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["patient_id"], "111222333") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["patient_name"], "Paciente Uno") self.assertTrue(runtime_cases["CASE-MANUAL-1"]["ready_for_epicrisis"]) self.assertEqual(runtime_cases["CASE-MANUAL-1"]["epicrisis_status"], "pendiente") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["epicrisis_rule_status"], "") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["epicrisis_rule_findings"], []) self.assertEqual(runtime_cases["CASE-MANUAL-1"]["epicrisis_blocking_reason"], "") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["epicrisis_missing_documents"], []) self.assertEqual(runtime_cases["CASE-MANUAL-1"]["epicrisis_last_rule_evaluation_at"], "") self.assertEqual(runtime_cases["CASE-MANUAL-1"]["runtime_origin"], "manual") self.assertEqual( collection.deleted[-1], { "usuario": "tester", "tipo_documento": "epicrisis_case_cache", "case_key": "CASE-MANUAL-1", }, )