from __future__ import annotations import re import unicodedata from datetime import date from decimal import ROUND_HALF_UP, Decimal from typing import Any, Protocol from app.soat_crosswalk.domain.models import CrosswalkResolution, CrosswalkStatus from app.soat_tariffs.application.surgical_costs import associate_surgical_costs from app.soat_tariffs.domain.models import SoatCatalogEntry, SoatTariffComponent from app.soat_tariffs.domain.ports import SoatTariffCatalogPort PROFESSIONAL_COMPONENTS = {"cirujano", "anestesia", "ayudantia"} class CrosswalkResolverPort(Protocol): def resolve( self, cups_code: str, year: int, *, billed_soat_code: str | None = None, ) -> CrosswalkResolution: ... def round_to_hundred(value: Decimal) -> int: return int((value / Decimal("100")).quantize(Decimal("1"), rounding=ROUND_HALF_UP) * 100) def _year(value: Any) -> int | None: text = str(value or "").strip() if not text: return None match = re.search(r"\b(20\d{2})\b", text) if not match: return None candidate = int(match.group(1)) try: date(candidate, 1, 1) except ValueError: return None return candidate def _component_value(component: SoatTariffComponent, unit_value: Decimal) -> int: if component.official_value is not None: return component.official_value return round_to_hundred(component.coefficient * unit_value) def _entry_values(entry: SoatCatalogEntry, unit_value: Decimal) -> tuple[int, dict[str, int]]: components: dict[str, int] = { component.name: _component_value(component, unit_value) for component in entry.components } # El decreto exige liquidar y redondear cada componente. Sumar los componentes # evita diferencias de centenas frente a redondear el coeficiente agregado. base = sum(components.values()) return base, components def tariff_entry_values(entry: SoatCatalogEntry, unit_value: Decimal) -> tuple[int, dict[str, int]]: """Devuelve tarifa base y componentes aplicando el redondeo normativo por componente.""" return _entry_values(entry, unit_value) def _normalize_billed_code(value: Any) -> str | None: raw = str(value or "").strip() if not raw.isdigit() or len(raw) not in {4, 5}: return None return raw.zfill(5) def _normalize_cups_code(value: Any) -> str | None: raw = str(value or "").strip() return raw if raw.isdigit() and len(raw) == 6 else None def _canonical_index(item: dict[str, Any], canonical_lines: list[dict[str, Any]]) -> int | None: sought = ( str(item.get("codigo_facturacion") or ""), str(item.get("codigo_referencia") or item.get("codigo_cups") or ""), str(item.get("fecha_servicio") or ""), ) for index, line in enumerate(canonical_lines): current = ( str(line.get("codigo_facturacion") or ""), str(line.get("codigo_referencia") or ""), str(line.get("fecha_servicio") or ""), ) if current == sought: return index return None def _description_evidence(billed: Any, catalogued: str) -> str: def tokens(value: Any) -> set[str]: ascii_text = unicodedata.normalize("NFKD", str(value or "")).encode("ascii", "ignore").decode() return {token for token in re.findall(r"[a-z0-9]+", ascii_text.casefold()) if len(token) > 2} billed_tokens = tokens(billed) catalogued_tokens = tokens(catalogued) if not billed_tokens: return "Descripcion facturada ausente; el codigo exacto sigue siendo la unica llave tarifaria." overlap = len(billed_tokens & catalogued_tokens) / max(1, len(catalogued_tokens)) if overlap >= 0.4: return "Descripcion facturada coherente como evidencia secundaria." return ( "Descripcion facturada divergente; requiere revision, sin alterar la coincidencia exacta por codigo." ) def _has_unambiguous_multiple_components( procedures: list[dict[str, Any]], canonical_lines: list[dict[str, Any]], principal_group: int ) -> bool: dates = {str(item.get("fecha_servicio") or "") for item in procedures} if len(dates) != 1 or "" in dates: return False component_codes = {str(line.get("codigo_facturacion") or "") for line in canonical_lines} room_code = str(39202 + principal_group) if principal_group <= 13 else "" material_code = "39303" if principal_group <= 9 else "39304" return room_code in component_codes and material_code in component_codes def valuate_factura_soat( factura: dict[str, Any], catalog: SoatTariffCatalogPort, crosswalk_resolver: CrosswalkResolverPort | None = None, ) -> dict[str, Any]: """Enriquece una proyección de factura sin modificar importes ni lineas originales.""" result = dict(factura) services = dict(result.get("servicios_procedimientos") or {}) procedures = [dict(item) for item in services.get("procedimientos_quirurgicos") or []] canonical_lines = [dict(item) for item in result.get("lineas_canonicas") or []] issue_year = _year((result.get("informacion_factura") or {}).get("fecha_emision")) resolved: list[ tuple[ int, dict[str, Any], SoatCatalogEntry, int, dict[str, int], bool, int, str, list[Any], CrosswalkResolution | None, bool, ] ] = [] pending = 0 versions: set[str] = set() for index, item in enumerate(procedures): service_year = _year(item.get("fecha_servicio")) year = service_year or issue_year used_issue_date = service_year is None and issue_year is not None code = _normalize_billed_code(item.get("codigo_soat") or item.get("codigo_facturacion")) cups_code = _normalize_cups_code(item.get("codigo_cups") or item.get("codigo_referencia")) valuation: dict[str, Any] if year is None: valuation = { "estado": "pendiente_revision", "regla_aplicada": "vigencia_no_determinada", "evidencia": ["No se encontró fecha de servicio ni fecha de emisión valida."], "indice_linea_canonica": _canonical_index(item, canonical_lines), } pending += 1 else: try: year_catalog = catalog.load(year) except (FileNotFoundError, ValueError): valuation = { "vigencia": year, "estado": "pendiente_revision", "regla_aplicada": "vigencia_no_disponible", "evidencia": [f"No existe catalogo tarifario valido para {year}."], "indice_linea_canonica": _canonical_index(item, canonical_lines), } pending += 1 else: entry = catalog.find_exact(year, code) if code is not None else None explicit_soat_valid = entry is not None crosswalk_resolution: CrosswalkResolution | None = None if crosswalk_resolver is not None and cups_code: try: crosswalk_resolution = crosswalk_resolver.resolve( cups_code, year, billed_soat_code=entry.code if entry is not None else None, ) except LookupError: crosswalk_resolution = None if ( entry is None and crosswalk_resolution is not None and crosswalk_resolution.status == CrosswalkStatus.RESOLVED and crosswalk_resolution.selected_soat_code ): entry = catalog.find_exact(year, crosswalk_resolution.selected_soat_code) if entry is None: rule = "codigo_soat_no_encontrado" evidence = [ "La valoracion exige un SOAT explicito valido o un cruce CUPS unico confirmado; " "no se uso similitud semantica." ] if code is None and cups_code and year != 2026: rule = "crosswalk_vigencia_no_disponible" if crosswalk_resolution is not None: rule = { CrosswalkStatus.AMBIGUOUS: "crosswalk_ambiguo", CrosswalkStatus.UNMAPPED: "crosswalk_no_mapeado", CrosswalkStatus.UNSUPPORTED_YEAR: "crosswalk_vigencia_no_disponible", }.get(crosswalk_resolution.status, rule) evidence.extend(crosswalk_resolution.evidence) valuation = { "vigencia": year, "unidad": year_catalog.unit, "estado": "pendiente_revision", "regla_aplicada": rule, "evidencia": evidence, "cruce_cups_soat": ( crosswalk_resolution.model_dump(mode="json") if crosswalk_resolution is not None else None ), "indice_linea_canonica": _canonical_index(item, canonical_lines), } pending += 1 else: base, components = _entry_values(entry, year_catalog.unit_value) versions.add(year_catalog.version) resolved.append( ( index, item, entry, base, components, used_issue_date, year, year_catalog.unit, year_catalog.generated_from, crosswalk_resolution, explicit_soat_valid, ) ) valuation = {} item["valoracion_soat"] = valuation procedures[index] = item if resolved: principal = max(resolved, key=lambda current: (current[3], -current[0])) principal_is_unique = sum(current[3] == principal[3] for current in resolved) == 1 multiple_is_unambiguous = ( len(resolved) > 1 and principal_is_unique and _has_unambiguous_multiple_components( [current[1] for current in resolved], canonical_lines, principal[2].surgical_group ) ) for ( index, item, entry, base, components, used_issue_date, year, unit, sources, crosswalk_resolution, explicit_soat_valid, ) in resolved: is_principal = index == principal[0] state = "liquidada" rule = "cirugia_unica_tarifa_base" liquidated = base liquidated_components = dict(components) assignment_source = "codigo_soat_explicito" if explicit_soat_valid else "crosswalk_cups_unico" evidence = [ ( f"Coincidencia exacta codigo_facturacion={entry.code}." if explicit_soat_valid else f"SOAT={entry.code} asignado por cruce CUPS unico confirmado." ), f"Descripcion del catalogo: {entry.description}", _description_evidence(item.get("descripcion"), entry.description), ] evidence.extend( f"Fuente: {source.url}; sha256={source.sha256}; origen={source.page_table}." for source in sources ) if used_issue_date: evidence.append("Vigencia sustituida por el ano de emision de la factura.") if crosswalk_resolution is not None: evidence.extend(crosswalk_resolution.evidence) if len(resolved) > 1: if multiple_is_unambiguous and is_principal: rule = "multiple_principal_100_por_ciento" elif multiple_is_unambiguous: rule = "multiple_misma_via_mismo_especialista_profesionales_50_por_ciento" liquidated_components = { name: round_to_hundred(Decimal(value) * Decimal("0.5")) for name, value in components.items() if name in PROFESSIONAL_COMPONENTS } liquidated = sum(liquidated_components.values()) else: state = "pendiente_revision" rule = "cirugias_multiples_ambiguas_tarifa_base" pending += 1 evidence.append( "No se estableció inequívocamente procedimiento principal, via, especialidad o componentes." ) if crosswalk_resolution is not None and crosswalk_resolution.status == CrosswalkStatus.CONFLICT: state = "conflicto" rule = "soat_explicito_contradice_crosswalk" pending += 1 procedures[index]["codigo_soat"] = entry.code procedures[index]["concepto_soat"] = entry.description procedures[index]["grupo_quirurgico_soat"] = entry.surgical_group procedures[index]["valoracion_soat"] = { "vigencia": year, "unidad": unit, "valor_base": base, "valor_liquidado": liquidated, "componentes": components, "componentes_liquidados": liquidated_components, "codigos_componentes": {component.name: component.code for component in entry.components}, "regla_aplicada": rule, "fuente_asignacion": assignment_source, "estado": state, "evidencia": evidence, "cruce_cups_soat": ( crosswalk_resolution.model_dump(mode="json") if crosswalk_resolution is not None else None ), "indice_linea_canonica": _canonical_index(item, canonical_lines), } services["procedimientos_quirurgicos"] = procedures result["servicios_procedimientos"] = services valued = [item.get("valoracion_soat") or {} for item in procedures] result["resumen_tarifario_soat"] = { "total_base": sum(int(item.get("valor_base") or 0) for item in valued), "total_liquidado": sum(int(item.get("valor_liquidado") or 0) for item in valued), "pendientes": pending, "version_catalogo": ", ".join(sorted(versions)) or None, "nota": "Valoración de referencia; no modifica el total declarado de la factura.", } return associate_surgical_costs(result)