from __future__ import annotations import unittest from copy import deepcopy from datetime import UTC from pathlib import Path from types import SimpleNamespace from typing import cast from unittest.mock import MagicMock, patch from app.batch_processing.infrastructure.mongo_repositories import MongoBatchCaseRepository from app.llm import LLMErrorKind, LLMProviderError from app.llm.renderers.clinical_documents import render_factura_html from app.llm.schemas import FacturaStructured from app.services.clinical_document_service import ( CaseIdentityResolution, ClinicalDocumentRequest, ClinicalDocumentService, ) from app.services.clinical_structured_extraction import ClinicalStructuredExtractionService from app.services.factura_deterministic import ( _classify_service_section, build_factura_structured, parse_factura_text, ) from app.soat_crosswalk.infrastructure.catalogs import XlsxCupsCatalog from modules.data_collection.read_pdf import ReadPdf JULIO_FACTURA_TEXT = Path("test/fixtures/factura_julio_cesar_acosta.txt").read_text(encoding="utf-8") JULIO_FACTURA_DIRTY_TEXT = Path("test/fixtures/factura_julio_cesar_acosta_dirty.txt").read_text(encoding="utf-8") BLANCA_FACTURA_TEXT = """ Inversiones Médicas Valle Salud S.A.S Factura Electrónica de Venta Prefijo No. Factura 71 57816 Número de Autorización de Facturación Electrónica: 18760000000000 NIT: 900.631.361-6 CALI (SANTIAGO DE CALI)-VALLE DEL CAUCA, 15 abril, 2024 10:09 a. m. Caso No. 176654 Señores: 860002503 - SEGUROS COMERCIALES BOLIVAR S.A Dirección: CRA 1 # 1 - 1 Convenio: FE0866 DIFERENCIAL SOAT - SEGUROS COMERCIALES BOLIVAR 2024 Fecha Ingreso: 17/03/2024 06:24 Paciente: BLANCA LIJIA RENGIFO AGUIRRE Fecha Egreso: 18/03/2024 10:54 CC: 66781911 Por: Urgencias No.Autorizacion: AUTH-7788 Tipo de servicio y/o mercancia 17-03-2024 PROC001 902210 Laparotomia exploratoria 1 $1,000,000.00 $1,000,000.00 17-03-2024 LAB001 Hemograma automatizado 1 $18,000.00 $18,000.00 17-03-2024 IMG001 760310 Fluoroscopia arco en c 1 $42,000.00 $42,000.00 17-03-2024 106002 Habitacion bipersonal 2 $120,000.00 $240,000.00 17-03-2024 HON001 Honorarios cirujano 1 $95,000.00 $95,000.00 17-03-2024 MED001 Acetaminofen 500 mg tabletas 20 $500.00 $10,000.00 17-03-2024 MED002 Dipirona solucion inyectable 1 g/2 ml 2 $17,150.00 $34,300.00 17-03-2024 MAT001 Jeringa descartable 10 ml 2 $2,500.00 $5,000.00 17-03-2024 SAL001 Derechos de sala de observacion 1 $150,000.00 $150,000.00 Vencimiento: 30 Dias a Partir de la Fecha de Radicacion Son: TRES MILLONES TRESCIENTOS SESENTA Y DOS MIL TRESCIENTOS SETENTA Y NUEVE EMISOR: Total de los Servicios: $3,362,379.00 Total Descuento: $0.00 Copagos - Pagos: $0.00 Total de la Factura: $3,362,379.00 """ JULIO_FACTURA_PYPDF2_FLATTENED_TEXT = """ 59182Factura Electrónica de Venta CONTRUBUYENTES DE RENTA. NO RESPONSABLES DE IVAInversiones Médicas Valle Salud S.A.S NIT: 900.631.361-6AV 4N # 14 - 12 Caso No. 178474Número de autorización de facturación electrónica 18764055185066 aprobada Fecha Factura: CALI (SANTIAGO DE CALI)-VALLE DEL CAUCA, 27 mayo, 2024 11:30 a.m.71Prefijo No. Factura Señores: 860524654 - ASEGURADORA SOLIDARIA DE COLOMBIA ENTIDAD COOPERATIVA Dirección: CARRERA 47 NO 72 144 Convenio: Fecha Ingreso: 07/05/2024 09:38 FE0858 DIFERENCIAL SOAT - ASEGURADORA SOLIDARIA DE COLOMBIA 2024 Paciente: Fecha Egreso: 10/05/2024 11:32 JULIO CESAR ACOSTA GUEVARA CC: 93387170 No.Autorizacion: Por: ASEGURADORA SOLIDARIA DE COLOMBIA ENTIDAD COOPERATIVA Total Precio Cant. Fecha Tipo de Servicio y/o Mercancia Cod. Cups-Cum 09-05-2024 13170 793101 Osteosíntesis en clavícula 1.00 $0.00 $0.00 09-05-2024 14332 817204 Ligamentorrafia o reinserción ligamentos (una a dos) 1.00 $0.00 $0.00 09-05-2024 18211 765401 Extracción de cuerpos libres intraarticulares en hombro, codo, rodilla, tobillo o articulación temporomandibular 1.00 $0.00 $0.00 09-05-2024 18215 818307 Condroplastia de hombro o rodilla 1.00 $0.00 $0.00 08-05-2024 19290 903823 Creatinina en suero, orina y otros 1.00 $19,800.00 $19,800.00 08-05-2024 19304 902210 Cuadro hemático o hemograma hematocrito y leucograma 1.00 $34,300.00 $34,300.00 09-05-2024 21602 873501 Portátiles con fluoroscopia y/o intensificador de imagen (practicado en quirófanos); al valor del estudio, agregar: 1.00 $209,900.00 $209,900.00 08-05-2024 38122 129M02 Habitación bipersonal Segundo Nivel 1.00 $346,900.00 $346,900.00 09-05-2024 39005 Grupo 07 Honorarios Cirujano o Ginecoobstetra 1.00 $185,700.00 $185,700.00 10-05-2024 5496 44405-3 CEFALEXINA 500MG CAPSULA 20.00 $1,400.00 $28,000.00 07-05-2024 5926 JERINGA 10ML 3.00 $500.00 $1,500.00 Total de los Servicios: $8,085,212.00Vencimiento: 30 Dias a Partir de la Fecha de Radicacion Total Descuento: - $0.00 Copagos - Pagos: $0.00 Son: OCHO MILLONES OCHENTA Y CINCO MIL DOSCIENTOS DOCE $8,085,212.00 Total de la Factura: """ class _FakeInsertResult: def __init__(self, inserted_id: str) -> None: self.inserted_id = inserted_id class _FakeCollection: def __init__(self) -> None: self.inserted: list[dict] = [] def find_one(self, query: dict, sort=None): # noqa: ANN001 return None def insert_one(self, document: dict) -> _FakeInsertResult: self.inserted.append(deepcopy(document)) return _FakeInsertResult("analysis-1") def update_one(self, query: dict, update: dict) -> None: # noqa: ANN001 return None class _FakeBatchCaseRepository(MongoBatchCaseRepository): def __init__(self) -> None: self.upserts: list[tuple[str, str, dict]] = [] def upsert_user_case(self, username: str, case_key: str, payload: dict) -> dict: self.upserts.append((username, case_key, deepcopy(payload))) return {"usuario": username, "case_key": case_key, **payload} class _FailingStructuredExtractionService: def extract(self, **kwargs): # noqa: ANN003 return FacturaStructured.model_validate({"dt": "factura", "pv": "bad"}) class _FailingGeminiFacturaExtractionService: def extract(self, **kwargs): # noqa: ANN003 raise LLMProviderError( "rate limit provider raw detail", kind=LLMErrorKind.RATE_LIMITED, provider="gemini", model="gemini-2.5-flash", retryable=True, ) class ReadPdfExtractionTest(unittest.TestCase): def test_read_pdf_prefers_pdfplumber_layout_and_keeps_page_boundaries(self) -> None: first_page = MagicMock() first_page.extract_text.return_value = "Factura Electrónica de Venta\n09-05-2024 13170 793101 Proc 1 $0 $0" second_page = MagicMock() second_page.extract_text.return_value = "Total de la Factura: $1,000.00" pdf = MagicMock() pdf.__enter__.return_value.pages = [first_page, second_page] with patch("modules.data_collection.read_pdf.pdfplumber.open", return_value=pdf): text = ReadPdf.read(b"%PDF fixture") self.assertIn("\n\f\n", text) self.assertIn("Factura Electrónica de Venta", text) first_page.extract_text.assert_called_once_with(layout=True, x_tolerance=1, y_tolerance=3) class FacturaDeterministicParserTest(unittest.TestCase): def test_official_cups_chapters_classify_surgical_and_non_surgical_codes(self) -> None: catalog = XlsxCupsCatalog() for code in ("765401", "793101", "804001", "808112", "817204", "818307"): self.assertEqual(catalog.chapter(code), 13) self.assertEqual(catalog.classification(code), "quirurgico") self.assertEqual( _classify_service_section( "Procedimiento clínico", codigo_facturacion="interno", codigo_referencia=code, ), "procedimientos_quirurgicos", ) for code in ("890202", "890602", "890701", "931001"): self.assertEqual(catalog.classification(code), "no_quirurgico") self.assertEqual( _classify_service_section( "Atención clínica", codigo_facturacion="interno", codigo_referencia=code, ), "procedimientos_no_quirurgicos", ) def test_diagnostic_aids_take_precedence_over_non_surgical_cups(self) -> None: self.assertEqual( _classify_service_section( "Hemograma automatizado", codigo_facturacion="interno", codigo_referencia="890202", ), "examenes_laboratorio", ) self.assertEqual( _classify_service_section( "Radiografía de hombro", codigo_facturacion="interno", codigo_referencia="890202", ), "imagenologia", ) def test_legacy_invoice_schema_defaults_non_surgical_procedures_to_empty(self) -> None: model = FacturaStructured.model_validate({"document_type": "factura"}) self.assertEqual(model.servicios_procedimientos.procedimientos_no_quirurgicos, []) def test_parse_factura_text_handles_pypdf2_flattened_invoice_text(self) -> None: factura = parse_factura_text(JULIO_FACTURA_PYPDF2_FLATTENED_TEXT) self.assertEqual(factura["nombre_paciente"], "JULIO CESAR ACOSTA GUEVARA") self.assertEqual(factura["proveedor"]["nombre_institucion"], "Inversiones Médicas Valle Salud S.A.S") self.assertEqual(factura["proveedor"]["nit"], "900.631.361-6") self.assertEqual(factura["proveedor"]["direccion"], "AV 4N # 14 - 12") self.assertEqual(factura["informacion_factura"]["prefijo"], "71") self.assertEqual(factura["informacion_factura"]["numero_factura"], "59182") self.assertEqual(factura["informacion_factura"]["numero_caso"], "178474") self.assertEqual(factura["informacion_paciente"]["numero_identificacion"], "93387170") self.assertEqual(factura["informacion_paciente"]["fecha_ingreso"], "07/05/2024 09:38") self.assertEqual(factura["informacion_paciente"]["fecha_egreso"], "10/05/2024 11:32") self.assertEqual(factura["analisis_financiero"]["total_servicios"], "$8,085,212.00") self.assertEqual(factura["analisis_financiero"]["valor_total_factura"], "$8,085,212.00") self.assertGreaterEqual(len(factura["lineas_canonicas"]), 10) self.assertNotIn("line_items", factura["low_confidence_sections"]) self.assertTrue( any( item["descripcion"] == "Extracción de cuerpos libres intraarticulares en hombro, codo, rodilla, tobillo o articulación temporomandibular" for item in factura["lineas_canonicas"] ) ) self.assertTrue( any( item["medicamento"] == "CEFALEXINA 500MG CAPSULA" for item in factura["servicios_procedimientos"]["medicamentos"] ) ) def test_parse_factura_text_recovers_blanca_header_fields(self) -> None: factura = parse_factura_text(BLANCA_FACTURA_TEXT) self.assertEqual(factura["proveedor"]["nombre_institucion"], "Inversiones Médicas Valle Salud S.A.S") self.assertEqual(factura["proveedor"]["nit"], "900.631.361-6") self.assertEqual(factura["informacion_factura"]["prefijo"], "71") self.assertEqual(factura["informacion_factura"]["numero_factura"], "57816") self.assertEqual(factura["informacion_factura"]["numero_caso"], "176654") self.assertEqual( factura["informacion_factura"]["fecha_vencimiento"], "30 Dias a Partir de la Fecha de Radicacion", ) self.assertEqual( factura["pagador"]["tipo_convenio"], "FE0866 DIFERENCIAL SOAT - SEGUROS COMERCIALES BOLIVAR 2024", ) self.assertEqual(factura["informacion_paciente"]["fecha_ingreso"], "17/03/2024 06:24") self.assertEqual(factura["informacion_paciente"]["fecha_egreso"], "18/03/2024 10:54") self.assertEqual( factura["analisis_financiero"]["valor_en_letras"], "TRES MILLONES TRESCIENTOS SESENTA Y DOS MIL TRESCIENTOS SETENTA Y NUEVE", ) self.assertEqual(factura["analisis_financiero"]["valor_total_factura"], "$3,362,379.00") def test_parse_factura_text_classifies_services_without_losing_medication_names(self) -> None: servicios = parse_factura_text(BLANCA_FACTURA_TEXT)["servicios_procedimientos"] self.assertEqual( servicios["procedimientos_no_quirurgicos"][0]["descripcion"], "Laparotomia exploratoria", ) self.assertEqual(servicios["examenes_laboratorio"][0]["descripcion"], "Hemograma automatizado") self.assertEqual(servicios["imagenologia"][0]["descripcion"], "Fluoroscopia arco en c") self.assertEqual(servicios["hospitalizacion"][0]["habitacion"], "Habitacion bipersonal") self.assertEqual(servicios["honorarios_medicos"][0]["profesional"], "Honorarios cirujano") self.assertEqual( [item["medicamento"] for item in servicios["medicamentos"]], ["Acetaminofen 500 mg tabletas", "Dipirona solucion inyectable 1 g/2 ml"], ) self.assertEqual( [item["descripcion"] for item in servicios["otros_servicios"]], ["Jeringa descartable 10 ml", "Derechos de sala de observacion"], ) def test_parse_factura_text_handles_multi_page_invoice_with_canonical_lines(self) -> None: factura = parse_factura_text(JULIO_FACTURA_TEXT) self.assertEqual(factura["nombre_paciente"], "JULIO CESAR ACOSTA GUEVARA") self.assertEqual( factura["pagador"]["aseguradora_eps"], "ASEGURADORA SOLIDARIA DE COLOMBIA ENTIDAD COOPERATIVA", ) self.assertEqual(factura["pagador"]["nit_pagador"], "860524654") self.assertEqual(factura["proveedor"]["direccion"], "AV 4N # 14 - 12") self.assertEqual( factura["informacion_factura"]["fecha_vencimiento"], "30 Dias a Partir de la Fecha de Radicacion", ) self.assertEqual(factura["analisis_financiero"]["total_servicios"], "$8,085,212.00") self.assertEqual(factura["analisis_financiero"]["valor_total_factura"], "$8,085,212.00") self.assertEqual( factura["analisis_financiero"]["valor_en_letras"], "OCHO MILLONES OCHENTA Y CINCO MIL DOSCIENTOS DOCE", ) self.assertTrue(factura["lineas_canonicas"]) self.assertEqual( factura["lineas_canonicas"][2]["descripcion"], "Extracción de cuerpos libres intraarticulares en hombro, codo, rodilla, tobillo o articulación temporomandibular", ) self.assertEqual( factura["lineas_canonicas"][3]["descripcion"], "Condroplastia de hombro o rodilla", ) self.assertEqual( sum(1 for item in factura["lineas_canonicas"] if item["descripcion"] == "JERINGA 10ML"), 2, ) self.assertTrue( any( item["medicamento"] == "CEFALEXINA 500MG CAPSULA" for item in factura["servicios_procedimientos"]["medicamentos"] ) ) self.assertTrue( any( item["habitacion"] == "Habitación bipersonal Segundo Nivel" for item in factura["servicios_procedimientos"]["hospitalizacion"] ) ) def test_parse_factura_text_sanitizes_dirty_invoice_identity_and_dates(self) -> None: factura = parse_factura_text(JULIO_FACTURA_DIRTY_TEXT) self.assertEqual(factura["nombre_paciente"], "JULIO CESAR ACOSTA GUEVARA") self.assertEqual( factura["informacion_factura"]["fecha_emision"], "27 mayo, 2024 11:30 a.m.", ) self.assertEqual(factura["informacion_paciente"]["fecha_ingreso"], "07/05/2024 09:38") self.assertEqual(factura["informacion_paciente"]["fecha_egreso"], "10/05/2024 11:32") self.assertEqual( factura["pagador"]["aseguradora_eps"], "ASEGURADORA SOLIDARIA DE COLOMBIA ENTIDAD COOPERATIVA", ) self.assertEqual(factura["pagador"]["numero_autorizacion"], "") self.assertEqual(factura["informacion_factura"]["fecha_vencimiento"], "30 Dias a Partir de la Fecha de Radicacion") self.assertEqual(len(factura["lineas_canonicas"]), 3) def test_build_factura_structured_overrides_bad_llm_fields_and_renders_scrollable_tables(self) -> None: broken_structured = FacturaStructured.model_validate( { "pn": "Paciente incorrecto", "pv": {"ni": "900.631.361-6", "nt": "S.A.S"}, "df": {"pf": "71", "nf": "7150001", "fv": "2024-09-01"}, "pc": {"nm": "Paciente incorrecto", "fg": "No especificado"}, } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) normalized_model, factura_json = build_factura_structured( raw_text=BLANCA_FACTURA_TEXT, current_structured=broken_structured, patient_name="BLANCA LIJIA RENGIFO AGUIRRE", patient_id="66781911", case_number="176654", ) rendered = render_factura_html(normalized_model) self.assertEqual( normalized_model.proveedor.nombre_institucion, "Inversiones Médicas Valle Salud S.A.S" ) self.assertEqual(normalized_model.proveedor.nit, "900.631.361-6") self.assertEqual(normalized_model.datos_factura.numero_factura, "57816") self.assertIsNone(normalized_model.datos_factura.fecha_vencimiento) self.assertEqual(normalized_model.paciente.fecha_egreso, "2024-03-18T10:54-05:00") self.assertIsNone(normalized_model.datos_factura.fecha_emision) self.assertEqual(factura_json["lineas_canonicas"][0]["fecha_servicio"], "2024-03-17") self.assertEqual( factura_json["servicios_procedimientos"]["otros_servicios"][0]["descripcion"], "Jeringa descartable 10 ml", ) self.assertTrue(factura_json["lineas_canonicas"]) self.assertIn("Líneas facturadas canónicas", rendered) self.assertIn("ui-table-cell--wrap", rendered) self.assertIn('class="ui-table-wrap ui-table-wrap--dense-wide"', rendered) self.assertIn('data-local-date="true"', rendered) self.assertIn('datetime="2024-03-18T10:54-05:00"', rendered) self.assertIn("Otros servicios", rendered) def test_build_factura_structured_prefers_valid_llm_fields_before_deterministic_fallback(self) -> None: current_structured = FacturaStructured.model_validate( { "pn": "JULIO CESAR ACOSTA GUEVARA", "pv": { "ni": "Inversiones Médicas Valle Salud S.A.S", "nt": "900.631.361-6", "dr": "AV 4N # 14 - 12", }, "df": {"pf": "71", "nf": "57816", "fe": "27 mayo, 2024 11:30 a.m."}, "pg": { "ae": "ASEGURADORA SOLIDARIA DE COLOMBIA ENTIDAD COOPERATIVA", "np": "860524654", }, "pc": {"nm": "JULIO CESAR ACOSTA GUEVARA", "id": "93387170"}, } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) normalized_model, factura_json = build_factura_structured( raw_text=JULIO_FACTURA_DIRTY_TEXT, current_structured=current_structured, patient_name="JULIO CESAR ACOSTA GUEVARA", patient_id="93387170", case_number="178474", ) self.assertEqual(normalized_model.patient_name, "JULIO CESAR ACOSTA GUEVARA") self.assertEqual(normalized_model.proveedor.nombre_institucion, "Inversiones Médicas Valle Salud S.A.S") self.assertEqual( normalized_model.pagador.aseguradora_eps, "ASEGURADORA SOLIDARIA DE COLOMBIA ENTIDAD COOPERATIVA", ) self.assertEqual(normalized_model.datos_factura.numero_factura, "57816") self.assertEqual(normalized_model.datos_factura.fecha_emision, "2024-05-27T11:30-05:00") self.assertEqual(normalized_model.datos_factura.fecha_vencimiento, "2024-06-26") self.assertEqual(factura_json["nombre_paciente"], "JULIO CESAR ACOSTA GUEVARA") def test_render_factura_html_uses_empty_state_instead_of_synthetic_no_especificado_rows(self) -> None: rendered = render_factura_html( FacturaStructured.model_validate( { "dt": "factura", "pn": "JULIO CESAR ACOSTA GUEVARA", "pv": {"ni": "Clinica Demo"}, } ) ) self.assertIn("No se pudieron estructurar líneas facturadas.", rendered) self.assertIn("Sin medicamentos detectados.", rendered) self.assertNotIn("No se pudieron estructurar líneas facturadas.", rendered) def test_factura_schema_forces_document_type_and_normalizes_invoice_dates(self) -> None: validated = FacturaStructured.model_validate( { "dt": "27 mayo, 2024", "df": {"fe": "27 mayo, 2024 11:30 a.m.", "fv": "30 Dias a Partir de la Fecha de Radicacion"}, "pc": {"fi": "07/05/2024 09:38", "fg": "10/05/2024 11:32"}, "lc": [{"fs": "07-05-2024", "d": "Servicio demo"}], } ) self.assertEqual(validated.document_type, "factura") self.assertEqual(validated.datos_factura.fecha_emision, "2024-05-27T11:30-05:00") self.assertEqual(validated.datos_factura.fecha_vencimiento, "2024-06-26") self.assertEqual(validated.paciente.fecha_ingreso, "2024-05-07T09:38-05:00") self.assertEqual(validated.paciente.fecha_egreso, "2024-05-10T11:32-05:00") self.assertEqual(validated.lineas_canonicas[0].fecha_servicio, "2024-05-07") class FacturaPipelineNormalizationTest(unittest.TestCase): def test_process_and_persist_populates_flattened_factura_without_llm_success(self) -> None: collection = _FakeCollection() class _FacturaClinicalDocumentService(ClinicalDocumentService): def resolve_case_identity(self, 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=["manual_case"], review_required=False, review_messages=[], ) service = _FacturaClinicalDocumentService( 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=None, ) service.batch_case_repository = _FakeBatchCaseRepository() service.structured_extraction_service = cast( ClinicalStructuredExtractionService, _FailingStructuredExtractionService(), ) result = service.process_and_persist( ClinicalDocumentRequest( raw_text=JULIO_FACTURA_PYPDF2_FLATTENED_TEXT, detected_type="factura", username="tester", original_name="FACTURA JULIO CESAR ACOSTA GUEVARA.pdf", ) ) self.assertEqual(result["analysis_structured"]["df"]["nf"], "59182") self.assertEqual(result["factura_json"]["informacion_factura"]["numero_factura"], "59182") self.assertGreaterEqual(len(result["factura_json"]["lineas_canonicas"]), 10) self.assertIn("validacion_financiera", result["factura_json"]) self.assertIn("validacion_financiera", collection.inserted[0]["factura_json"]) self.assertIn("Líneas facturadas canónicas", result["analisis_html"]) self.assertNotIn("No se pudieron estructurar líneas facturadas.", result["analisis_html"]) def test_process_and_persist_reuses_historical_invoice_without_financial_mutation(self) -> None: historical = { "_id": "historical-analysis-1", "usuario": "tester", "tipo_documento": "factura", "effective_document_type": "factura", "batch_file_id": "historical-file-1", "ingestion_source": "batch", "legacy_historia_id": "legacy-history-1", "factura_json": { "lineas_canonicas": [ { "cantidad": "2.00", "valor_unitario": "$185,700.00", "total": "$371.40", } ] }, } class _ExistingCollection(_FakeCollection): def find_one(self, query: dict, sort=None): # noqa: ANN001 return deepcopy(historical) def update_one(self, query: dict, update: dict) -> None: # noqa: ANN001 raise AssertionError("La reutilización no debe actualizar el documento histórico") class _FacturaClinicalDocumentService(ClinicalDocumentService): def resolve_case_identity(self, request: ClinicalDocumentRequest) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="historical-case", case_resolution_status="confirmed", review_required=False, ) collection = _ExistingCollection() service = _FacturaClinicalDocumentService( mongo_storage=SimpleNamespace(), mongo_analyses=SimpleNamespace(collection=collection), client_groq=None, client_gemini=None, cie10_retriever=None, cups_retriever=None, colombia_tz=UTC, llm_router=None, ) result = service.process_and_persist( ClinicalDocumentRequest( raw_text="Factura histórica", detected_type="factura", username="tester", original_name="FACTURA HISTORICA.pdf", batch_file_id="historical-file-1", ingestion_source="batch", ) ) self.assertTrue(result["reused"]) self.assertNotIn("validacion_financiera", result["factura_json"]) self.assertEqual(result["factura_json"], historical["factura_json"]) self.assertEqual(collection.inserted, []) def test_process_and_persist_rewrites_factura_structured_and_factura_json(self) -> None: collection = _FakeCollection() broken_factura = FacturaStructured.model_validate( { "pn": "Paciente incorrecto", "pv": {"ni": "900.631.361-6", "nt": "S.A.S"}, "df": {"pf": "71", "nf": "7150001", "fv": "2024-09-01"}, "pg": {"ae": "SEGUROS COMERCIALES BOLIVAR S.A"}, "pc": {"nm": "Paciente incorrecto", "fg": "No especificado"}, } ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) llm_router = SimpleNamespace( generate_structured=lambda request: SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content=broken_factura, ) ) class _FacturaClinicalDocumentService(ClinicalDocumentService): def resolve_case_identity(self, request: ClinicalDocumentRequest) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="66781911-176654-blanca-lijia-rengifo-aguirre", case_number="176654", patient_id="66781911", patient_name="BLANCA LIJIA RENGIFO AGUIRRE", case_resolution_status="confirmed", case_resolution_evidence=["deterministic_case_number"], review_required=False, review_messages=[], ) service = _FacturaClinicalDocumentService( 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 = _FakeBatchCaseRepository() result = service.process_and_persist( ClinicalDocumentRequest( raw_text=BLANCA_FACTURA_TEXT, detected_type="factura", username="tester", original_name="FACTURA BLANCA LIJIA RENGIFO AGUIRRE.pdf", ) ) self.assertEqual(result["analysis_structured"]["pv"]["ni"], "Inversiones Médicas Valle Salud S.A.S") self.assertEqual(result["analysis_structured"]["pv"]["nt"], "900.631.361-6") self.assertEqual(result["analysis_structured"]["df"]["nf"], "57816") self.assertEqual(result["analysis_structured"]["df"]["nc"], "176654") self.assertEqual( result["analysis_structured"]["pg"]["tc"], "FE0866 DIFERENCIAL SOAT - SEGUROS COMERCIALES BOLIVAR 2024", ) self.assertEqual(result["analysis_structured"]["pc"]["fg"], "2024-03-18T10:54-05:00") self.assertNotIn("fe", result["analysis_structured"]["df"]) self.assertNotIn("fv", result["analysis_structured"]["df"]) self.assertEqual( result["factura_json"]["servicios_procedimientos"]["medicamentos"][0]["medicamento"], "Acetaminofen 500 mg tabletas", ) self.assertEqual( result["factura_json"]["servicios_procedimientos"]["otros_servicios"][1]["descripcion"], "Derechos de sala de observacion", ) self.assertIn('class="ui-table-wrap ', result["analisis_html"]) inserted = collection.inserted[0] self.assertEqual(inserted["analysis_structured"]["df"]["nf"], "57816") self.assertEqual(inserted["factura_json"]["informacion_factura"]["numero_factura"], "57816") self.assertNotIn("fecha_emision", inserted["factura_json"]["informacion_factura"]) self.assertNotIn("fecha_vencimiento", inserted["factura_json"]["informacion_factura"]) def test_process_and_persist_uses_targeted_repair_only_for_low_confidence_invoice_sections(self) -> None: collection = _FakeCollection() low_confidence_text = """ Clinica Central del Valle Factura Electrónica de Venta Prefijo No. Factura AA 9001 NIT: 900.111.222-3 Fecha Factura: CALI (SANTIAGO DE CALI)-VALLE DEL CAUCA, 15 abril, 2024 10:09 a.m. Señores: Caso No. 176654 Convenio: PLAN BASICO Fecha Ingreso: 17/03/2024 06:24 Paciente: COLOMBIA ENTIDAD COOPERATIVA Fecha Egreso: 18/03/2024 10:54 CC: 66781911 No.Autorizacion: COLOMBIA ENTIDAD COOPERATIVA Tipo de servicio y/o mercancia 17-03-2024 PROC001 902210 Laparotomia exploratoria uno $1,000,000.00 $1,000,000.00 Copagos - Pagos: Total de la Factura: $0.00 $1,000,000.00 """ broken_factura = {"dt": "factura", "pn": "Paciente incorrecto", "pg": {"ae": "Caso No. 176654"}} repair_calls: list[str] = [] def _generate_structured(request): model_name = request.output_model.__name__ if model_name == "_FacturaIdentityRepair": repair_calls.append("identity") return SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content={ "dt": "factura", "pn": "BLANCA LIJIA RENGIFO AGUIRRE", "pc": {"nm": "BLANCA LIJIA RENGIFO AGUIRRE", "id": "66781911"}, }, ) if model_name == "_FacturaHeaderRepair": repair_calls.append("header") return SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content={ "dt": "factura", "pv": {"dr": "CRA 1 # 1 - 1"}, "pg": {"ae": "SEGUROS COMERCIALES BOLIVAR S.A", "np": "860002503"}, }, ) if model_name == "_FacturaSummaryRepair": repair_calls.append("summary") return SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content={ "dt": "factura", "rf": {"ts": "$1,000,000.00", "cg": "$0.00", "vt": "$1,000,000.00"}, }, ) if model_name == "_FacturaLineItemsRepair": repair_calls.append("line_items") return SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content={ "dt": "factura", "lc": [ { "fs": "17-03-2024", "cf": "PROC001", "cr": "902210", "d": "Laparotomia exploratoria", "ct": "procedimientos_quirurgicos", "q": "1", "vu": "$1,000,000.00", "tt": "$1,000,000.00", } ], }, ) return SimpleNamespace( provider="gemini", model="gemini-2.5-flash-lite", metrics={}, content=broken_factura, ) class _FacturaClinicalDocumentService(ClinicalDocumentService): def resolve_case_identity(self, request: ClinicalDocumentRequest) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="66781911-176654-blanca-lijia-rengifo-aguirre", case_number="176654", patient_id="66781911", patient_name="BLANCA LIJIA RENGIFO AGUIRRE", case_resolution_status="confirmed", case_resolution_evidence=["manual_case"], review_required=False, review_messages=[], ) service = _FacturaClinicalDocumentService( 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(generate_structured=_generate_structured), ) service.batch_case_repository = _FakeBatchCaseRepository() result = service.process_and_persist( ClinicalDocumentRequest( raw_text=low_confidence_text, detected_type="factura", username="tester", original_name="FACTURA BAJA CONFIANZA.pdf", ) ) self.assertEqual(repair_calls, ["identity", "header", "line_items"]) self.assertEqual(result["analysis_structured"]["pg"]["ae"], "SEGUROS COMERCIALES BOLIVAR S.A") self.assertEqual(result["analysis_structured"]["pg"]["np"], "860002503") self.assertEqual(result["analysis_structured"]["pv"]["dr"], "CRA 1 # 1 - 1") self.assertEqual(result["analysis_structured"]["pn"], "BLANCA LIJIA RENGIFO AGUIRRE") self.assertEqual(result["analysis_structured"]["rf"]["ts"], "$1,000,000.00") self.assertEqual(result["analysis_structured"]["rf"]["vt"], "$1,000,000.00") self.assertTrue(result["factura_json"]["lineas_canonicas"]) self.assertIn( "Se preservo la identidad confirmada del caso sobre la lectura estructurada de la factura.", result["review_messages"], ) def test_process_and_persist_hides_raw_pydantic_factura_validation_error(self) -> None: collection = _FakeCollection() class _FacturaClinicalDocumentService(ClinicalDocumentService): def resolve_case_identity(self, request: ClinicalDocumentRequest) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="66781911-176654-blanca-lijia-rengifo-aguirre", case_number="176654", patient_id="66781911", patient_name="BLANCA LIJIA RENGIFO AGUIRRE", case_resolution_status="confirmed", case_resolution_evidence=["manual_case"], review_required=False, review_messages=[], ) service = _FacturaClinicalDocumentService( 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=None, ) service.batch_case_repository = _FakeBatchCaseRepository() service.structured_extraction_service = cast( ClinicalStructuredExtractionService, _FailingStructuredExtractionService(), ) result = service.process_and_persist( ClinicalDocumentRequest( raw_text=BLANCA_FACTURA_TEXT, detected_type="factura", username="tester", original_name="FACTURA ERROR VALIDACION.pdf", ) ) self.assertEqual(result["error_analisis"], "No fue posible validar la factura estructurada.") self.assertNotIn("literal_error", result["error_analisis"]) self.assertEqual(result["analysis_structured"]["dt"], "factura") def test_process_and_persist_hides_raw_gemini_factura_provider_error(self) -> None: collection = _FakeCollection() class _FacturaClinicalDocumentService(ClinicalDocumentService): def resolve_case_identity(self, request: ClinicalDocumentRequest) -> CaseIdentityResolution: return CaseIdentityResolution( case_key="66781911-176654-blanca-lijia-rengifo-aguirre", case_number="176654", patient_id="66781911", patient_name="BLANCA LIJIA RENGIFO AGUIRRE", case_resolution_status="confirmed", case_resolution_evidence=["manual_case"], review_required=False, review_messages=[], ) service = _FacturaClinicalDocumentService( 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=None, ) service.batch_case_repository = _FakeBatchCaseRepository() service.structured_extraction_service = cast( ClinicalStructuredExtractionService, _FailingGeminiFacturaExtractionService(), ) result = service.process_and_persist( ClinicalDocumentRequest( raw_text=BLANCA_FACTURA_TEXT, detected_type="factura", username="tester", original_name="FACTURA ERROR GEMINI.pdf", ) ) self.assertEqual(result["error_analisis"], "No fue posible validar la factura estructurada.") self.assertNotIn("rate limit", result["error_analisis"]) self.assertEqual(result["analysis_structured"]["dt"], "factura")