from __future__ import annotations import re import unicodedata from dataclasses import dataclass, field from decimal import Decimal, InvalidOperation from typing import Any from app.clinical_pipeline.domain import MonetaryAmount from app.llm.schemas import ( FacturaStructured, normalize_factura_date_value, normalize_factura_due_date_value, ) from app.services.patient_name_extraction import sanitize_patient_name_candidate from app.soat_crosswalk.application.resolution import CupsSoatResolver from app.soat_crosswalk.infrastructure.catalogs import JsonCupsSoatCrosswalk, XlsxCupsCatalog from app.soat_tariffs.application.valuation import valuate_factura_soat from app.soat_tariffs.infrastructure.json_catalog import JsonSoatTariffCatalog _CUPS_CATALOG = XlsxCupsCatalog() _DATE_LINE_PATTERN = re.compile(r"^\d{2}-\d{2}-\d{4}$") _QUANTITY_PATTERN = re.compile(r"^\d+(?:[.,]\d+)?$") _MONEY_PATTERN = re.compile(r"^-?\$\s*[\d.,]+$") _CODE_PATTERN = re.compile(r"^[A-Z0-9][A-Z0-9.-]+$", re.IGNORECASE) _NIT_PATTERN = re.compile(r"\bNIT:\s*([0-9.\-]+)", re.IGNORECASE) _CASE_PATTERN = re.compile(r"\bCaso\s+No\.\s*([A-Z0-9-]{4,20}?)(?=\s|N[uú]mero|$)", re.IGNORECASE) _PATIENT_ID_PATTERN = re.compile(r"\bCC:\s*([0-9.]+)", re.IGNORECASE) _PATIENT_WITH_ID_PATTERN = re.compile( r"Paciente:\s*(?:Fecha\s+Egreso:\s*[\d/:\s]+)?(?P.+?)\s+CC:\s*(?P[0-9.]+)", re.IGNORECASE | re.DOTALL, ) _PREFIJO_FACTURA_PATTERN = re.compile( r"Prefijo\s+No\.\s*Factura\s+([A-Z0-9-]+)\s+([A-Z0-9-]+)", re.IGNORECASE | re.DOTALL, ) _FACTURA_TITLE_NUMBER_PATTERN = re.compile( r"(?P\d{4,20})\s*Factura\s+Electr[oó]nica\s+de\s+Venta", re.IGNORECASE, ) _INLINE_INVOICE_ROW_PATTERN = re.compile( r"^\s*(?P\d{2}-\d{2}-\d{4})\s+" r"(?P[A-Z0-9][A-Z0-9.-]*)" r"(?:\s+(?P(?=[A-Z0-9.-]*\d)[A-Z0-9][A-Z0-9.-]*))?\s+" r"(?P.+?)\s+" r"(?P\d+(?:[.,]\d+)?)\s+" r"(?P-?\$\s*[\d.,]+)\s+" r"(?P-?\$\s*[\d.,]+)\s*$", re.IGNORECASE, ) _INLINE_INVOICE_ROW_START_PATTERN = re.compile( r"^\s*(?P\d{2}-\d{2}-\d{4})\s+" r"(?P[A-Z0-9][A-Z0-9.-]*)" r"(?:\s+(?P(?=[A-Z0-9.-]*\d)[A-Z0-9][A-Z0-9.-]*))?\s+" r"(?P.+)$", re.IGNORECASE, ) _INLINE_INVOICE_ROW_END_PATTERN = re.compile( r"^(?P.*?)\s+" r"(?P\d+(?:[.,]\d+)?)\s+" r"(?P-?\$\s*[\d.,]+)\s+" r"(?P-?\$\s*[\d.,]+)\s*$", re.IGNORECASE, ) _PAYER_PATTERN = re.compile(r"^([0-9.]+)\s*-\s*(.+)$") _CITY_AND_DATE_PATTERN = re.compile( r"^(?P.+?),\s*(?P\d{1,2}\s+[A-Za-záéíóúñ]+,\s+\d{4}\s+\d{1,2}:\d{2}(?:\s*[ap]\.m\.)?)$", re.IGNORECASE, ) _AM_PM_FIX_PATTERN = re.compile(r"(\d{1,2}:\d{2})\s*\n+\s*([ap])\.m\.", re.IGNORECASE) _VENCIMIENTO_PATTERN = re.compile( r"Vencimiento:\s*(.+?)(?:\s+Son:|\s+Total de los Servicios:|\s+Total Descuento:|\s+Copagos - Pagos:)", re.IGNORECASE | re.DOTALL, ) _VALOR_LETRAS_PATTERN = re.compile( r"Son:\s*(.+?)(?:\s+EMISOR:|\s+Total de los Servicios:|\s+Total Descuento:|\s+Copagos - Pagos:)", re.IGNORECASE | re.DOTALL, ) _SERVICE_HEADER_MARKERS = { "tipo de servicio y/o mercancia", "tipo de servicio y/o mercancía", } _TABLE_NOISE_MARKERS = { "fecha", "cod.", "cod", "cups-cum", "cant.", "cant", "precio", "total", "tipo de servicio y/o mercancia", "tipo de servicio y/o mercancía", } _FOOTER_CONTAINS = ( "proveedor tecnologico", "proveedor tecnológico", "página ", "facture s.a.s", "applicationresponse", "identificador evento:", "fecha evento:", "codigo evento:", "código evento:", "evento:", "* principal:", "* pbx:", ) _SUMMARY_LABELS = ( "total de los servicios:", "total descuento:", "copagos - pagos:", "total de la factura:", ) _HEADER_STOP_LABELS = { "convenio:", "fecha ingreso:", "paciente:", "fecha egreso:", "cc:", "por:", "no.autorizacion:", } _HEADER_LABELS = { "señores:", "dirección:", "convenio:", "fecha ingreso:", "paciente:", "fecha egreso:", "cc:", "por:", "no.autorizacion:", } _LAB_KEYWORDS = ( "hemograma", "cuadro hematico", "cuadro hemático", "protrombina", "tromboplastina", "leucograma", "creatinina", "nitrogeno ureico", "nitrógeno uréico", ) _IMAGING_KEYWORDS = ( "radiografia", "radiografía", "rx ", "rayos x", "fluoroscopia", "fluoroscop", "intensificador de imagen", "tomografia", "tomografía", "ecografia", "ecografía", "electrocardiograma", ) _HOSPITALIZATION_KEYWORDS = ( "habitacion", "habitación", "cama", "estancia", ) _HONORARIOS_KEYWORDS = ( "honorarios", "anestesiologo", "anestesiólogo", "ayudantia", "ayudantía", "cirujano", "ginecoobstetra", "consulta preanestesica", "consulta preanestésica", ) _OTHER_SERVICE_KEYWORDS = ( "sala de observacion", "sala de observación", "sala observacion", "derechos de sala", "derecho sala", "materiales", "material", "sutura", "curacion", "curación", "jeringa", "cateter", "catéter", "abocat", "buretrol", "equipo de venoclisis", "venoclisis", "equipo", "oxigeno", "oxígeno", "gases anestesicos", "gases anestésicos", "dispositivo", "cabestrillo", "tapon heparinizado", "tapón heparinizado", "sala cirugia", "sala cirugía", ) _MEDICATION_KEYWORDS = ( "acetaminofen", "cefalexina", "cefazolina", "cloruro de sodio", "solucion salina", "solución salina", "dipirona", "ketorolaco", "omeprazol", "tramadol", "dexketoprofeno", "capsula", "cápsula", "capsulas", "cápsulas", "tabletas", "tableta", "solucion inyectable", "solución inyectable", "ampollas", "ampolla", "frasco vial", "bolsa x", ) _PROCEDURE_KEYWORDS = ( "laparotomia", "laparotomía", "osteosintesis", "osteosíntesis", "ligamentorrafia", "reinsercion ligamentos", "reinserción ligamentos", "extraccion de cuerpos libres", "extracción de cuerpos libres", "condroplastia", "valoracion inicial intrahospitalaria", "valoración inicial intrahospitalaria", "consulta de urgencias", ) _NOISE_LINE_PATTERNS = ( re.compile(r"^\d{1,2}/\d{1,2}/\d{4}\s+\d{1,2}:\d{2}:\d{2}[ap]?\.\s*m\.?$", re.IGNORECASE), re.compile(r"^\d{10,}-\d+$"), ) _SHORT_DATE_PATTERN = re.compile(r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b") _SHORT_DATE_TIME_PATTERN = re.compile(r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}(?:\s+\d{1,2}:\d{2})?\b") _SERVICE_DATE_PATTERN = re.compile(r"\b\d{2}-\d{2}-\d{4}\b") _LONG_DATE_TIME_PATTERN = re.compile( r"\b\d{1,2}\s+[A-Za-záéíóúñ]+,\s+\d{4}\s+\d{1,2}:\d{2}(?:\s*[ap]\.m\.)?\b", re.IGNORECASE, ) _COMPANY_NAME_TOKENS = { "aseguradora", "entidad", "cooperativa", "eps", "ips", "s.a.s", "sas", "clinica", "clínica", "inversiones", "medicas", "médicas", "salud", } _INVALID_NAME_CONTEXT_TOKENS = { "caso", "fecha", "prefijo", "factura", "direccion", "dirección", "convenio", "autorizacion", "autorización", "tipo", "servicio", "mercancia", "mercancía", } def _clean_text(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "")).strip() def _normalize_ascii(value: Any) -> str: normalized = unicodedata.normalize("NFKD", str(value or "")) return "".join(char for char in normalized if not unicodedata.combining(char)).lower() def _normalize_invoice_text(raw_text: str) -> str: text = str(raw_text or "").replace("\r\n", "\n") text = _AM_PM_FIX_PATTERN.sub(r"\1 \2.m.", text) replacements = ( (r"(?<=\d)(Factura\s+Electr[oó]nica\s+de\s+Venta)", r"\n\1"), (r"(NO\s+RESPONSABLES\s+DE\s+IVA)(?=[A-ZÁÉÍÓÚÑ])", r"\1\n"), (r"(NIT:\s*[0-9.\-]+)(?=AV\b)", r"\1\n"), (r"(?<=\d)(N[uú]mero\s+de\s+autorizaci[oó]n)", r"\n\1"), (r"([ap]\.m\.)([A-Z0-9-]{1,10})(Prefijo\s+No\.\s*Factura)", r"\1\n\2\n\3"), (r"(\$[\d.,]+)(Vencimiento:)", r"\1\n\2"), (r"(\$[\d.,]+)(Total\s+Descuento:)", r"\1\n\2"), (r"(\$[\d.,]+)(Copagos\s+-\s+Pagos:)", r"\1\n\2"), (r"(\$[\d.,]+)(Total\s+de\s+la\s+Factura:)", r"\1\n\2"), ) for pattern, replacement in replacements: text = re.sub(pattern, replacement, text, flags=re.IGNORECASE) return text def _coalesce(*values: Any) -> str: for value in values: text = _clean_text(value) if text: return text return "" def _digits(value: Any) -> str: return re.sub(r"\D", "", str(value or "")) def _to_decimal(value: str) -> Decimal | None: parsed = MonetaryAmount.try_parse(value) return parsed.amount if parsed is not None else None def _format_decimal(value: Decimal, *, money: bool = False) -> str: quantized = value.quantize(Decimal("0.01")) text = f"{quantized:,.2f}" if money: return f"${text}" return text def _sanitize_entity_name(value: Any) -> str: cleaned = _clean_text(value) if not cleaned: return "" normalized = _normalize_ascii(cleaned) if any(token in normalized for token in _INVALID_NAME_CONTEXT_TOKENS): return "" if len(re.findall(r"[a-záéíóúñ]+", normalized, re.IGNORECASE)) < 2: return "" if len(_digits(cleaned)) > max(4, len(cleaned) // 3): return "" return cleaned def _sanitize_patient_name(value: Any) -> str: candidate = sanitize_patient_name_candidate(_clean_text(value)) if not candidate: return "" normalized = _normalize_ascii(candidate) if any(token in normalized for token in _COMPANY_NAME_TOKENS): return "" return candidate def _sanitize_identifier(value: Any, *, min_digits: int = 5) -> str: digits = _digits(value) return digits if len(digits) >= min_digits else "" def _sanitize_nit(value: Any) -> str: cleaned = _clean_text(value) digits = _digits(cleaned) if len(digits) < 6: return "" return cleaned if re.fullmatch(r"[0-9.\-]+", cleaned) else digits def _sanitize_case_number(value: Any) -> str: cleaned = _clean_text(value).upper() if not cleaned: return "" digits = re.match(r"\d{4,20}", cleaned) if digits: return digits.group(0) match = re.search(r"[A-Z0-9-]{4,20}", cleaned) return _clean_text(match.group(0)) if match else "" def _sanitize_code(value: Any) -> str: cleaned = _clean_text(value) return cleaned if cleaned and _is_code_token(cleaned) else "" def _sanitize_prefijo(value: Any) -> str: cleaned = _clean_text(value).upper() return cleaned if re.fullmatch(r"[A-Z0-9-]{1,10}", cleaned) else "" def _sanitize_quantity(value: Any) -> str: cleaned = _clean_text(value) return cleaned if cleaned and _QUANTITY_PATTERN.match(cleaned) else "" def _sanitize_money(value: Any) -> str: amount = MonetaryAmount.try_parse(value) return amount.formatted() if amount is not None else "" def _extract_first_date(value: Any, *, allow_time: bool = False, service_date: bool = False) -> str: cleaned = _clean_text(value) if not cleaned: return "" if service_date: match = _SERVICE_DATE_PATTERN.search(cleaned) return _clean_text(match.group(0)) if match else "" if allow_time: match = _SHORT_DATE_TIME_PATTERN.search(cleaned) return _clean_text(match.group(0)) if match else "" match = _SHORT_DATE_PATTERN.search(cleaned) return _clean_text(match.group(0)) if match else "" def _extract_issue_date_value(value: Any) -> str: cleaned = _clean_text(value) if not cleaned: return "" am_pm_match = re.search( r"\b\d{1,2}\s+[A-Za-záéíóúñ]+,\s+\d{4}\s+\d{1,2}:\d{2}\s*[ap]\.m\.", cleaned, flags=re.IGNORECASE, ) if am_pm_match: return _clean_text(am_pm_match.group(0)) match = _LONG_DATE_TIME_PATTERN.search(cleaned) if match: return _clean_text(match.group(0)) return "" def _sanitize_due_date(value: Any) -> str: cleaned = _clean_summary_value(_clean_text(value)) if not cleaned: return "" if _SHORT_DATE_PATTERN.search(cleaned): return _extract_first_date(cleaned) if len(cleaned) > 80: return "" return cleaned def _split_pages(raw_text: str) -> list[list[str]]: normalized = _normalize_invoice_text(raw_text) pages: list[list[str]] = [] for raw_page in normalized.split("\x0c"): page_lines = [_clean_text(line) for line in raw_page.splitlines()] page_lines = [line for line in page_lines if line] if page_lines: pages.append(page_lines) return pages def _flatten_pages(pages: list[list[str]]) -> list[str]: return [line for page in pages for line in page] def _is_noise_line(line: str) -> bool: normalized = _normalize_ascii(line) if normalized.startswith("vencimiento:") or normalized.startswith("son:") or _is_summary_label(line): return False if any(marker in normalized for marker in _FOOTER_CONTAINS): return True return any(pattern.match(line) for pattern in _NOISE_LINE_PATTERNS) def _is_table_noise_line(line: str) -> bool: normalized = _normalize_ascii(line) return normalized in _TABLE_NOISE_MARKERS def _is_summary_label(line: str) -> bool: normalized = _normalize_ascii(line) return any(normalized.startswith(label) for label in _SUMMARY_LABELS) def _is_code_token(token: str) -> bool: cleaned = _clean_text(token) normalized = _normalize_ascii(cleaned) if not cleaned or normalized in _TABLE_NOISE_MARKERS: return False if _DATE_LINE_PATTERN.match(cleaned) or _MONEY_PATTERN.match(cleaned): return False if _QUANTITY_PATTERN.match(cleaned): return bool(re.fullmatch(r"\d{4,}", cleaned)) return bool(_CODE_PATTERN.match(cleaned)) def _safe_value_after_label(lines: list[str], label: str, *, max_lines: int = 2) -> str: normalized_label = _normalize_ascii(label) for index, line in enumerate(lines): normalized_line = _normalize_ascii(line) label_position = normalized_line.find(normalized_label) if label_position < 0: continue remainder = _clean_text(line[label_position + len(label) :].lstrip(": ")) values = [remainder] if remainder else [] probe = index + 1 while probe < len(lines) and len(values) < max_lines: candidate = _clean_text(lines[probe]) normalized_candidate = _normalize_ascii(candidate) if not candidate: probe += 1 continue if normalized_candidate in _HEADER_LABELS or candidate.endswith(":"): break values.append(candidate) probe += 1 return _clean_text(" ".join(values)) return "" def _extract_amount_after_label(lines: list[str], label: str, *, max_lookahead: int = 5) -> str: normalized_label = _normalize_ascii(label) for index, line in enumerate(lines): normalized_line = _normalize_ascii(line) if not normalized_line.startswith(normalized_label): continue inline = _clean_text(line[len(label) :].lstrip(": ")) inline_amount = _extract_money_token(inline) if inline_amount: return inline_amount for probe in range(index + 1, min(len(lines), index + 1 + max_lookahead)): candidate = _clean_text(lines[probe]) amount = _extract_money_token(candidate) if amount: return amount if candidate.endswith(":"): break return "" def _extract_money_token(text: str) -> str: match = re.search(r"-?\s*\$\s*[\d.,]+", _clean_text(text)) return re.sub(r"\s+", "", match.group(0)) if match else "" def _extract_match_text(pattern: re.Pattern[str], text: str) -> str: match = pattern.search(text) return _clean_text(match.group(1)) if match else "" def _extract_inline_date_after_label(text: str, label: str) -> str: normalized_label = re.escape(label) pattern = re.compile( rf"{normalized_label}\s*:?\s*(?P\d{{1,2}}[/-]\d{{1,2}}[/-]\d{{2,4}}(?:\s+\d{{1,2}}:\d{{2}})?)", re.IGNORECASE, ) match = pattern.search(text) return _clean_text(match.group("date")) if match else "" def _extract_inline_money_after_label(text: str, label: str) -> str: pattern = re.compile(rf"{re.escape(label)}\s*:?\s*(?P-?\$\s*[\d.,]+)", re.IGNORECASE) match = pattern.search(text) return _clean_text(match.group("amount")) if match else "" def _extract_city_and_issue_date(first_page_lines: list[str]) -> tuple[str, str]: raw = _safe_value_after_label(first_page_lines, "Fecha Factura:", max_lines=2) if not raw: return "", "" issue_date = _extract_issue_date_value(raw) match = _CITY_AND_DATE_PATTERN.match(raw) if not match: city = _clean_text(raw.split(",")[0]) if "," in raw else "" return city, issue_date return _clean_text(match.group("city")), issue_date or _clean_text(match.group("date")) def _extract_provider_name(first_page_lines: list[str]) -> str: for index, line in enumerate(first_page_lines): if "factura electronica de venta" not in _normalize_ascii(line): continue for probe in range(index - 1, -1, -1): candidate = _clean_text(first_page_lines[probe]) normalized = _normalize_ascii(candidate) if not candidate: continue if "numero de autorizacion de facturacion electronica" in normalized: continue if candidate.endswith(":") or _is_noise_line(candidate): continue sanitized = _sanitize_entity_name(candidate) if sanitized: return sanitized compact = " ".join(first_page_lines) match = re.search( r"NO\s+RESPONSABLES\s+DE\s+IVA\s*(?P.+?)\s+NIT:", compact, flags=re.IGNORECASE, ) if match: return _sanitize_entity_name(match.group("provider")) return "" def _extract_provider_address(first_page_lines: list[str], provider_name: str) -> str: compact = " ".join(first_page_lines) inline_match = re.search(r"NIT:\s*[0-9.\-]+\s*(?P

AV\s+[^,\n]+?)(?:\s+Caso\s+No\.|\s*$)", compact) if inline_match: return _clean_text(inline_match.group("address")) for line in first_page_lines[:14]: candidate = _clean_text(line) if re.match(r"^(AV|CRA|CR|CALLE|CL|CARRERA)\b", candidate, flags=re.IGNORECASE): return candidate if not provider_name: return "" for index, line in enumerate(first_page_lines): if _clean_text(line) != provider_name: continue for probe in range(index + 1, len(first_page_lines)): candidate = _clean_text(first_page_lines[probe]) normalized = _normalize_ascii(candidate) if not candidate: continue if normalized == "factura electronica de venta": continue if normalized in {"prefijo", "no. factura"} or "prefijo no. factura" in normalized: break if candidate.endswith(":") or _is_noise_line(candidate): continue return candidate return "" def _extract_invoice_header_values(first_page_lines: list[str]) -> tuple[str, str]: compact = " ".join(first_page_lines) match = _PREFIJO_FACTURA_PATTERN.search(compact) if match: return _clean_text(match.group(1)), _clean_text(match.group(2)) title_match = _FACTURA_TITLE_NUMBER_PATTERN.search(compact) number = _clean_text(title_match.group("number")) if title_match else "" prefijo_match = re.search(r"(?:^|\s)([A-Z0-9-]{1,10})\s+Prefijo\s+No\.\s*Factura", compact, re.IGNORECASE) if prefijo_match and number: return _clean_text(prefijo_match.group(1)), number values: list[str] = [] capture = False for line in first_page_lines: normalized = _normalize_ascii(line) if normalized in {"prefijo", "no. factura"}: capture = True continue if not capture: continue if "contrubuyentes de renta" in normalized or "contribuyentes de renta" in normalized: break if _is_code_token(line): values.append(_clean_text(line)) if len(values) >= 2: return values[0], values[1] return "", "" def _slice_between(lines: list[str], start_label: str, stop_labels: tuple[str, ...]) -> list[str]: normalized_start = _normalize_ascii(start_label) normalized_stop = {_normalize_ascii(label) for label in stop_labels} start_index = None for index, line in enumerate(lines): if _normalize_ascii(line).startswith(normalized_start): start_index = index + 1 break if start_index is None: return [] values: list[str] = [] for line in lines[start_index:]: normalized = _normalize_ascii(line) if normalized in normalized_stop: break if line.endswith(":"): continue if _is_noise_line(line): continue values.append(_clean_text(line)) return [line for line in values if line] def _extract_payer_fields(first_page_lines: list[str]) -> tuple[str, str]: raw = _safe_value_after_label(first_page_lines, "Señores:", max_lines=1) match = _PAYER_PATTERN.match(raw) if match: payer_name = _sanitize_entity_name(match.group(2)) payer_nit = _sanitize_identifier(match.group(1), min_digits=6) if payer_name and payer_nit: return payer_name, payer_nit payer_candidates = _slice_between(first_page_lines, "Señores:", ("Convenio:", "Fecha Ingreso:")) for candidate in payer_candidates: if _normalize_ascii(candidate).startswith("caso no."): continue match = _PAYER_PATTERN.match(candidate) if match: payer_name = _sanitize_entity_name(match.group(2)) payer_nit = _sanitize_identifier(match.group(1), min_digits=6) if payer_name and payer_nit: return payer_name, payer_nit return "", "" def _looks_like_authorization(value: str, *, payer_name: str) -> bool: normalized = _normalize_ascii(value) if not normalized: return False if normalized == _normalize_ascii(payer_name): return False if normalized.startswith("colombia entidad cooperativa") or "aseguradora solidaria de" in normalized: return False if any(token in normalized for token in ("carrera", "calle", "avenida", "direccion", "dirección", "av ")): return False if "caso no" in normalized or "fecha" in normalized or "tipo de servicio" in normalized: return False return bool(re.search(r"[a-z0-9]", value, re.IGNORECASE) and (re.search(r"\d", value) or "-" in value)) def _extract_authorization(first_page_lines: list[str], payer_name: str) -> str: value = _safe_value_after_label(first_page_lines, "No.Autorizacion:", max_lines=2) return value if _looks_like_authorization(value, payer_name=payer_name) else "" def _extract_convenio(first_page_lines: list[str]) -> str: raw = _safe_value_after_label(first_page_lines, "Convenio:", max_lines=2) raw = re.sub(r"^Fecha\s+Ingreso:\s*[\d/:\s]+", "", raw, flags=re.IGNORECASE) raw = re.sub(r"\s+Fecha\s+Ingreso:\s*[\d/:\s]+.*$", "", raw, flags=re.IGNORECASE) return _clean_text(raw) def _extract_summary_lines(flat_lines: list[str]) -> list[str]: start_index = None for preferred in ("vencimiento:", "son:", "total de los servicios:", "copagos - pagos:"): for index in range(len(flat_lines) - 1, -1, -1): if _normalize_ascii(flat_lines[index]).startswith(preferred): start_index = index break if start_index is not None: break if start_index is None: return [] return [line for line in flat_lines[start_index:] if not _is_noise_line(line)] def _extract_vencimiento(summary_lines: list[str]) -> str: compact = " ".join(summary_lines) match = _VENCIMIENTO_PATTERN.search(compact) return _sanitize_due_date(match.group(1) if match else "") def _extract_valor_en_letras(summary_lines: list[str]) -> str: compact = " ".join(summary_lines) match = _VALOR_LETRAS_PATTERN.search(compact) value = _clean_summary_value(match.group(1) if match else "") value = re.sub(r"\s+\$[\d.,]+.*$", "", value).strip() value = re.sub(r"\s+Total\s+de\s+la\s+Factura:.*$", "", value, flags=re.IGNORECASE).strip() return value def _clean_summary_value(value: str) -> str: cleaned = _clean_text(value) if not cleaned: return "" normalized = _normalize_ascii(cleaned) for marker in _FOOTER_CONTAINS: position = normalized.find(marker) if position >= 0: return cleaned[:position].strip(" .,-") return cleaned.strip(" .,-") def _is_suspicious_summary_value(value: str) -> bool: normalized = _normalize_ascii(value) return any(marker in normalized for marker in _FOOTER_CONTAINS) @dataclass(slots=True) class _RawInvoiceLine: fecha_servicio: str codigo_facturacion: str codigo_referencia: str = "" descripcion_parts: list[str] = field(default_factory=list) cantidad: str = "" valor_unitario: str = "" total: str = "" @property def descripcion(self) -> str: return _clean_text(" ".join(self.descripcion_parts)) @dataclass(slots=True) class FacturaCanonicalLine: fecha_servicio: str codigo_facturacion: str codigo_referencia: str descripcion: str categoria: str capitulo_cups: int | None clasificacion_cups: str cantidad: str valor_unitario: str total: str def to_payload(self) -> dict[str, Any]: return { "fecha_servicio": self.fecha_servicio, "codigo_facturacion": self.codigo_facturacion, "codigo_referencia": self.codigo_referencia, "descripcion": self.descripcion, "categoria": self.categoria, "capitulo_cups": self.capitulo_cups, "clasificacion_cups": self.clasificacion_cups, "cantidad": self.cantidad, "valor_unitario": self.valor_unitario, "total": self.total, } @dataclass(slots=True) class FacturaParseAnalysis: nombre_paciente: str proveedor: dict[str, str] informacion_factura: dict[str, str] pagador: dict[str, str] informacion_paciente: dict[str, str] lineas_canonicas: list[FacturaCanonicalLine] servicios_procedimientos: dict[str, list[dict[str, str]]] analisis_financiero: dict[str, str] observaciones: list[str] low_confidence_sections: set[str] repair_contexts: dict[str, str] def to_factura_json(self) -> dict[str, Any]: return { "nombre_paciente": self.nombre_paciente, "proveedor": dict(self.proveedor), "informacion_factura": dict(self.informacion_factura), "pagador": dict(self.pagador), "informacion_paciente": dict(self.informacion_paciente), "lineas_canonicas": [line.to_payload() for line in self.lineas_canonicas], "servicios_procedimientos": dict(self.servicios_procedimientos), "analisis_financiero": dict(self.analisis_financiero), "observaciones": list(self.observaciones), "low_confidence_sections": sorted(self.low_confidence_sections), "repair_contexts": dict(self.repair_contexts), } def _extract_service_region(flat_lines: list[str]) -> list[str]: service_lines: list[str] = [] after_header = False after_authorization = False for line in flat_lines: normalized = _normalize_ascii(line) if normalized.startswith("no.autorizacion:"): after_authorization = True continue if any(marker in normalized for marker in _SERVICE_HEADER_MARKERS): after_header = True continue if after_authorization and normalized in _TABLE_NOISE_MARKERS: after_header = True continue if not after_header: continue if normalized.startswith("vencimiento:") or normalized.startswith("son:") or _is_summary_label(line): break if _is_noise_line(line) or _is_table_noise_line(line): continue service_lines.append(_clean_text(line)) return service_lines def _extract_service_region_from_pages(pages: list[list[str]]) -> list[str]: service_lines: list[str] = [] for page in pages: after_header = False after_authorization = False for line in page: normalized = _normalize_ascii(line) if normalized.startswith("no.autorizacion:"): after_authorization = True continue if any(marker in normalized for marker in _SERVICE_HEADER_MARKERS): after_header = True continue if after_authorization and normalized in _TABLE_NOISE_MARKERS: after_header = True continue if not after_header: continue if normalized.startswith("vencimiento:") or normalized.startswith("son:") or _is_summary_label(line): break if _is_noise_line(line) or _is_table_noise_line(line): continue service_lines.append(_clean_text(line)) return service_lines def _extract_financial_amounts(summary_lines: list[str]) -> dict[str, str]: labels = { "total de los servicios:": "total_servicios", "total descuento:": "descuentos", "copagos - pagos:": "copagos", "total de la factura:": "valor_total_factura", } results = {value: "" for value in labels.values()} trailing_amounts: list[str] = [] pending_keys: list[str] = [] for line in summary_lines: normalized = _normalize_ascii(line) matched_key = next((key for label, key in labels.items() if normalized.startswith(label)), "") if matched_key: pending_keys.append(matched_key) inline_amount = _extract_money_token(line) if inline_amount: results[matched_key] = inline_amount pending_keys = [key for key in pending_keys if key != matched_key] continue amount = _extract_money_token(line) if not amount: continue while pending_keys: pending_key = pending_keys.pop(0) if not results[pending_key]: results[pending_key] = amount break else: trailing_amounts.append(amount) continue if not results["copagos"] and trailing_amounts: results["copagos"] = trailing_amounts[0] if not results["valor_total_factura"] and len(trailing_amounts) >= 2: results["valor_total_factura"] = trailing_amounts[1] elif not results["valor_total_factura"] and trailing_amounts: results["valor_total_factura"] = trailing_amounts[-1] return results def _extract_patient_block(first_page_lines: list[str], first_page_compact: str) -> dict[str, str]: patient_name = "" inline_patient = _PATIENT_WITH_ID_PATTERN.search(first_page_compact) if inline_patient: patient_name = _sanitize_patient_name(inline_patient.group("name")) for index, line in enumerate(first_page_lines): if patient_name: break if not _normalize_ascii(line).startswith("paciente:"): continue inline_remainder = _clean_text(line[len("Paciente:") :]) inline_candidate = _sanitize_patient_name(inline_remainder) if inline_candidate: patient_name = inline_candidate break for probe in range(index + 1, min(index + 5, len(first_page_lines))): candidate = _sanitize_patient_name(first_page_lines[probe]) if candidate: patient_name = candidate break break patient_id = _sanitize_identifier( inline_patient.group("id") if inline_patient else _extract_match_text(_PATIENT_ID_PATTERN, first_page_compact) ) return { "nombre_completo": patient_name, "numero_identificacion": patient_id, "fecha_ingreso": _extract_first_date( _extract_inline_date_after_label(first_page_compact, "Fecha Ingreso") or _safe_value_after_label(first_page_lines, "Fecha Ingreso:", max_lines=1), allow_time=True, ), "fecha_egreso": _extract_first_date( _extract_inline_date_after_label(first_page_compact, "Fecha Egreso") or _safe_value_after_label(first_page_lines, "Fecha Egreso:", max_lines=1), allow_time=True, ), "estancia_hospitalaria": "", } def _canonical_line_from_raw(row: _RawInvoiceLine) -> FacturaCanonicalLine | None: description = _clean_text(row.descripcion) fecha_servicio = _extract_first_date(row.fecha_servicio, service_date=True) codigo_facturacion = _sanitize_code(row.codigo_facturacion) codigo_referencia = _sanitize_code(row.codigo_referencia) cantidad = _sanitize_quantity(row.cantidad) valor_unitario = _sanitize_money(row.valor_unitario) total = _sanitize_money(row.total) if not all([description, fecha_servicio, codigo_facturacion, cantidad, valor_unitario, total]): return None chapter, cups_classification = _cups_metadata(codigo_referencia) category = _classify_service_section( description, codigo_facturacion=codigo_facturacion, codigo_referencia=codigo_referencia, ) return FacturaCanonicalLine( fecha_servicio=fecha_servicio, codigo_facturacion=codigo_facturacion, codigo_referencia=codigo_referencia, descripcion=description, categoria=category, capitulo_cups=chapter, clasificacion_cups=cups_classification or "", cantidad=cantidad, valor_unitario=valor_unitario, total=total, ) def _canonical_line_from_payload(payload: Any) -> FacturaCanonicalLine | None: if not isinstance(payload, dict): return None fecha_servicio = normalize_factura_date_value( payload.get("fecha_servicio") or payload.get("fs"), allow_time=False, ) or _extract_first_date(payload.get("fecha_servicio") or payload.get("fs"), service_date=True) codigo_facturacion = _sanitize_code(payload.get("codigo_facturacion") or payload.get("cf")) codigo_referencia = _sanitize_code(payload.get("codigo_referencia") or payload.get("cr")) descripcion = _clean_text(payload.get("descripcion") or payload.get("d")) raw_quantity = payload.get("cantidad") if "cantidad" in payload else payload.get("q") raw_unit_value = payload.get("valor_unitario") if "valor_unitario" in payload else payload.get("vu") raw_total = payload.get("total") if "total" in payload else payload.get("tt") cantidad = _sanitize_quantity(raw_quantity) or _clean_text(raw_quantity) valor_unitario = _sanitize_money(raw_unit_value) or _clean_text(raw_unit_value) total = _sanitize_money(raw_total) or _clean_text(raw_total) if not all([fecha_servicio, codigo_facturacion, descripcion]): return None chapter, cups_classification = _cups_metadata(codigo_referencia) category = _classify_service_section( descripcion, codigo_facturacion=codigo_facturacion, codigo_referencia=codigo_referencia, ) return FacturaCanonicalLine( fecha_servicio=fecha_servicio, codigo_facturacion=codigo_facturacion, codigo_referencia=codigo_referencia, descripcion=descripcion, categoria=category, capitulo_cups=chapter, clasificacion_cups=cups_classification or "", cantidad=cantidad, valor_unitario=valor_unitario, total=total, ) def _canonical_payloads(lines: list[FacturaCanonicalLine]) -> list[dict[str, Any]]: return [line.to_payload() for line in lines] def _sanitize_canonical_payloads(items: list[dict[str, Any]] | list[Any]) -> list[dict[str, str]]: sanitized: list[FacturaCanonicalLine] = [] for item in items: line = _canonical_line_from_payload(item) if line is not None: sanitized.append(line) return _canonical_payloads(_group_contiguous_lines(sanitized)) def _build_lines_from_payloads(items: list[dict[str, str]]) -> list[FacturaCanonicalLine]: lines: list[FacturaCanonicalLine] = [] for item in items: line = _canonical_line_from_payload(item) if line is not None: lines.append(line) return lines def _raw_invoice_line_from_inline_match(match: re.Match[str]) -> _RawInvoiceLine: row = _RawInvoiceLine( fecha_servicio=_clean_text(match.group("fecha")), codigo_facturacion=_clean_text(match.group("codigo_facturacion")), codigo_referencia=_clean_text(match.group("codigo_referencia") or ""), cantidad=_clean_text(match.group("cantidad")), valor_unitario=_extract_money_token(match.group("valor_unitario")) or _clean_text(match.group("valor_unitario")), total=_extract_money_token(match.group("total")) or _clean_text(match.group("total")), ) body = _clean_text(match.group("body")) if body: row.descripcion_parts.append(body) return row def _parse_raw_lines(service_lines: list[str]) -> list[_RawInvoiceLine]: rows: list[_RawInvoiceLine] = [] pending_inline_row: _RawInvoiceLine | None = None index = 0 while index < len(service_lines): current_line = service_lines[index] inline_match = _INLINE_INVOICE_ROW_PATTERN.match(current_line) if inline_match: rows.append(_raw_invoice_line_from_inline_match(inline_match)) pending_inline_row = None index += 1 continue inline_start = _INLINE_INVOICE_ROW_START_PATTERN.match(current_line) if inline_start and not _extract_money_token(current_line): pending_inline_row = _RawInvoiceLine( fecha_servicio=_clean_text(inline_start.group("fecha")), codigo_facturacion=_clean_text(inline_start.group("codigo_facturacion")), codigo_referencia=_clean_text(inline_start.group("codigo_referencia") or ""), descripcion_parts=[_clean_text(inline_start.group("body"))], ) index += 1 continue inline_end = _INLINE_INVOICE_ROW_END_PATTERN.match(current_line) if pending_inline_row is not None and inline_end: body = _clean_text(inline_end.group("body")) if body: pending_inline_row.descripcion_parts.append(body) pending_inline_row.cantidad = _clean_text(inline_end.group("cantidad")) pending_inline_row.valor_unitario = ( _extract_money_token(inline_end.group("valor_unitario")) or _clean_text(inline_end.group("valor_unitario")) ) pending_inline_row.total = _extract_money_token(inline_end.group("total")) or _clean_text( inline_end.group("total") ) rows.append(pending_inline_row) pending_inline_row = None index += 1 continue if pending_inline_row is not None and current_line and not _DATE_LINE_PATTERN.match(current_line): pending_inline_row.descripcion_parts.append(current_line) index += 1 continue if ( rows and not _DATE_LINE_PATTERN.match(current_line) and not _INLINE_INVOICE_ROW_START_PATTERN.match(current_line) and not _extract_money_token(current_line) and not _is_table_noise_line(current_line) and not _is_noise_line(current_line) ): rows[-1].descripcion_parts.append(current_line) index += 1 continue pending_inline_row = None if not _DATE_LINE_PATTERN.match(current_line): index += 1 continue row = _RawInvoiceLine(fecha_servicio=current_line, codigo_facturacion="") index += 1 if index >= len(service_lines) or not _is_code_token(service_lines[index]): continue row.codigo_facturacion = service_lines[index] index += 1 if index < len(service_lines) and _is_code_token(service_lines[index]): row.codigo_referencia = service_lines[index] index += 1 while index < len(service_lines): candidate = service_lines[index] if _QUANTITY_PATTERN.match(candidate) or _DATE_LINE_PATTERN.match(candidate): break if _is_table_noise_line(candidate) or _is_noise_line(candidate): index += 1 continue row.descripcion_parts.append(candidate) index += 1 if index >= len(service_lines) or not _QUANTITY_PATTERN.match(service_lines[index]): continue row.cantidad = service_lines[index] index += 1 if index >= len(service_lines) or not _MONEY_PATTERN.match(service_lines[index]): continue row.valor_unitario = _extract_money_token(service_lines[index]) or _clean_text(service_lines[index]) index += 1 if index >= len(service_lines) or not _MONEY_PATTERN.match(service_lines[index]): continue row.total = _extract_money_token(service_lines[index]) or _clean_text(service_lines[index]) index += 1 rows.append(row) return rows def _repair_split_descriptions(rows: list[_RawInvoiceLine]) -> list[_RawInvoiceLine]: if not rows: return rows repaired: list[_RawInvoiceLine] = [] pending_blank: _RawInvoiceLine | None = None for row in rows: if not row.descripcion and pending_blank is None: pending_blank = row continue if pending_blank is not None and len(row.descripcion_parts) >= 2: # El texto aplanado ubica primero la continuación de la fila vacía y # después la descripcion de la fila actual. Conservar ese orden evita # intercambiar descripciones entre codigos consecutivos. pending_blank.descripcion_parts = row.descripcion_parts[:-1] row.descripcion_parts = row.descripcion_parts[-1:] repaired.append(pending_blank) pending_blank = None elif pending_blank is not None: pending_blank = None if row.descripcion: repaired.append(row) return repaired def _cups_metadata(code: str) -> tuple[int | None, str | None]: normalized = _digits(code) if not re.fullmatch(r"\d{6}", normalized): return None, None try: return _CUPS_CATALOG.chapter(normalized), _CUPS_CATALOG.classification(normalized) except (OSError, ValueError): return None, None def _classify_service_section(description: str, *, codigo_facturacion: str, codigo_referencia: str) -> str: normalized = _normalize_ascii(description) if any(keyword in normalized for keyword in _MEDICATION_KEYWORDS): return "medicamentos" if any(keyword in normalized for keyword in _LAB_KEYWORDS): return "examenes_laboratorio" if any(keyword in normalized for keyword in _IMAGING_KEYWORDS): return "imagenologia" if any(keyword in normalized for keyword in _HONORARIOS_KEYWORDS): return "honorarios_medicos" if any(keyword in normalized for keyword in _HOSPITALIZATION_KEYWORDS): return "hospitalizacion" if any(keyword in normalized for keyword in _OTHER_SERVICE_KEYWORDS): return "otros_servicios" _chapter, cups_classification = _cups_metadata(codigo_referencia) if cups_classification == "quirurgico": return "procedimientos_quirurgicos" if cups_classification == "no_quirurgico": return "procedimientos_no_quirurgicos" if any(keyword in normalized for keyword in _PROCEDURE_KEYWORDS): return "procedimientos_quirurgicos" if not codigo_referencia and re.fullmatch(r"\d{4,6}", _digits(codigo_facturacion)): return "otros_servicios" return "otros_servicios" def _group_contiguous_lines(lines: list[FacturaCanonicalLine]) -> list[FacturaCanonicalLine]: grouped: list[FacturaCanonicalLine] = [] for line in lines: if not grouped: grouped.append(line) continue previous = grouped[-1] same_signature = ( previous.fecha_servicio == line.fecha_servicio and previous.codigo_facturacion == line.codigo_facturacion and previous.codigo_referencia == line.codigo_referencia and previous.descripcion == line.descripcion and previous.categoria == line.categoria and previous.valor_unitario == line.valor_unitario ) if not same_signature: grouped.append(line) continue merged_quantity = _merge_numeric_text(previous.cantidad, line.cantidad) merged_total = _calculate_line_total(merged_quantity, previous.valor_unitario) grouped[-1] = FacturaCanonicalLine( fecha_servicio=previous.fecha_servicio, codigo_facturacion=previous.codigo_facturacion, codigo_referencia=previous.codigo_referencia, descripcion=previous.descripcion, categoria=previous.categoria, capitulo_cups=previous.capitulo_cups, clasificacion_cups=previous.clasificacion_cups, cantidad=merged_quantity or previous.cantidad, valor_unitario=previous.valor_unitario, total=merged_total or previous.total, ) return grouped def _merge_numeric_text(left: str, right: str) -> str: left_amount = _to_quantity_decimal(left) right_amount = _to_quantity_decimal(right) if left_amount is None or right_amount is None: return left return _format_decimal(left_amount + right_amount) def _to_quantity_decimal(value: str) -> Decimal | None: cleaned = _clean_text(value).replace(" ", "") if not cleaned: return None if re.fullmatch(r"-?\d+[.,]\d{2}", cleaned) or ("," in cleaned and "." not in cleaned): cleaned = cleaned.replace(",", ".") try: return Decimal(cleaned) except InvalidOperation: return None def _merge_money_text(left: str, right: str) -> str: left_amount = _to_decimal(left) right_amount = _to_decimal(right) if left_amount is None or right_amount is None: return left return _format_decimal(left_amount + right_amount, money=True) def _calculate_line_total(quantity: str, unit_value: str) -> str: quantity_amount = _to_quantity_decimal(quantity) unit_amount = MonetaryAmount.try_parse(unit_value) if quantity_amount is None or unit_amount is None: return "" return MonetaryAmount(amount=quantity_amount * unit_amount.amount).formatted() @dataclass(slots=True) class FacturaFinancialReconciliation: lineas_canonicas: list[dict[str, str]] resumen_financiero: dict[str, str] validacion_financiera: dict[str, Any] def reconcile_factura_financials( lineas_canonicas: list[dict[str, Any]], resumen_financiero: dict[str, Any], ) -> FacturaFinancialReconciliation: """Reconcilia importes en memoria sin consultar proveedores externos.""" tolerance = MonetaryAmount.CENT reasons: list[str] = [] details: list[dict[str, Any]] = [] corrected_lines: list[dict[str, str]] = [] selected_totals: list[Decimal] = [] corrected_count = 0 for index, source in enumerate(lineas_canonicas): line = {str(key): _clean_text(value) for key, value in dict(source).items() if value is not None} quantity = _to_quantity_decimal(line.get("cantidad", "")) unit_value = MonetaryAmount.try_parse(line.get("valor_unitario", "")) extracted_total = MonetaryAmount.try_parse(line.get("total", "")) line_codes: list[str] = [] if quantity is None: line_codes.append("financial_line_quantity_invalid") if unit_value is None: line_codes.append("financial_line_unit_value_invalid") if extracted_total is None: line_codes.append("financial_line_total_invalid") if quantity is not None: line["cantidad"] = _format_decimal(quantity) if unit_value is not None: line["valor_unitario"] = unit_value.formatted() calculated_total = ( MonetaryAmount(amount=quantity * unit_value.amount) if quantity is not None and unit_value is not None else None ) selected_total = extracted_total if calculated_total is not None: if extracted_total is None: selected_total = calculated_total elif not extracted_total.within(calculated_total, tolerance=tolerance): selected_total = calculated_total line["total"] = calculated_total.formatted() corrected_count += 1 details.append( { "indice": index, "codigo_facturacion": line.get("codigo_facturacion", ""), "codigo_referencia": line.get("codigo_referencia", ""), "total_extraido": source.get("total"), "total_calculado": calculated_total.formatted(), } ) elif extracted_total is not None: line["total"] = extracted_total.formatted() if selected_total is not None: selected_totals.append(selected_total.amount) if line_codes: reasons.extend(line_codes) details.append( { "indice": index, "codigo_facturacion": line.get("codigo_facturacion", ""), "codigo_referencia": line.get("codigo_referencia", ""), "codigos": line_codes, } ) corrected_lines.append(line) extracted_summary = { key: _clean_text(resumen_financiero.get(key)) for key in ("total_servicios", "descuentos", "copagos", "valor_total_factura") } parsed_summary = {key: MonetaryAmount.try_parse(value) for key, value in extracted_summary.items()} for key in ("total_servicios", "valor_total_factura"): if extracted_summary[key] and parsed_summary[key] is None: reasons.append(f"financial_{key}_invalid") all_line_operands_valid = len(selected_totals) == len(corrected_lines) and bool(corrected_lines) calculated_services = MonetaryAmount(amount=sum(selected_totals, Decimal("0.00"))) if all_line_operands_valid else None if not corrected_lines: reasons.append("financial_lines_missing") discounts = parsed_summary["descuentos"] copayments = parsed_summary["copagos"] missing_adjustments = [key for key in ("descuentos", "copagos") if not extracted_summary[key]] invalid_adjustments = [ key for key in ("descuentos", "copagos") if extracted_summary[key] and parsed_summary[key] is None ] if invalid_adjustments: reasons.extend(f"financial_{key}_invalid" for key in invalid_adjustments) if missing_adjustments: extracted_services = parsed_summary["total_servicios"] extracted_final = parsed_summary["valor_total_factura"] may_infer_zero = ( extracted_services is not None and extracted_final is not None and extracted_services.within(extracted_final, tolerance=tolerance) and not invalid_adjustments and all( adjustment is None or adjustment.amount == Decimal("0.00") for adjustment in (discounts, copayments) ) ) if may_infer_zero: zero = MonetaryAmount(amount=Decimal("0.00")) discounts = discounts or zero copayments = copayments or zero else: reasons.append("financial_adjustment_missing") normalized_summary = { key: _clean_text(value) for key, value in resumen_financiero.items() if value is not None } calculated_final: MonetaryAmount | None = None if calculated_services is not None and ( not extracted_summary["total_servicios"] or parsed_summary["total_servicios"] is not None ): normalized_summary["total_servicios"] = calculated_services.formatted() if discounts is not None: normalized_summary["descuentos"] = discounts.formatted() if copayments is not None: normalized_summary["copagos"] = copayments.formatted() if calculated_services is not None and discounts is not None and copayments is not None: calculated_final = MonetaryAmount( amount=calculated_services.amount - discounts.amount - copayments.amount ) if not extracted_summary["valor_total_factura"] or parsed_summary["valor_total_factura"] is not None: normalized_summary["valor_total_factura"] = calculated_final.formatted() unique_reasons = list(dict.fromkeys(reasons)) reconciled = not unique_reasons and calculated_services is not None and calculated_final is not None validation = { "version": "v1", "estado": "reconciliada" if reconciled else "no_reconciliada", "tolerancia": MonetaryAmount(amount=tolerance).formatted(), "totales_extraidos": extracted_summary, "totales_calculados": { "total_servicios": calculated_services.formatted() if calculated_services else "", "valor_total_factura": calculated_final.formatted() if calculated_final else "", }, "lineas_corregidas": corrected_count, "codigos_calidad": unique_reasons, "detalle_lineas": details, } return FacturaFinancialReconciliation(corrected_lines, normalized_summary, validation) def _build_service_item(line: FacturaCanonicalLine) -> tuple[str, dict[str, Any]]: base = { "fecha_servicio": line.fecha_servicio, "codigo_facturacion": line.codigo_facturacion, "codigo_referencia": line.codigo_referencia, "cantidad": line.cantidad, "valor_unitario": line.valor_unitario, "total": line.total, } if line.categoria in {"procedimientos_quirurgicos", "procedimientos_no_quirurgicos"}: return line.categoria, { **base, "concepto": line.codigo_facturacion or "No especificado", "codigo_cups": line.codigo_referencia or "", "descripcion": line.descripcion, "capitulo_cups": line.capitulo_cups, "clasificacion_cups": line.clasificacion_cups, } if line.categoria == "examenes_laboratorio": return line.categoria, { **base, "prueba": line.codigo_facturacion or line.descripcion, "descripcion": line.descripcion, } if line.categoria == "imagenologia": return line.categoria, { **base, "estudio": line.codigo_facturacion or line.descripcion, "descripcion": line.descripcion, } if line.categoria == "hospitalizacion": return line.categoria, { **base, "habitacion": line.descripcion, "dias": line.cantidad, "tarifa": line.valor_unitario, "descripcion": line.descripcion, } if line.categoria == "honorarios_medicos": return line.categoria, { **base, "rol": line.codigo_facturacion or line.descripcion, "profesional": line.descripcion, } if line.categoria == "medicamentos": return line.categoria, { **base, "medicamento": line.descripcion, "dosis": "", } return "otros_servicios", {**base, "descripcion": line.descripcion} def _build_service_sections(lines: list[FacturaCanonicalLine]) -> dict[str, list[dict[str, Any]]]: sections = { "procedimientos_quirurgicos": [], "procedimientos_no_quirurgicos": [], "examenes_laboratorio": [], "imagenologia": [], "hospitalizacion": [], "honorarios_medicos": [], "medicamentos": [], "otros_servicios": [], } for line in lines: section, item = _build_service_item(line) sections[section].append(item) return sections def _parse_canonical_lines(service_lines: list[str]) -> tuple[list[FacturaCanonicalLine], set[str]]: raw_rows = _repair_split_descriptions(_parse_raw_lines(service_lines)) canonical_lines: list[FacturaCanonicalLine] = [] low_confidence_sections: set[str] = set() for row in raw_rows: line = _canonical_line_from_raw(row) if line is None: low_confidence_sections.add("line_items") continue canonical_lines.append(line) grouped = _group_contiguous_lines(canonical_lines) return grouped, low_confidence_sections def analyze_factura_text(raw_text: str) -> FacturaParseAnalysis: pages = _split_pages(raw_text) flat_lines = [line for line in _flatten_pages(pages) if not _is_noise_line(line)] first_page_lines = pages[0] if pages else [] first_page_compact = "\n".join(first_page_lines) compact_text = "\n".join(flat_lines) summary_lines = _extract_summary_lines(flat_lines) provider_name = _extract_provider_name(first_page_lines) provider_address = _extract_provider_address(first_page_lines, provider_name) city_department, fecha_emision = _extract_city_and_issue_date(first_page_lines) prefijo, numero_factura = _extract_invoice_header_values(first_page_lines) payer_name, payer_nit = _extract_payer_fields(first_page_lines) service_region = _extract_service_region_from_pages(pages) lineas_canonicas, low_confidence_sections = _parse_canonical_lines(service_region) servicios = _build_service_sections(lineas_canonicas) fecha_vencimiento = _extract_vencimiento(summary_lines) valor_en_letras = _extract_valor_en_letras(summary_lines) financial_amounts = _extract_financial_amounts(summary_lines or flat_lines) total_servicios = financial_amounts["total_servicios"] or _extract_amount_after_label( summary_lines or flat_lines, "Total de los Servicios:", ) or _extract_inline_money_after_label(compact_text, "Total de los Servicios") descuentos = financial_amounts["descuentos"] or _extract_amount_after_label( summary_lines or flat_lines, "Total Descuento:", ) or _extract_inline_money_after_label(compact_text, "Total Descuento") copagos = financial_amounts["copagos"] or _extract_amount_after_label( summary_lines or flat_lines, "Copagos - Pagos:", ) or _extract_inline_money_after_label(compact_text, "Copagos - Pagos") valor_total_factura = financial_amounts["valor_total_factura"] or _extract_amount_after_label( summary_lines or flat_lines, "Total de la Factura:", ) or _extract_inline_money_after_label(compact_text, "Total de la Factura") patient_info = _extract_patient_block(first_page_lines, first_page_compact) patient_name = str(patient_info.get("nombre_completo") or "").strip() provider_nit = _sanitize_nit(_extract_match_text(_NIT_PATTERN, first_page_compact)) provider_address = _clean_text(provider_address) prefijo = _sanitize_prefijo(prefijo) numero_factura = _sanitize_case_number(numero_factura) case_number = _sanitize_case_number(_extract_match_text(_CASE_PATTERN, first_page_compact)) total_servicios = _sanitize_money(total_servicios) descuentos = "$0.00" if _clean_text(descuentos) == "$0.00" else _sanitize_money(descuentos) copagos = "$0.00" if _clean_text(copagos) == "$0.00" else _sanitize_money(copagos) valor_total_factura = _sanitize_money(valor_total_factura) if not patient_name: low_confidence_sections.add("identity") if patient_name and patient_name in (payer_name, provider_name): low_confidence_sections.add("identity") if not payer_name or not payer_nit or not provider_address or not provider_nit: low_confidence_sections.add("header") if not total_servicios or not valor_total_factura or _is_suspicious_summary_value(fecha_vencimiento): low_confidence_sections.add("summary") if not lineas_canonicas: low_confidence_sections.add("line_items") return FacturaParseAnalysis( nombre_paciente=patient_name, proveedor={ "nombre_institucion": provider_name, "nit": provider_nit, "direccion": provider_address, "ciudad_departamento": city_department, }, informacion_factura={ "numero_factura": numero_factura, "prefijo": prefijo, "fecha_emision": fecha_emision, "numero_caso": case_number, "fecha_vencimiento": fecha_vencimiento, }, pagador={ "aseguradora_eps": payer_name, "nit_pagador": payer_nit, "tipo_convenio": _extract_convenio(first_page_lines), "numero_autorizacion": _extract_authorization(first_page_lines, payer_name), }, informacion_paciente=patient_info, lineas_canonicas=lineas_canonicas, servicios_procedimientos=servicios, analisis_financiero={ "total_servicios": total_servicios, "descuentos": descuentos, "copagos": copagos, "valor_total_factura": valor_total_factura, "valor_en_letras": valor_en_letras, }, observaciones=[], low_confidence_sections=low_confidence_sections, repair_contexts={ "identity": "\n".join(first_page_lines), "header": "\n".join(first_page_lines), "summary": "\n".join(summary_lines), "line_items": "\n".join(service_region[:160]), }, ) def parse_factura_text(raw_text: str) -> dict[str, Any]: return analyze_factura_text(raw_text).to_factura_json() def _service_items_to_dicts(items: Any) -> list[dict[str, Any]]: normalized: list[dict[str, Any]] = [] for item in list(items or []): if isinstance(item, dict): normalized.append(dict(item)) continue model_dump = getattr(item, "model_dump", None) if callable(model_dump): normalized.append(dict(model_dump(exclude_none=True))) return normalized def _pick_service_list(parsed_items: Any, fallback_items: Any) -> list[dict[str, Any]]: preferred_items = _service_items_to_dicts(parsed_items) if preferred_items: return preferred_items return _service_items_to_dicts(fallback_items) def _pick_section(*values: Any) -> str: return _coalesce(*values) def _pick_preferred_value( validator, *values: Any, ) -> str: for value in values: text = _clean_text(value) if text and validator(text): return text return "" def _pick_money_value_preserving_invalid(*values: Any) -> str: first_nonempty = "" for value in values: text = _clean_text(value) if not text: continue first_nonempty = first_nonempty or text if MonetaryAmount.try_parse(text) is not None: return text return first_nonempty def _is_valid_name(value: str) -> bool: return bool(_sanitize_entity_name(value)) def _is_valid_patient_name(value: str) -> bool: return bool(_sanitize_patient_name(value)) def _is_valid_due_date(value: str) -> bool: return bool(normalize_factura_due_date_value(value, issue_date_value=None) or _sanitize_due_date(value)) def _is_valid_provider_address(value: str) -> bool: normalized = _normalize_ascii(value) return bool(value and "prefijo" not in normalized and "factura electronica" not in normalized) def _is_valid_authorization(value: str) -> bool: return _looks_like_authorization(value, payer_name="") def _is_valid_nit(value: str) -> bool: return bool(_sanitize_identifier(value, min_digits=6)) def _is_valid_patient_id(value: str) -> bool: return bool(_sanitize_identifier(value)) def _is_valid_case_number(value: str) -> bool: return bool(_sanitize_case_number(value)) def _is_valid_invoice_number(value: str) -> bool: sanitized = _sanitize_case_number(value) if not sanitized: return False return not (sanitized.isdigit() and len(sanitized) > 6) def _is_valid_issue_date(value: str) -> bool: return bool(normalize_factura_date_value(value, allow_time=True)) def _is_valid_datetime_value(value: str) -> bool: return bool(normalize_factura_date_value(value, allow_time=True)) def build_factura_structured( *, raw_text: str, current_structured: dict[str, Any] | None, patient_name: str = "", patient_id: str = "", case_number: str = "", repair_structured: dict[str, Any] | None = None, ) -> tuple[FacturaStructured, dict[str, Any]]: analysis = analyze_factura_text(raw_text) parsed = analysis.to_factura_json() current_model = FacturaStructured.model_validate(current_structured or {"dt": "factura"}) repair_model = FacturaStructured.model_validate(repair_structured or {"dt": "factura"}) prefer_repair_identity = "identity" in analysis.low_confidence_sections prefer_repair_header = "header" in analysis.low_confidence_sections prefer_repair_summary = "summary" in analysis.low_confidence_sections prefer_repair_lines = "line_items" in analysis.low_confidence_sections selected_canonical_lines = _sanitize_canonical_payloads( [line.model_dump(exclude_none=True) for line in current_model.lineas_canonicas] ) if prefer_repair_lines: repaired_canonical_lines = _sanitize_canonical_payloads( [line.model_dump(exclude_none=True) for line in repair_model.lineas_canonicas] ) if repaired_canonical_lines: selected_canonical_lines = repaired_canonical_lines if not selected_canonical_lines: selected_canonical_lines = _canonical_payloads(analysis.lineas_canonicas) selected_service_sections = ( _build_service_sections(_build_lines_from_payloads(selected_canonical_lines)) if selected_canonical_lines else {} ) normalized_payload = { "document_type": "factura", "patient_name": _pick_preferred_value( _is_valid_patient_name, patient_name, repair_model.patient_name if prefer_repair_identity else "", repair_model.patient_name if prefer_repair_header else "", current_model.patient_name, parsed["nombre_paciente"], ), "proveedor": { "nombre_institucion": _pick_preferred_value( _is_valid_name, current_model.proveedor.nombre_institucion, repair_model.proveedor.nombre_institucion if prefer_repair_header else "", parsed["proveedor"].get("nombre_institucion"), ), "nit": _pick_preferred_value( _is_valid_nit, current_model.proveedor.nit, repair_model.proveedor.nit if prefer_repair_header else "", parsed["proveedor"].get("nit"), ), "direccion": _pick_preferred_value( _is_valid_provider_address, current_model.proveedor.direccion, repair_model.proveedor.direccion if prefer_repair_header else "", parsed["proveedor"].get("direccion"), ), "ciudad_departamento": _pick_section( current_model.proveedor.ciudad_departamento, repair_model.proveedor.ciudad_departamento if prefer_repair_header else "", parsed["proveedor"].get("ciudad_departamento"), ), }, "datos_factura": { "numero_factura": _pick_preferred_value( _is_valid_invoice_number, _sanitize_case_number(current_model.datos_factura.numero_factura), _sanitize_case_number(parsed["informacion_factura"].get("numero_factura")), repair_model.datos_factura.numero_factura if prefer_repair_header else "", ), "prefijo": _pick_preferred_value( lambda value: bool(_sanitize_prefijo(value)), _sanitize_prefijo(current_model.datos_factura.prefijo), _sanitize_prefijo(parsed["informacion_factura"].get("prefijo")), repair_model.datos_factura.prefijo if prefer_repair_header else "", ), "fecha_emision": _pick_preferred_value( _is_valid_issue_date, current_model.datos_factura.fecha_emision, parsed["informacion_factura"].get("fecha_emision"), repair_model.datos_factura.fecha_emision if prefer_repair_header else "", ), "numero_caso": _pick_preferred_value( _is_valid_case_number, case_number, current_model.datos_factura.numero_caso, parsed["informacion_factura"].get("numero_caso"), repair_model.datos_factura.numero_caso if prefer_repair_header else "", ), "fecha_vencimiento": _pick_preferred_value( _is_valid_due_date, repair_model.datos_factura.fecha_vencimiento if prefer_repair_summary else "", parsed["informacion_factura"].get("fecha_vencimiento"), current_model.datos_factura.fecha_vencimiento, ), }, "pagador": { "aseguradora_eps": _pick_preferred_value( _is_valid_name, current_model.pagador.aseguradora_eps, repair_model.pagador.aseguradora_eps if prefer_repair_identity else "", repair_model.pagador.aseguradora_eps if prefer_repair_header else "", parsed["pagador"].get("aseguradora_eps"), ), "nit_pagador": _pick_preferred_value( _is_valid_nit, current_model.pagador.nit_pagador, repair_model.pagador.nit_pagador if prefer_repair_identity else "", repair_model.pagador.nit_pagador if prefer_repair_header else "", parsed["pagador"].get("nit_pagador"), ), "tipo_convenio": _pick_section( current_model.pagador.tipo_convenio, repair_model.pagador.tipo_convenio if prefer_repair_header else "", parsed["pagador"].get("tipo_convenio"), ), "numero_autorizacion": _pick_preferred_value( _is_valid_authorization, current_model.pagador.numero_autorizacion, repair_model.pagador.numero_autorizacion if prefer_repair_header else "", parsed["pagador"].get("numero_autorizacion"), ), }, "paciente": { "nombre_completo": _pick_preferred_value( _is_valid_patient_name, patient_name, current_model.paciente.nombre_completo, repair_model.paciente.nombre_completo if prefer_repair_identity else "", repair_model.paciente.nombre_completo if prefer_repair_header else "", parsed["informacion_paciente"].get("nombre_completo"), ), "numero_identificacion": _pick_preferred_value( _is_valid_patient_id, patient_id, current_model.paciente.numero_identificacion, repair_model.paciente.numero_identificacion if prefer_repair_identity else "", repair_model.paciente.numero_identificacion if prefer_repair_header else "", parsed["informacion_paciente"].get("numero_identificacion"), ), "fecha_ingreso": _pick_preferred_value( _is_valid_datetime_value, current_model.paciente.fecha_ingreso, repair_model.paciente.fecha_ingreso if prefer_repair_identity else "", repair_model.paciente.fecha_ingreso if prefer_repair_header else "", parsed["informacion_paciente"].get("fecha_ingreso"), ), "fecha_egreso": _pick_preferred_value( _is_valid_datetime_value, current_model.paciente.fecha_egreso, repair_model.paciente.fecha_egreso if prefer_repair_identity else "", repair_model.paciente.fecha_egreso if prefer_repair_header else "", parsed["informacion_paciente"].get("fecha_egreso"), ), "estancia_hospitalaria": _pick_section( current_model.paciente.estancia_hospitalaria, repair_model.paciente.estancia_hospitalaria if prefer_repair_header else "", parsed["informacion_paciente"].get("estancia_hospitalaria"), ), }, "lineas_canonicas": selected_canonical_lines, "servicios_procedimientos": { "procedimientos_quirurgicos": _pick_service_list( current_model.servicios_procedimientos.procedimientos_quirurgicos, selected_service_sections.get("procedimientos_quirurgicos", []) or parsed["servicios_procedimientos"]["procedimientos_quirurgicos"], ), "procedimientos_no_quirurgicos": _pick_service_list( current_model.servicios_procedimientos.procedimientos_no_quirurgicos, selected_service_sections.get("procedimientos_no_quirurgicos", []) or parsed["servicios_procedimientos"]["procedimientos_no_quirurgicos"], ), "examenes_laboratorio": _pick_service_list( current_model.servicios_procedimientos.examenes_laboratorio, selected_service_sections.get("examenes_laboratorio", []) or parsed["servicios_procedimientos"]["examenes_laboratorio"], ), "imagenologia": _pick_service_list( current_model.servicios_procedimientos.imagenologia, selected_service_sections.get("imagenologia", []) or parsed["servicios_procedimientos"]["imagenologia"], ), "hospitalizacion": _pick_service_list( current_model.servicios_procedimientos.hospitalizacion, selected_service_sections.get("hospitalizacion", []) or parsed["servicios_procedimientos"]["hospitalizacion"], ), "honorarios_medicos": _pick_service_list( current_model.servicios_procedimientos.honorarios_medicos, selected_service_sections.get("honorarios_medicos", []) or parsed["servicios_procedimientos"]["honorarios_medicos"], ), "medicamentos": _pick_service_list( current_model.servicios_procedimientos.medicamentos, selected_service_sections.get("medicamentos", []) or parsed["servicios_procedimientos"]["medicamentos"], ), "otros_servicios": _pick_service_list( getattr(current_model.servicios_procedimientos, "otros_servicios", []), selected_service_sections.get("otros_servicios", []) or parsed["servicios_procedimientos"]["otros_servicios"], ), }, "resumen_financiero": { "total_servicios": _pick_money_value_preserving_invalid( current_model.resumen_financiero.total_servicios, repair_model.resumen_financiero.total_servicios if prefer_repair_summary else "", parsed["analisis_financiero"].get("total_servicios"), ), "descuentos": _pick_money_value_preserving_invalid( current_model.resumen_financiero.descuentos, repair_model.resumen_financiero.descuentos if prefer_repair_summary else "", parsed["analisis_financiero"].get("descuentos"), ), "copagos": _pick_money_value_preserving_invalid( current_model.resumen_financiero.copagos, repair_model.resumen_financiero.copagos if prefer_repair_summary else "", parsed["analisis_financiero"].get("copagos"), ), "valor_total_factura": _pick_money_value_preserving_invalid( current_model.resumen_financiero.valor_total_factura, repair_model.resumen_financiero.valor_total_factura if prefer_repair_summary else "", parsed["analisis_financiero"].get("valor_total_factura"), ), "valor_en_letras": _pick_section( current_model.resumen_financiero.valor_en_letras, repair_model.resumen_financiero.valor_en_letras if prefer_repair_summary else "", parsed["analisis_financiero"].get("valor_en_letras"), ), "observaciones_importantes": _pick_section( current_model.resumen_financiero.observaciones_importantes, parsed["observaciones"][0] if parsed["observaciones"] else "", ), }, } reconciliation = reconcile_factura_financials( normalized_payload["lineas_canonicas"], normalized_payload["resumen_financiero"], ) normalized_payload["lineas_canonicas"] = reconciliation.lineas_canonicas normalized_payload["servicios_procedimientos"] = _build_service_sections( _build_lines_from_payloads(reconciliation.lineas_canonicas) ) normalized_payload["resumen_financiero"] = reconciliation.resumen_financiero financial_validation = reconciliation.validacion_financiera low_confidence_sections = set(analysis.low_confidence_sections) if financial_validation["estado"] == "reconciliada": if not _is_suspicious_summary_value(parsed["informacion_factura"].get("fecha_vencimiento", "")): low_confidence_sections.discard("summary") else: low_confidence_sections.add("financial") normalized_model = FacturaStructured.model_validate(normalized_payload) factura_json = { "nombre_paciente": normalized_model.patient_name, "proveedor": normalized_model.proveedor.model_dump(exclude_none=True), "informacion_factura": normalized_model.datos_factura.model_dump(exclude_none=True), "pagador": normalized_model.pagador.model_dump(exclude_none=True), "informacion_paciente": normalized_model.paciente.model_dump(exclude_none=True), "lineas_canonicas": [ line.model_dump(exclude_none=True) for line in normalized_model.lineas_canonicas ], "servicios_procedimientos": normalized_model.servicios_procedimientos.model_dump(exclude_none=True), "analisis_financiero": normalized_model.resumen_financiero.model_dump( exclude_none=True, exclude={"observaciones_importantes"}, ), "validacion_financiera": financial_validation, "observaciones": ( [normalized_model.resumen_financiero.observaciones_importantes] if normalized_model.resumen_financiero.observaciones_importantes else [] ), "low_confidence_sections": sorted(low_confidence_sections), "repair_contexts": dict(analysis.repair_contexts), } tariff_catalog = JsonSoatTariffCatalog() cups_catalog = XlsxCupsCatalog() crosswalk = JsonCupsSoatCrosswalk( cups_catalog=cups_catalog, tariff_catalog=tariff_catalog, ) factura_json = valuate_factura_soat( factura_json, tariff_catalog, CupsSoatResolver( cups_catalog=cups_catalog, crosswalk=crosswalk, tariff_catalog=tariff_catalog, ), ) enriched_model_payload = normalized_model.model_dump(exclude_none=True) enriched_model_payload["servicios_procedimientos"] = factura_json["servicios_procedimientos"] enriched_model_payload["resumen_tarifario_soat"] = factura_json["resumen_tarifario_soat"] normalized_model = FacturaStructured.model_validate(enriched_model_payload) return normalized_model, factura_json