import asyncio from decimal import Decimal from io import BytesIO from types import SimpleNamespace from typing import Any, cast from fastapi import Request from openpyxl import load_workbook from app.batch_processing.infrastructure.batch_epicrisis_excel import ( BatchEpicrisisExcelCaseRow, BatchEpicrisisExcelWorkbookBuilder, ) from app.case_epicrisis.application.cost_ordering import ( COST_ORDER_VERSION, apply_factura_cost_ordering, attach_and_sort_costs, extract_invoice_cost_lines, match_billed_cost, match_billed_surgical_group, normalize_money, ) from app.case_epicrisis.application.utils import ( build_pdf_primary_ayudas_diagnosticas, build_pdf_primary_medicamentos, build_pdf_primary_procedimientos, build_procedimientos_clinicos, is_compatible_cached_context, ) from app.core.services import AppServices from app.models import UserInDB from app.routes.pages.epicrisis_pdf_cache import ( guardar_epicrisis_pdf_cache, imprimir_ultimo_pdf_epicrisis_paciente, ) def _factura(lines=None, services=None): return { "factura_json": { "lineas_canonicas": lines or [], "servicios_procedimientos": services or {}, } } def _valued_procedure( canonical_index: int | None, group: object, *, cups: str, rule: str = "cirugia_unica_tarifa_base", state: str = "liquidada", crosswalk_status: str | None = None, ) -> dict[str, Any]: valuation: dict[str, Any] = { "estado": state, "regla_aplicada": rule, "indice_linea_canonica": canonical_index, } if crosswalk_status: valuation["cruce_cups_soat"] = {"status": crosswalk_status} return { "codigo_cups": cups, "grupo_quirurgico_soat": group, "valoracion_soat": valuation, } def test_normalize_money_accepts_colombian_and_normalized_formats() -> None: assert normalize_money("$ 1.234.567,89") == Decimal("1234567.89") assert normalize_money("1,234,567.89") == Decimal("1234567.89") assert normalize_money("1234.50") == Decimal("1234.50") assert normalize_money("0") == Decimal("0") def test_normalize_money_rejects_invalid_and_negative_values() -> None: assert normalize_money("") is None assert normalize_money("sin valor") is None assert normalize_money("1.2.3") is None assert normalize_money("-$ 10.000") is None def test_cost_uses_quantity_times_unit_value_when_total_is_missing() -> None: lines = extract_invoice_cost_lines( _factura( [ { "categoria": "medicamentos", "codigo_referencia": "CUM-1", "descripcion": "Dipirona", "cantidad": "2", "valor_unitario": "$ 1.500,50", } ] ) ) cost = match_billed_cost( {"codigo_referencia": "CUM-1", "nombre": "Dipirona"}, category="medicamento", lines=lines, ) assert cost.valor_total == Decimal("3001.00") assert cost.origen == "cantidad_por_valor_unitario" assert cost.coincidencia == "codigo_exacto" def test_repeated_invoice_lines_select_largest_individual_total() -> None: lines = extract_invoice_cost_lines( _factura( [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "123456", "descripcion": "Lavado quirúrgico", "total": "80.000", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "123456", "descripcion": "Lavado quirúrgico", "total": "120.000", }, ] ) ) cost = match_billed_cost( {"codigo_cups": "123456", "descripcion": "Lavado quirúrgico"}, category="procedimiento", lines=lines, ) assert cost.valor_total == Decimal("120000") assert cost.linea_factura_id == "lineas_canonicas:1" def test_unique_text_matches_and_distinct_codes_make_same_text_ambiguous() -> None: unique_lines = extract_invoice_cost_lines( _factura( [ { "categoria": "imagenologia", "codigo_facturacion": "RX01", "descripcion": "Radiografía de tórax", "total": "50.000", } ] ) ) assert ( match_billed_cost( {"tipo": "imagen", "nombre": "Radiografia de torax"}, category="imagen", lines=unique_lines, ).coincidencia == "texto_normalizado_unico" ) ambiguous_lines = extract_invoice_cost_lines( _factura( [ { "categoria": "imagenologia", "codigo_facturacion": "RX01", "descripcion": "Radiografía de tórax", "total": "50.000", }, { "categoria": "imagenologia", "codigo_facturacion": "RX02", "descripcion": "Radiografía de tórax", "total": "70.000", }, ] ) ) ambiguous = match_billed_cost( {"tipo": "imagen", "nombre": "Radiografía de tórax"}, category="imagen", lines=ambiguous_lines, ) assert ambiguous.valor_total is None assert ambiguous.coincidencia == "ambigua" def test_procedure_cost_and_group_never_match_a_different_cups_by_description() -> None: lines = extract_invoice_cost_lines( _factura( [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "765401", "descripcion": "Reparación de hombro", "total": "344.200", } ], { "procedimientos_quirurgicos": [ { **_valued_procedure(0, 11, cups="765401"), "descripcion": "Reparación de hombro", } ] }, ) ) different_code = {"codigo_cups": "818307", "descripcion": "Reparación de hombro"} cost = match_billed_cost(different_code, category="procedimiento", lines=lines) assert cost.valor_total is None assert cost.coincidencia == "inexistente" assert match_billed_surgical_group(different_code, lines=lines) is None def test_stable_descending_sort_keeps_ties_and_unknown_items_last() -> None: lines = extract_invoice_cost_lines( _factura( [ {"categoria": "medicamentos", "descripcion": "A", "total": "10"}, {"categoria": "medicamentos", "descripcion": "B", "total": "20"}, {"categoria": "medicamentos", "descripcion": "C", "total": "20"}, {"categoria": "medicamentos", "descripcion": "D", "total": "0"}, ] ) ) ordered = attach_and_sort_costs( [ {"nombre": "A"}, {"nombre": "B"}, {"nombre": "C"}, {"nombre": "D"}, {"nombre": "Manual"}, ], category="medicamento", lines=lines, ) assert [item["nombre"] for item in ordered] == ["B", "C", "A", "D", "Manual"] def test_context_orders_medications_and_aids_without_mixing_aid_categories() -> None: context = { "factura": _factura( [ {"categoria": "medicamentos", "descripcion": "Acetaminofén", "total": "15.000"}, {"categoria": "medicamentos", "descripcion": "Cefazolina", "total": "90.000"}, {"categoria": "imagenologia", "descripcion": "RX tórax", "total": "40.000"}, {"categoria": "examenes_laboratorio", "descripcion": "Hemograma", "total": "25.000"}, ] ), "medicamentos_caso": [ {"nombre": "Acetaminofén", "medicamento": "Acetaminofén", "estados": ["administrado"]}, {"nombre": "Cefazolina", "medicamento": "Cefazolina", "estados": ["administrado"]}, {"nombre": "Manual", "medicamento": "Manual", "estados": ["administrado"]}, ], "ayudas_diagnosticas": [ {"tipo": "laboratorio", "nombre": "Hemograma", "fuente": "factura", "exportable": True}, {"tipo": "imagen", "nombre": "RX tórax", "fuente": "factura", "exportable": True}, {"tipo": "imagen", "nombre": "Ayuda manual", "fuente": "manual_pdf", "exportable": True}, ], } apply_factura_cost_ordering(context) assert [item["nombre"] for item in context["medicamentos_caso"]] == [ "Cefazolina", "Acetaminofén", "Manual", ] assert [item["nombre"] for item in context["ayudas_diagnosticas"]] == [ "RX tórax", "Hemograma", "Ayuda manual", ] assert [item["nombre"] for item in build_pdf_primary_medicamentos(context)] == [ "Cefazolina", "Acetaminofén", "Manual", ] assert [item["nombre"] for item in build_pdf_primary_ayudas_diagnosticas(context)] == [ "RX tórax", "Hemograma", "Ayuda manual", ] def test_procedures_sort_by_descending_soat_group_before_descending_price() -> None: canonical = [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": cups, "descripcion": f"Grupo {group}", "total": total, } for cups, group, total in ( ("230001", 23, "10"), ("200001", 20, "20"), ("130001", 13, "300000"), ("020001", 2, "900000"), ) ] services = { "procedimientos_quirurgicos": [ _valued_procedure(index, group, cups=cups) for index, (cups, group) in enumerate( (("230001", 23), ("200001", 20), ("130001", 13), ("020001", 2)) ) ] } lines = extract_invoice_cost_lines(_factura(canonical, services)) ordered = attach_and_sort_costs( [ {"codigo_cups": "020001", "descripcion": "Grupo 2"}, {"codigo_cups": "130001", "descripcion": "Grupo 13"}, {"codigo_cups": "200001", "descripcion": "Grupo 20"}, {"codigo_cups": "230001", "descripcion": "Grupo 23"}, ], category="procedimiento", lines=lines, ) assert [item["grupo_quirurgico_soat"] for item in ordered] == [23, 20, 13, 2] def test_procedure_sort_is_stable_by_price_inside_group_and_for_ungrouped_items() -> None: canonical = [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": cups, "descripcion": name, **({"total": total} if total is not None else {}), } for cups, name, total in ( ("230001", "Grupo caro", "200"), ("230002", "Grupo empate A", "100"), ("230003", "Grupo empate B", "100"), ("230004", "Grupo desconocido", None), ("990001", "Sin grupo caro", "900"), ("990002", "Sin grupo barato", "50"), ("990003", "Sin grupo desconocido", None), ) ] services = { "procedimientos_quirurgicos": [ _valued_procedure(index, 23, cups=cups) for index, cups in enumerate(("230001", "230002", "230003", "230004")) ] } lines = extract_invoice_cost_lines(_factura(canonical, services)) ordered = attach_and_sort_costs( [ {"codigo_cups": cups, "descripcion": name} for cups, name in ( ("230004", "Grupo desconocido"), ("230003", "Grupo empate B"), ("230002", "Grupo empate A"), ("230001", "Grupo caro"), ("990002", "Sin grupo barato"), ("990003", "Sin grupo desconocido"), ("990001", "Sin grupo caro"), ("999999", "Manual"), ) ], category="procedimiento", lines=lines, ) assert [item["descripcion"] for item in ordered] == [ "Grupo caro", "Grupo empate B", "Grupo empate A", "Grupo desconocido", "Sin grupo caro", "Sin grupo barato", "Sin grupo desconocido", "Manual", ] def test_invalid_or_contradictory_official_groups_do_not_group_procedures() -> None: canonical = [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "111111", "descripcion": "Contradictorio", "total": "500", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "222222", "descripcion": "Invalido", "total": "600", }, ] services = { "procedimientos_quirurgicos": [ _valued_procedure(0, 23, cups="111111"), _valued_procedure(0, 20, cups="111111"), _valued_procedure(1, 24, cups="222222"), ] } ordered = attach_and_sort_costs( [ {"codigo_cups": "111111", "descripcion": "Contradictorio"}, {"codigo_cups": "222222", "descripcion": "Invalido"}, ], category="procedimiento", lines=extract_invoice_cost_lines(_factura(canonical, services)), ) assert [item["descripcion"] for item in ordered] == ["Invalido", "Contradictorio"] assert all(item["grupo_quirurgico_soat"] is None for item in ordered) def test_explicit_soat_conflict_keeps_group_but_ambiguous_crosswalk_does_not() -> None: canonical = [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "111111", "descripcion": "SOAT explicito", "total": "10", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "222222", "descripcion": "Cruce ambiguo", "total": "900", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "333333", "descripcion": "Cruce no mapeado", "total": "800", }, ] services = { "procedimientos_quirurgicos": [ _valued_procedure( 0, 23, cups="111111", rule="soat_explicito_contradice_crosswalk", state="conflicto", crosswalk_status="conflict", ), _valued_procedure( 1, 20, cups="222222", rule="crosswalk_ambiguo", state="pendiente_revision", crosswalk_status="ambiguous", ), _valued_procedure( 2, 20, cups="333333", rule="crosswalk_no_mapeado", state="pendiente_revision", crosswalk_status="unmapped", ), ] } ordered = attach_and_sort_costs( [ {"codigo_cups": "222222", "descripcion": "Cruce ambiguo"}, {"codigo_cups": "333333", "descripcion": "Cruce no mapeado"}, {"codigo_cups": "111111", "descripcion": "SOAT explicito"}, ], category="procedimiento", lines=extract_invoice_cost_lines(_factura(canonical, services)), ) assert [item["descripcion"] for item in ordered] == [ "SOAT explicito", "Cruce ambiguo", "Cruce no mapeado", ] assert [item["grupo_quirurgico_soat"] for item in ordered] == [23, None, None] def test_service_procedure_group_is_direct_fallback_only_without_canonical_lines() -> None: service = _valued_procedure(None, 23, cups="111111") | { "descripcion": "Servicio directo", "total": "10", } direct_lines = extract_invoice_cost_lines(_factura(services={"procedimientos_quirurgicos": [service]})) with_canonical_lines = extract_invoice_cost_lines( _factura( [{"categoria": "medicamentos", "descripcion": "Otro", "total": "1"}], {"procedimientos_quirurgicos": [service]}, ) ) direct = attach_and_sort_costs( [{"codigo_cups": "111111", "descripcion": "Servicio directo"}], category="procedimiento", lines=direct_lines, ) canonical = attach_and_sort_costs( [{"codigo_cups": "111111", "descripcion": "Servicio directo"}], category="procedimiento", lines=with_canonical_lines, ) assert direct[0]["grupo_quirurgico_soat"] == 23 assert canonical[0]["grupo_quirurgico_soat"] is None def test_procedure_groups_and_pdf_projection_follow_cost_order() -> None: context = { "orden_costo_version": COST_ORDER_VERSION, "factura": _factura( [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "010101", "descripcion": "Procedimiento menor", "total": "20.000", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "020202", "descripcion": "Procedimiento mayor", "total": "200.000", }, ], { "procedimientos_quirurgicos": [ _valued_procedure(0, 23, cups="010101"), _valued_procedure(1, 2, cups="020202"), ] }, ), "procedimientos_hc": [ "010101 Procedimiento menor", "020202 Procedimiento mayor", "Procedimiento manual", ], "procedimientos_factura": [], "codigos_desde_soat": [], "pdf_draft": {}, } procedures = build_procedimientos_clinicos(context) assert [item["codigo_cups"] for item in procedures[:2]] == ["010101", "020202"] assert procedures[-1]["descripcion"] == "Procedimiento manual" assert build_pdf_primary_procedimientos(context) == [ "010101 - Procedimiento menor", "020202 - Procedimiento mayor", "Procedimiento manual", ] def test_services_are_used_only_as_category_fallback() -> None: lines = extract_invoice_cost_lines( _factura( [ { "categoria": "medicamentos", "descripcion": "Dipirona", "total": "10.000", } ], { "medicamentos": [{"medicamento": "Dipirona", "total": "99.000"}], "imagenologia": [{"estudio": "TAC", "descripcion": "TAC", "total": "80.000"}], }, ) ) medication = match_billed_cost({"nombre": "Dipirona"}, category="medicamento", lines=lines) image = match_billed_cost({"nombre": "TAC"}, category="imagen", lines=lines) assert medication.valor_total == Decimal("10000") assert image.valor_total == Decimal("80000") def test_curated_procedures_and_pdf_projection_follow_soat_group_order() -> None: context = { "factura": _factura( [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "010101", "descripcion": "Grupo alto", "total": "10", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "020202", "descripcion": "Grupo bajo", "total": "900", }, ], { "procedimientos_quirurgicos": [ _valued_procedure(0, 23, cups="010101"), _valued_procedure(1, 2, cups="020202"), ] }, ), "procedimientos_curados": [ {"codigo_cups": "020202", "description": "Grupo bajo"}, {"codigo_cups": "010101", "description": "Grupo alto"}, ], "pdf_primary_procedimientos": [ "020202 - Grupo bajo", "010101 - Grupo alto", ], } apply_factura_cost_ordering(context) assert [item["codigo_cups"] for item in context["procedimientos_curados"]] == [ "010101", "020202", ] assert context["pdf_primary_procedimientos"] == [ "010101 - Grupo alto", "020202 - Grupo bajo", ] def test_batch_excel_preserves_cost_order_without_rendering_cost_metadata() -> None: context = { "factura": _factura( [ { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "010101", "descripcion": "Procedimiento menor", "total": "20.000", }, { "categoria": "procedimientos_quirurgicos", "codigo_referencia": "020202", "descripcion": "Procedimiento mayor", "total": "200.000", }, {"categoria": "medicamentos", "descripcion": "Medicamento menor", "total": "10.000"}, {"categoria": "medicamentos", "descripcion": "Medicamento mayor", "total": "90.000"}, {"categoria": "imagenologia", "descripcion": "RX menor", "total": "30.000"}, {"categoria": "imagenologia", "descripcion": "TAC mayor", "total": "120.000"}, ], { "procedimientos_quirurgicos": [ { **_valued_procedure(0, 23, cups="010101"), "descripcion": "Procedimiento menor", }, { **_valued_procedure(1, 2, cups="020202"), "descripcion": "Procedimiento mayor", }, ] }, ), "procedimientos_hc": ["010101 Procedimiento menor", "020202 Procedimiento mayor"], "procedimientos_factura": [], "codigos_desde_soat": [], "pdf_draft": {}, "medicamentos_caso": [ { "nombre": "Medicamento menor", "medicamento": "Medicamento menor", "estados": ["administrado"], }, { "nombre": "Medicamento mayor", "medicamento": "Medicamento mayor", "estados": ["administrado"], }, ], "ayudas_diagnosticas": [ {"tipo": "imagen", "nombre": "RX menor", "fuente": "factura", "exportable": True}, {"tipo": "imagen", "nombre": "TAC mayor", "fuente": "factura", "exportable": True}, ], } apply_factura_cost_ordering(context) context["procedimientos_clinicos"] = build_procedimientos_clinicos(context) content = BatchEpicrisisExcelWorkbookBuilder().build( batch_id="batch-cost-order", rows=[ BatchEpicrisisExcelCaseRow( case_key="case-cost-order", patient_name="Paciente", patient_id="CC 1", case_number="1", procedure_description="", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=context, ) ], ) sheet = load_workbook(BytesIO(content))["case-cost-order"] visible = [str(cell.value) for row in sheet.iter_rows() for cell in row if cell.value is not None] assert visible.index("Procedimiento menor") < visible.index("Procedimiento mayor") assert visible.index("Medicamento mayor") < visible.index("Medicamento menor") assert visible.index("TAC mayor") < visible.index("RX menor") assert not any( "costo_facturado" in value or "grupo_quirurgico_soat" in value or value in {"200000", "120000", "90000"} for value in visible ) def test_pdf_cache_preserves_order_and_cost_version() -> None: class RequestStub: async def json(self): return { "html_pdf": "Epicrisis", "nombre_paciente": "Paciente", "procedimientos": ["020202 - Mayor", "010101 - Menor", "Manual"], "medicamentos": [ {"nombre": "Mayor", "medicamento": "Mayor"}, {"nombre": "Menor", "medicamento": "Menor"}, ], "ayudas_diagnosticas": [ {"tipo": "imagen", "nombre": "TAC"}, {"tipo": "laboratorio", "nombre": "Hemograma"}, ], } class CollectionStub: def __init__(self): self.inserted: Any = None def insert_one(self, document: dict[str, Any]): self.inserted = document return SimpleNamespace(inserted_id="cache-cost-order") collection = CollectionStub() services = SimpleNamespace( mongo_analyses=SimpleNamespace( collection=SimpleNamespace(database={"epicrisis_pdf_cache": collection}) ) ) result = asyncio.run( guardar_epicrisis_pdf_cache( cast(Request, RequestStub()), cast(UserInDB, SimpleNamespace(username="tester")), cast(AppServices, services), ) ) assert result["cache_id"] == "cache-cost-order" assert collection.inserted["orden_costo_version"] == COST_ORDER_VERSION assert collection.inserted["procedimientos"] == [ "020202 - Mayor", "010101 - Menor", "Manual", ] assert [item["medicamento"] for item in collection.inserted["medicamentos_detallados"]] == [ "Mayor", "Menor", ] assert [item["nombre"] for item in collection.inserted["ayudas_diagnosticas"]] == [ "TAC", "Hemograma", ] def test_case_cache_rejects_previous_cost_order_version() -> None: context = { "historia": {}, "quirurgico": {}, "factura": {}, "prescripcion": [], "auto_soat_results": [], "qx_agent_results": [], "manual_soat_results": [], "pdf_draft": {}, "procedimientos_hc": [], "medicamentos_hc": [], "medicamentos_hc_display": [], "medicamentos_caso": [], "medicamentos_prescripcion": [], "medicamentos_glosa": [], "procedimientos_factura": [], "imagenes_diagnosticas": [], "diagnosticos_consolidados": [], "diagnosticos_pendientes_validacion": [], "diagnosticos_consolidacion_hallazgos": [], "ayudas_diagnosticas": [], "orden_costo_version": COST_ORDER_VERSION, } assert COST_ORDER_VERSION == "soat_group_then_associated_surgical_cost_v4" assert is_compatible_cached_context(context) assert not is_compatible_cached_context({**context, "orden_costo_version": "factura_total_v1"}) def test_latest_pdf_cache_ignores_previous_cost_order_versions() -> None: class CollectionStub: def __init__(self): self.query: dict[str, Any] | None = None def find_one(self, query, **_kwargs): self.query = query return None collection = CollectionStub() services = SimpleNamespace( mongo_analyses=SimpleNamespace( collection=SimpleNamespace(database={"epicrisis_pdf_cache": collection}) ) ) response = asyncio.run( imprimir_ultimo_pdf_epicrisis_paciente( case_key="case-cost-order", current_user=cast(UserInDB, SimpleNamespace(username="tester")), services=cast(AppServices, services), ) ) assert response.status_code == 404 assert collection.query == { "usuario": "tester", "orden_costo_version": COST_ORDER_VERSION, "case_key": "case-cost-order", } def test_pdf_cache_filters_unclassified_procedures_and_keeps_compatible_text_list() -> None: class RequestStub: def __init__(self, procedures): self.procedures = procedures async def json(self): return { "html_pdf": "Epicrisis", "nombre_paciente": "Paciente", "procedimientos": ["Texto legado que no debe prevalecer"], "procedimientos_clasificados": self.procedures, } class CollectionStub: def __init__(self): self.inserted: Any = None def insert_one(self, document: dict[str, Any]): self.inserted = document return SimpleNamespace(inserted_id="cache-classified") collection = CollectionStub() services = SimpleNamespace( mongo_analyses=SimpleNamespace( collection=SimpleNamespace(database={"epicrisis_pdf_cache": collection}) ) ) asyncio.run( guardar_epicrisis_pdf_cache( cast( Request, RequestStub( [ { "codigo_cups": "010101", "descripcion": "Biopsia", "clasificacion": "quirurgico", }, {"descripcion": "Pendiente", "clasificacion": "sin_clasificar"}, {"descripcion": "Desconocido", "clasificacion": "administrativo"}, {"descripcion": "Ausente"}, ] ), ), cast(UserInDB, SimpleNamespace(username="tester")), cast(AppServices, services), ) ) assert collection.inserted["procedimientos"] == ["010101 - Biopsia"] assert collection.inserted["procedimientos_clasificados"] == [ { "codigo_cups": "010101", "descripcion": "Biopsia", "clasificacion": "quirurgico", } ] result = asyncio.run( guardar_epicrisis_pdf_cache( cast( Request, RequestStub( [ {"descripcion": "Pendiente", "clasificacion": "sin_clasificar"}, {"descripcion": "Ausente"}, ] ), ), cast(UserInDB, SimpleNamespace(username="tester")), cast(AppServices, services), ) ) assert result["cache_id"] == "cache-classified" assert collection.inserted["procedimientos"] == [] assert collection.inserted["procedimientos_clasificados"] == []