from __future__ import annotations import hashlib import json import re import unicodedata from collections.abc import Iterable from datetime import UTC, datetime from decimal import ROUND_HALF_UP, Decimal from typing import Any from app.case_epicrisis.domain.curation import ( CURATION_SCHEMA_VERSION, CandidateOrigin, CaseCuration, CatalogReference, CodeSystem, CodingConflict, CurationDecision, CurationDecisionType, CurationStatus, EvidenceReference, ObjectiveData, ObjectiveDiagnosis, ObjectiveProcedure, PostoperativeSignal, ProcedureReconciliationStatus, ProcedureState, SoatCurationCandidate, SoatCurationValuation, dump_curation_model, ) from app.case_epicrisis.domain.ports import CodingCatalogRegistry from app.services.deterministic_signals import ( DeterministicSignalSnapshot, resolve_postoperative_procedure, ) _TECHNICAL_CODE_VALUES = {"", "PENDIENTE", "SIN CODIGO", "SIN CÓDIGO", "ERROR", "NO DISPONIBLE"} _PRIMARY_TERMS = ("FRACTURA", "INFECCION", "INFECCIÓN", "NEOPLASIA", "INFARTO") _LOCATION_GROUPS = ( ({"PROXIMAL", "EPIFISIS", "EPÍFISIS", "SUPERIOR"}, {"DIAFISIS", "DIÁFISIS"}), ({"HOMBRO"}, {"CODO", "PIE", "RODILLA", "TOBILLO"}), ) _REFERENCE_COMPONENTS = {"cirujano", "anestesia", "ayudantia", "sala", "materiales"} _PROFESSIONAL_REFERENCE_COMPONENTS = {"cirujano", "anestesia", "ayudantia"} def _ascii_upper(value: Any) -> str: normalized = unicodedata.normalize("NFKD", str(value or "")) return "".join(char for char in normalized if not unicodedata.combining(char)).upper() def _clean_text(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "")).strip() def _normalized_key(value: Any) -> str: return re.sub(r"[^A-Z0-9]+", "-", _ascii_upper(value)).strip("-").lower() def _stable_id(prefix: str, *values: Any) -> str: raw = "|".join(_clean_text(value) for value in values) return f"{prefix}:{hashlib.sha256(raw.encode()).hexdigest()[:20]}" def _real_code(value: Any) -> str: code = _clean_text(value) return "" if _ascii_upper(code) in _TECHNICAL_CODE_VALUES else code def _context_type(value: Any) -> str: text = _ascii_upper(value) if re.search(r"\b(POST|POSTOP|POST-OP|POSTOPERATORIO|POSOPERATORIO)\b", text): return "post" if re.search(r"\b(PRE|PREQUIRURGICO|PREOPERATORIO)\b", text): return "pre" return "no_clasificado" def _is_primary_description(value: str) -> bool: upper = _ascii_upper(value) return any(term in upper for term in _PRIMARY_TERMS) def _evidence_for_document( document: dict[str, Any], *, section: str, excerpt: str, page: int | None = None, ) -> EvidenceReference: return EvidenceReference( document_id=str(document.get("_id") or document.get("id") or ""), document_type=str(document.get("tipo_documento") or document.get("document_type") or "desconocido"), filename=str(document.get("nombre_archivo") or ""), page=page if page and page > 0 else None, section=section, excerpt=_clean_text(excerpt)[:280], source_hash=str(document.get("source_file_hash") or ""), ) def _coerce_page(value: Any) -> int | None: if value is None: return None try: page = int(value) except (TypeError, ValueError): return None return page if page > 0 else None def _inline_evidence( signals: DeterministicSignalSnapshot | None, *, system: CodeSystem, code: str, ) -> dict[str, Any] | None: if signals is None: return None entries = signals.inline_cie10 if system == CodeSystem.CIE10 else signals.inline_cups code_key = re.sub(r"[^A-Z0-9]", "", _ascii_upper(code)) for item in entries: item_code = item.get("codigo") if system == CodeSystem.CIE10 else item.get("codigo_cups") if re.sub(r"[^A-Z0-9]", "", _ascii_upper(item_code)) == code_key: return item return None def _postoperative_signal_model( raw: dict[str, Any], *, document: dict[str, Any], ) -> PostoperativeSignal: return PostoperativeSignal.model_validate( { **raw, "page": _coerce_page(raw.get("page") or document.get("source_page_start")), "document_id": str(document.get("_id") or document.get("id") or ""), "document_type": str(document.get("tipo_documento") or document.get("document_type") or ""), "filename": str(document.get("nombre_archivo") or ""), } ) def _procedure_matches_context( procedure: ObjectiveProcedure, context: dict[str, Any], ) -> bool: context_code = _real_code(context.get("codigo_cups")) if context_code and procedure.codigo_cups: return context_code == _real_code(procedure.codigo_cups) procedure_text = _normalized_key(procedure.description) context_text = _normalized_key(context.get("description")) if not procedure_text or not context_text: return False return procedure_text in context_text or context_text in procedure_text def _apply_postoperative_context( *, document: dict[str, Any], document_type: str, procedures: list[ObjectiveProcedure], deterministic_signals: DeterministicSignalSnapshot | None, ) -> list[PostoperativeSignal]: if document_type not in {"historia_clinica", "quirurgico"} or deterministic_signals is None: return [] resolved_signals: list[PostoperativeSignal] = [] for raw_signal in deterministic_signals.postoperative_signals: signal = _postoperative_signal_model(raw_signal, document=document) context = resolve_postoperative_procedure( raw_signal, deterministic_signals.procedure_contexts, ) if context is None: signal.correlation_reason = ( "POP no tiene un procedimiento explícito único en la misma sección o bloque de evolución." ) resolved_signals.append(signal) continue candidates = [procedure for procedure in procedures if _procedure_matches_context(procedure, context)] if len(candidates) != 1: signal.correlation_reason = ( "El procedimiento contextual no coincide de forma única con un procedimiento actual extraído." ) resolved_signals.append(signal) continue procedure = candidates[0] if ProcedureState.PERFORMED not in procedure.states: procedure.states.append(ProcedureState.PERFORMED) procedure.is_primary = True procedure.evidence.append( _evidence_for_document( document, section=signal.section or "postoperatorio", excerpt=signal.excerpt, page=signal.page, ) ) procedure.evidence = [ evidence for index, evidence in enumerate(procedure.evidence) if evidence not in procedure.evidence[:index] ] procedure.warnings = list(dict.fromkeys([*procedure.warnings, "postoperatorio_contextual"])) signal.correlation_status = "correlacionado" signal.correlation_reason = "POP correlacionado con el procedimiento explícito actual más cercano." signal.procedure_item_id = procedure.item_id resolved_signals.append(signal) return resolved_signals def _candidate_status( *, catalog_registry: CodingCatalogRegistry, system: CodeSystem, code: str, description: str, explicit: bool, ) -> tuple[CurationStatus, list[str], CatalogReference]: reference = catalog_registry.reference(system) warnings: list[str] = [] if not explicit: warnings.append("codigo_inferido_requiere_revision") if not reference.available: warnings.append("catalogo_no_disponible") elif not catalog_registry.contains(system, code): warnings.append("codigo_no_encontrado_en_catalogo") elif not catalog_registry.description_matches(system, code, description): warnings.append("codigo_descripcion_incompatibles") if explicit and not warnings: return CurationStatus.AUTO_VALIDATED, warnings, reference return CurationStatus.PENDING_REVIEW, warnings, reference def build_document_objective_data( document: dict[str, Any], *, catalog_registry: CodingCatalogRegistry, deterministic_signals: DeterministicSignalSnapshot | None = None, ) -> ObjectiveData: document_type = str(document.get("tipo_documento") or document.get("document_type") or "") diagnoses: list[ObjectiveDiagnosis] = [] procedures: list[ObjectiveProcedure] = [] for item in document.get("codigos_cie10") or []: if not isinstance(item, dict): continue code = _real_code(item.get("codigo")) description = _clean_text(item.get("descripcion") or item.get("diagnostico")) original = _clean_text(item.get("diagnostico") or description) if not description: continue inline = _inline_evidence(deterministic_signals, system=CodeSystem.CIE10, code=code) explicit = bool(code and inline) status = CurationStatus.PENDING_REVIEW warnings = ["diagnostico_sin_codigo"] if not code else [] catalog = None if code: status, code_warnings, catalog = _candidate_status( catalog_registry=catalog_registry, system=CodeSystem.CIE10, code=code, description=description, explicit=explicit, ) warnings.extend(code_warnings) evidence = _evidence_for_document( document, section="diagnosticos", excerpt=(inline or {}).get("excerpt") or original, page=int((inline or {}).get("page") or document.get("source_page_start") or 0) or None, ) diagnoses.append( ObjectiveDiagnosis( item_id=_stable_id("dx", document.get("_id"), code, original), code=code, description=description, original_text=original, origin=CandidateOrigin.EXPLICIT if explicit else CandidateOrigin.RAG, status=status, evidence=[evidence], catalog=catalog, context_type=_context_type(original), is_primary=_is_primary_description(original), warnings=list(dict.fromkeys(warnings)), ) ) procedure_state = { "quirurgico": ProcedureState.PERFORMED, "factura": ProcedureState.BILLED, }.get(document_type, ProcedureState.DOCUMENTED) for item in document.get("codigos_cups") or []: if not isinstance(item, dict): continue code = _real_code(item.get("codigo_cups") or item.get("codigo")) description = _clean_text(item.get("procedimiento") or item.get("descripcion")) if not description: continue inline = _inline_evidence(deterministic_signals, system=CodeSystem.CUPS, code=code) explicit = bool(code and inline) status = CurationStatus.PENDING_REVIEW warnings = ["procedimiento_sin_cups"] if not code else [] catalog = None if code: status, code_warnings, catalog = _candidate_status( catalog_registry=catalog_registry, system=CodeSystem.CUPS, code=code, description=description, explicit=explicit, ) warnings.extend(code_warnings) procedures.append( ObjectiveProcedure( item_id=_stable_id("px", document.get("_id"), code, description, procedure_state), description=description, original_text=description, codigo_cups=code, origin=CandidateOrigin.EXPLICIT if explicit else CandidateOrigin.RAG, status=status, states=[procedure_state], evidence=[ _evidence_for_document( document, section="procedimientos", excerpt=(inline or {}).get("excerpt") or description, page=int((inline or {}).get("page") or document.get("source_page_start") or 0) or None, ) ], catalog=catalog, is_primary=procedure_state == ProcedureState.PERFORMED, warnings=list(dict.fromkeys(warnings)), classification=( getattr(catalog_registry, "classification", lambda *_: None)(CodeSystem.CUPS, code) or ("quirurgico" if document_type == "quirurgico" else "sin_clasificar") ), ) ) if document_type == "factura": procedures.extend(_build_invoice_procedures(document, catalog_registry)) else: procedures.extend(_build_uncoded_procedures(document, procedure_state)) postoperative_signals = _apply_postoperative_context( document=document, document_type=document_type, procedures=procedures, deterministic_signals=deterministic_signals, ) retryable_errors = [] if document.get("error_cups"): retryable_errors.append("cups_provider_unavailable") for procedure in procedures: if not procedure.codigo_cups: procedure.status = CurationStatus.EXTERNAL_SERVICE_PENDING procedure.warnings = list( dict.fromkeys([*procedure.warnings, "codificacion_cups_reintentable"]) ) return ObjectiveData( diagnoses=_dedupe_diagnoses(diagnoses), procedures=_dedupe_procedures(procedures), catalogs=catalog_registry.references(), processing_status="partial" if retryable_errors else "completed", retryable_errors=retryable_errors, postoperative_signals=postoperative_signals, ) def _build_invoice_procedures( document: dict[str, Any], catalog_registry: CodingCatalogRegistry, ) -> list[ObjectiveProcedure]: factura_json = document.get("factura_json") or {} services = factura_json.get("servicios_procedimientos") or {} results: list[ObjectiveProcedure] = [] invoice_sections = ( ("procedimientos_quirurgicos", "quirurgico"), ("procedimientos_no_quirurgicos", "no_quirurgico"), ) for section, classification in invoice_sections: for item in services.get(section) or []: if not isinstance(item, dict): continue description = _clean_text(item.get("descripcion") or item.get("concepto")) cups = _real_code(item.get("codigo_cups") or item.get("codigo_referencia")) billing = _real_code(item.get("codigo_facturacion") or item.get("concepto")) explicit_soat = _real_code(item.get("codigo_soat")) valuation = item.get("valoracion_soat") if isinstance(item.get("valoracion_soat"), dict) else {} crosswalk_value = valuation.get("cruce_cups_soat") crosswalk: dict[str, Any] = crosswalk_value if isinstance(crosswalk_value, dict) else {} soat_candidates = [ SoatCurationCandidate( soat_code=str(candidate.get("soat_code") or ""), description=str(candidate.get("soat_description") or ""), surgical_group=candidate.get("surgical_group"), reference=str(candidate.get("reference") or ""), ) for candidate in crosswalk.get("candidates") or [] if isinstance(candidate, dict) and candidate.get("soat_code") ] if not description: continue status = CurationStatus.PENDING_REVIEW warnings = [] catalog = None if cups: status, warnings, catalog = _candidate_status( catalog_registry=catalog_registry, system=CodeSystem.CUPS, code=cups, description=description, explicit=True, ) else: warnings.append("factura_sin_codigo_cups") crosswalk_status = str(crosswalk.get("status") or "") if crosswalk_status in {"ambiguous", "unmapped", "conflict"}: status = ( CurationStatus.CONFLICT if crosswalk_status == "conflict" else CurationStatus.PENDING_REVIEW ) warnings.append(f"crosswalk_{crosswalk_status}") results.append( ObjectiveProcedure( item_id=_stable_id("px", document.get("_id"), cups, billing, description, "facturado"), description=description, original_text=description, codigo_cups=cups, codigo_soat=explicit_soat, codigo_facturacion=billing, selected_soat_code=str(crosswalk.get("selected_soat_code") or ""), soat_candidates=soat_candidates, origin=CandidateOrigin.EXPLICIT, status=status, states=[ProcedureState.BILLED], evidence=[ _evidence_for_document( document, section="servicios_facturados", excerpt=description, page=int(document.get("source_page_start") or 0) or None, ) ], catalog=catalog, warnings=warnings, classification=classification, ) ) return results def _build_uncoded_procedures( document: dict[str, Any], state: ProcedureState, ) -> list[ObjectiveProcedure]: coded_descriptions = { _normalized_key(item.get("procedimiento") or item.get("descripcion")) for item in document.get("codigos_cups") or [] if isinstance(item, dict) } results: list[ObjectiveProcedure] = [] for raw in document.get("procedimientos_extraidos") or []: description = _clean_text(raw) if not description or _normalized_key(description) in coded_descriptions: continue results.append( ObjectiveProcedure( item_id=_stable_id("px", document.get("_id"), description, state), description=description, original_text=description, origin=CandidateOrigin.LLM, status=CurationStatus.PENDING_REVIEW, states=[state], evidence=[ _evidence_for_document( document, section="procedimientos", excerpt=description, page=int(document.get("source_page_start") or 0) or None, ) ], warnings=["procedimiento_sin_cups"], ) ) return results def _dedupe_diagnoses(items: Iterable[ObjectiveDiagnosis]) -> list[ObjectiveDiagnosis]: merged: dict[str, ObjectiveDiagnosis] = {} for item in items: key = f"{item.code}|{_normalized_key(item.description)}" if key not in merged: merged[key] = item continue current = merged[key] current.evidence.extend(e for e in item.evidence if e not in current.evidence) current.warnings = list(dict.fromkeys([*current.warnings, *item.warnings])) return list(merged.values()) def _dedupe_procedures(items: Iterable[ObjectiveProcedure]) -> list[ObjectiveProcedure]: merged: dict[str, ObjectiveProcedure] = {} for item in items: key = item.codigo_cups or f"{item.codigo_facturacion}|{_normalized_key(item.description)}" if key not in merged: merged[key] = item continue current = merged[key] current.states = list(dict.fromkeys([*current.states, *item.states])) current.evidence.extend(e for e in item.evidence if e not in current.evidence) current.warnings = list(dict.fromkeys([*current.warnings, *item.warnings])) current.codigo_facturacion = current.codigo_facturacion or item.codigo_facturacion current.codigo_soat = current.codigo_soat or item.codigo_soat known_candidates = {candidate.soat_code for candidate in current.soat_candidates} current.soat_candidates.extend( candidate for candidate in item.soat_candidates if candidate.soat_code not in known_candidates ) current.selected_soat_code = current.selected_soat_code or item.selected_soat_code current.selected_soat_valuation = current.selected_soat_valuation or item.selected_soat_valuation if ProcedureState.BILLED in item.states: current.description = item.description current.original_text = item.original_text current.codigo_facturacion = item.codigo_facturacion or current.codigo_facturacion current.classification = item.classification or current.classification elif current.classification == "sin_clasificar" and item.classification != "sin_clasificar": current.classification = item.classification procedures = list(merged.values()) for item in procedures: states = set(item.states) if item.classification == "no_quirurgico" and ProcedureState.BILLED in states: item.reconciliation_status = ProcedureReconciliationStatus.NON_SURGICAL_BILLED item.reconciliation_reason = "Procedimiento no quirúrgico facturado; no exige soporte quirúrgico." item.status = CurationStatus.AUTO_VALIDATED elif {ProcedureState.PERFORMED, ProcedureState.BILLED} <= states: item.reconciliation_status = ProcedureReconciliationStatus.PERFORMED_AND_BILLED item.reconciliation_reason = "El CUPS exacto aparece en factura y soporte quirúrgico." if item.status not in {CurationStatus.CONFLICT, CurationStatus.EXTERNAL_SERVICE_PENDING}: item.status = CurationStatus.AUTO_VALIDATED elif ProcedureState.PERFORMED in states: item.reconciliation_status = ProcedureReconciliationStatus.PERFORMED_NOT_BILLED item.reconciliation_reason = "El CUPS realizado no aparece facturado con el mismo código." item.status = CurationStatus.PENDING_REVIEW item.warnings = list(dict.fromkeys([*item.warnings, "posible_glosa_realizado_no_facturado"])) elif ProcedureState.BILLED in states and item.classification == "quirurgico": item.reconciliation_status = ProcedureReconciliationStatus.BILLED_WITHOUT_EXACT_SUPPORT item.reconciliation_reason = ( "El CUPS facturado no aparece en el soporte quirúrgico con el mismo código." ) item.status = CurationStatus.PENDING_REVIEW item.warnings = list( dict.fromkeys([*item.warnings, "posible_glosa_facturado_sin_soporte_exacto"]) ) return procedures def _objective_from_document(document: dict[str, Any]) -> ObjectiveData | None: payload = document.get("datos_objetivos") if not isinstance(payload, dict): return None try: return ObjectiveData.model_validate(payload) except ValueError: return None def reconcile_case_objective_data( documents: Iterable[dict[str, Any] | None], *, preserved_context: dict[str, Any] | None = None, ) -> CaseCuration | None: objective_documents = [ objective for document in documents if isinstance(document, dict) if (objective := _objective_from_document(document)) is not None ] if not objective_documents: return None diagnoses = _dedupe_diagnoses( diagnosis for objective in objective_documents for diagnosis in objective.diagnoses ) procedures = _dedupe_procedures( procedure for objective in objective_documents for procedure in objective.procedures ) postoperative_signals = [ signal for objective in objective_documents for signal in objective.postoperative_signals ] catalogs = _dedupe_catalogs( reference for objective in objective_documents for reference in objective.catalogs ) _restore_decisions(diagnoses, procedures, preserved_context or {}) conflicts = _detect_diagnosis_conflicts(diagnoses) conflict_ids = {item_id for conflict in conflicts for item_id in conflict.item_ids} for item in [*diagnoses, *procedures]: if item.item_id in conflict_ids and item.status not in { CurationStatus.APPROVED, CurationStatus.CORRECTED, CurationStatus.REJECTED, }: item.status = CurationStatus.CONFLICT version = _decision_aware_version( _case_version(objective_documents, catalogs), diagnoses, procedures, ) pending = [ item.item_id for item in [*diagnoses, *procedures] if item.status in { CurationStatus.PENDING_REVIEW, CurationStatus.EXTERNAL_SERVICE_PENDING, CurationStatus.CONFLICT, } ] blocking_conflict = next((conflict for conflict in conflicts if conflict.blocking), None) return CaseCuration( version=version, diagnoses=diagnoses, procedures=procedures, conflicts=conflicts, pending_item_ids=pending, catalogs=catalogs, postoperative_signals=postoperative_signals, generation_blocked=blocking_conflict is not None, blocking_reason=blocking_conflict.message if blocking_conflict else "", ) def _dedupe_catalogs(items: Iterable[CatalogReference]) -> list[CatalogReference]: result: dict[tuple[CodeSystem, str], CatalogReference] = {} for item in items: result[(item.system, item.fingerprint)] = item return list(result.values()) def _case_version( objectives: list[ObjectiveData], catalogs: list[CatalogReference], ) -> str: payload = { "schema": CURATION_SCHEMA_VERSION, "objectives": [dump_curation_model(item) for item in objectives], "catalogs": [dump_curation_model(item) for item in catalogs], } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(encoded.encode()).hexdigest() def _decision_aware_version( base_version: str, diagnoses: list[ObjectiveDiagnosis], procedures: list[ObjectiveProcedure], ) -> str: decisions = [ dump_curation_model(decision) for item in [*diagnoses, *procedures] for decision in item.decisions ] encoded = json.dumps( {"base": base_version, "decisions": decisions}, sort_keys=True, separators=(",", ":"), ensure_ascii=True, ) return hashlib.sha256(encoded.encode()).hexdigest() def _detect_diagnosis_conflicts(diagnoses: list[ObjectiveDiagnosis]) -> list[CodingConflict]: conflicts: list[CodingConflict] = [] for index, left in enumerate(diagnoses): for right in diagnoses[index + 1 :]: left_text = _ascii_upper(f"{left.original_text} {left.description}") right_text = _ascii_upper(f"{right.original_text} {right.description}") if not left.code or not right.code or left.code == right.code: continue if "HUMERO" not in left_text or "HUMERO" not in right_text: continue if not any( bool(left_terms & set(left_text.split())) and bool(right_terms & set(right_text.split())) or bool(left_terms & set(right_text.split())) and bool(right_terms & set(left_text.split())) for left_terms, right_terms in _LOCATION_GROUPS ): continue left.is_primary = True right.is_primary = True conflicts.append( CodingConflict( conflict_id=_stable_id("conflict", left.item_id, right.item_id), kind="anatomia_incompatible", severity="error", message=( f"Conflicto anatómico entre {left.code} ({left.description}) y " f"{right.code} ({right.description})." ), item_ids=[left.item_id, right.item_id], blocking=True, ) ) return conflicts def _restore_decisions( diagnoses: list[ObjectiveDiagnosis], procedures: list[ObjectiveProcedure], context: dict[str, Any], ) -> None: previous = {} for raw in [ *(context.get("diagnosticos_curados") or []), *(context.get("procedimientos_curados") or []), ]: if isinstance(raw, dict) and raw.get("item_id"): previous[str(raw["item_id"])] = raw for item in [*diagnoses, *procedures]: raw = previous.get(item.item_id) if not raw: continue item.decisions = [ CurationDecision.model_validate(decision) for decision in raw.get("decisions") or [] if isinstance(decision, dict) ] if item.decisions: item.status = CurationStatus(str(raw.get("status") or item.status)) if isinstance(item, ObjectiveProcedure): selected = _clean_text(raw.get("selected_soat_code")) if selected: item.selected_soat_code = selected item.codigo_soat = selected if isinstance(raw.get("selected_soat_valuation"), dict): item.selected_soat_valuation = SoatCurationValuation.model_validate( raw["selected_soat_valuation"] ) def apply_curation_decision( curation: CaseCuration, *, item_id: str, decision: CurationDecisionType, actor: str, expected_version: str, corrected_code: str = "", corrected_description: str = "", reason: str = "", selected_soat_code: str = "", crosswalk_version: str = "", tariff_catalog_version: str = "", source_reference: str = "", selected_soat_valuation: SoatCurationValuation | None = None, calculate_reference: bool = False, manual_soat_code: str = "", manual_surgical_group: int | None = None, selected_components: list[str] | None = None, ) -> CaseCuration: if expected_version != curation.version: raise ValueError("La curación cambió; recarga el caso antes de guardar la decisión.") item = next( ( candidate for candidate in [*curation.diagnoses, *curation.procedures] if candidate.item_id == item_id ), None, ) if item is None: raise LookupError("Elemento de curación no encontrado.") discrepancy = isinstance(item, ObjectiveProcedure) and item.reconciliation_status in { ProcedureReconciliationStatus.PERFORMED_NOT_BILLED, ProcedureReconciliationStatus.BILLED_WITHOUT_EXACT_SUPPORT, } if discrepancy and not _clean_text(reason): raise ValueError("La decisión sobre una discrepancia requiere un motivo.") if decision == CurationDecisionType.CORRECT and not ( corrected_code or corrected_description or selected_soat_code or manual_soat_code or manual_surgical_group ): raise ValueError("Una corrección requiere código, descripción o candidato SOAT seleccionado.") if selected_soat_code: if not isinstance(item, ObjectiveProcedure): raise ValueError("Solo un procedimiento puede seleccionar un candidato SOAT.") candidate_codes = {candidate.soat_code for candidate in item.soat_candidates} if selected_soat_code not in candidate_codes: raise ValueError( "El SOAT seleccionado no pertenece a los candidatos trazables del procedimiento." ) original_code = item.code if isinstance(item, ObjectiveDiagnosis) else item.codigo_cups record = CurationDecision( decision=decision, actor=actor, decided_at=datetime.now(UTC), original_code=original_code, original_description=item.description, corrected_code=corrected_code, corrected_description=corrected_description, reason=reason, expected_version=expected_version, selected_soat_code=selected_soat_code, crosswalk_version=crosswalk_version, tariff_catalog_version=tariff_catalog_version, source_reference=source_reference, calculate_reference=calculate_reference, manual_soat_code=manual_soat_code, manual_surgical_group=manual_surgical_group, selected_components=selected_components or [], ) item.decisions.append(record) if decision == CurationDecisionType.APPROVE: item.status = CurationStatus.APPROVED elif decision == CurationDecisionType.REJECT: item.status = CurationStatus.REJECTED else: item.status = CurationStatus.CORRECTED if corrected_description: item.description = _clean_text(corrected_description) if corrected_code: if isinstance(item, ObjectiveDiagnosis): item.code = _clean_text(corrected_code) else: item.codigo_cups = _clean_text(corrected_code) if isinstance(item, ObjectiveProcedure) and ( selected_soat_code or manual_soat_code or manual_surgical_group ): item.selected_soat_code = selected_soat_code or manual_soat_code item.codigo_soat = selected_soat_code or manual_soat_code item.selected_soat_valuation = selected_soat_valuation curation.pending_item_ids = [ candidate.item_id for candidate in [*curation.diagnoses, *curation.procedures] if candidate.status in { CurationStatus.PENDING_REVIEW, CurationStatus.EXTERNAL_SERVICE_PENDING, CurationStatus.CONFLICT, } ] unresolved_blocking = any( conflict.blocking and any( candidate.item_id in conflict.item_ids and candidate.status not in {CurationStatus.APPROVED, CurationStatus.CORRECTED, CurationStatus.REJECTED} for candidate in [*curation.diagnoses, *curation.procedures] ) for conflict in curation.conflicts ) curation.generation_blocked = unresolved_blocking curation.blocking_reason = ( next((conflict.message for conflict in curation.conflicts if conflict.blocking), "") if unresolved_blocking else "" ) curation.version = _decision_aware_version( expected_version, curation.diagnoses, curation.procedures, ) recalculate_reference_valuations(curation) return curation def recalculate_reference_valuations(curation: CaseCuration) -> None: """Liquida referencias separadas sin crear ni alterar líneas de factura.""" valued: list[tuple[ObjectiveProcedure, SoatCurationValuation]] = [] for item in curation.procedures: valuation = item.selected_soat_valuation if valuation is not None and valuation.base_tariff >= 0: valued.append((item, valuation)) if not valued: return maximum = max(valuation.base_tariff for _, valuation in valued) principals = [item for item, valuation in valued if valuation.base_tariff == maximum] if len(principals) != 1: error = "No existe un procedimiento principal tarifario único; revisa SOAT/grupo y componentes." for _, valuation in valued: valuation.status = "pendiente_revision" valuation.error = error valuation.liquidated_value = None valuation.liquidated_components = {} valuation.formula = "principal_tarifario_no_unico" return principal_id = principals[0].item_id multiple = len(valued) > 1 for item, valuation in valued: selected = set(valuation.selected_components or valuation.components) & _REFERENCE_COMPONENTS if item.item_id == principal_id or not multiple: liquidated = {name: value for name, value in valuation.components.items() if name in selected} formula = "principal_100_por_ciento_componentes_seleccionados" else: liquidated = { name: int( (Decimal(value) * Decimal("0.5") / Decimal("100")).quantize( Decimal("1"), rounding=ROUND_HALF_UP ) * 100 ) for name, value in valuation.components.items() if name in selected & _PROFESSIONAL_REFERENCE_COMPONENTS } formula = "secundario_50_por_ciento_componentes_profesionales_seleccionados" valuation.liquidated_components = liquidated valuation.liquidated_value = sum(liquidated.values()) valuation.formula = formula valuation.status = "liquidada" valuation.error = "" valuation.evidence = list( dict.fromkeys( [ *valuation.evidence, "Referencia SOAT separada; no crea líneas ni modifica el total facturado.", f"Regla determinista aplicada: {formula}.", ] ) ) def curation_context_payload(curation: CaseCuration) -> dict[str, Any]: publishable = {CurationStatus.AUTO_VALIDATED, CurationStatus.APPROVED, CurationStatus.CORRECTED} return { "curation_schema_version": curation.schema_version, "curation_version": curation.version, "diagnosticos_curados": [dump_curation_model(item) for item in curation.diagnoses], "procedimientos_curados": [dump_curation_model(item) for item in curation.procedures], "conflictos_curacion": [dump_curation_model(item) for item in curation.conflicts], "pendientes_curacion": list(curation.pending_item_ids), "catalogos_utilizados": [dump_curation_model(item) for item in curation.catalogs], "postoperative_signals": [dump_curation_model(item) for item in curation.postoperative_signals], "curation_generation_blocked": curation.generation_blocked, "curation_blocking_reason": curation.blocking_reason, "pdf_primary_diagnosticos": [ { "key": item.item_id, "codigo": item.code, "descripcion": item.description, "diagnostico": item.original_text or item.description, "estado_codificacion": item.status, "fuentes": [evidence.document_type for evidence in item.evidence], "tipo_contexto": item.context_type, "texto_presentacion": f"{item.code} - {item.description}" if item.code else item.description, } for item in curation.diagnoses if item.status in publishable ], "pdf_primary_procedimientos": [ { "key": item.item_id, "codigo_cups": item.codigo_cups, "descripcion": item.description, "clasificacion": item.classification, "clasificacion_automatica": item.classification, "clasificacion_manual": False, "soporte_clinico": ProcedureState.PERFORMED in item.states, "fuentes": [evidence.document_type for evidence in item.evidence], "estado_documental": [state.value for state in item.states], "evidencias": [dump_curation_model(evidence) for evidence in item.evidence], "advertencias": list(item.warnings), "estado_conciliacion": item.reconciliation_status, "razon_conciliacion": item.reconciliation_reason, } for item in curation.procedures if item.status in publishable and item.classification in {"quirurgico", "no_quirurgico"} and ( item.reconciliation_status in { ProcedureReconciliationStatus.PERFORMED_AND_BILLED, ProcedureReconciliationStatus.NON_SURGICAL_BILLED, } or item.status in {CurationStatus.APPROVED, CurationStatus.CORRECTED} ) ], } def curation_from_context(context: dict[str, Any]) -> CaseCuration | None: if not context.get("curation_schema_version") or not context.get("curation_version"): return None try: return CaseCuration.model_validate( { "schema_version": context.get("curation_schema_version"), "version": context.get("curation_version"), "diagnoses": context.get("diagnosticos_curados") or [], "procedures": context.get("procedimientos_curados") or [], "conflicts": context.get("conflictos_curacion") or [], "pending_item_ids": context.get("pendientes_curacion") or [], "catalogs": context.get("catalogos_utilizados") or [], "generation_blocked": bool(context.get("curation_generation_blocked")), "blocking_reason": context.get("curation_blocking_reason") or "", "postoperative_signals": context.get("postoperative_signals") or [], } ) except ValueError: return None