from __future__ import annotations import re from dataclasses import dataclass from typing import Any from app.llm.schemas import ( ClinicalStructuredModel, DocumentoQuirurgicoStructured, FacturaStructured, HistoriaClinicaStructured, ) from app.services.deterministic_signals import ( extract_deterministic_signals, resolve_postoperative_procedure, ) from app.services.document_identity_extraction import extract_document_identity from app.services.historia_antecedentes import ( extract_historia_antecedentes_structured, resolve_historia_antecedentes, ) from app.services.historia_recomendaciones import has_concrete_historia_recommendations from app.services.historia_summary import calculate_historia_summary_policy, is_valid_historia_resumen from app.services.patient_name_extraction import sanitize_patient_name_candidate _DATE_PATTERN = re.compile(r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b") _DATE_TIME_PATTERN = re.compile( r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}(?:\s+\d{1,2}:\d{2})?\b|\b\d{1,2}\s+[A-Za-záéíóúñ]+,\s+\d{4}\b", re.IGNORECASE, ) _MONEY_PATTERN = re.compile(r"^\$?\s*[\d.,]+$") _TABLE_SIGNAL_PATTERN = re.compile(r"\b(cod\.?|cups-cum|cant\.?|precio|total)\b", re.IGNORECASE) _FACTURA_FIELD_LABEL_PATTERN = re.compile( r"\b(prefijo|factura|caso|vencimiento|convenio|fecha ingreso|fecha egreso|paciente|señores)\b", re.IGNORECASE, ) _QX_SIGNAL_PATTERN = re.compile( r"\b(diagn[oó]stico|procedimiento|hallazgo|complicaci[oó]n|conclusi[oó]n)\b", re.IGNORECASE, ) _HISTORIA_SECTION_PATTERN = re.compile( r"\b(diagn[oó]stic|procedimient|medicament|motivo de consulta|enfermedad actual)\b", re.IGNORECASE, ) _HISTORIA_ANTECEDENT_SECTION_PATTERN = re.compile( r"(?im)^\s*ANTECEDENTES(?:\s+(?:PERSONALES|CL[IÍ]NICOS|M[EÉ]DICOS))?\s*$" ) _HISTORIA_ALLERGY_PATTERN = re.compile(r"\b(?:ALERGIAS?|AL[EÉ]RGICOS?)\s*:", re.IGNORECASE) @dataclass(frozen=True, slots=True) class ClinicalExtractionQualityAssessment: status: str low_confidence: bool should_retry: bool reason_codes: tuple[str, ...] validator_version: str = "v5" def to_payload(self) -> dict[str, Any]: return { "status": self.status, "low_confidence": self.low_confidence, "should_retry": self.should_retry, "reason_codes": list(self.reason_codes), "validator_version": self.validator_version, } def assess_clinical_extraction( *, document_type: str, raw_text: str, analysis_model: ClinicalStructuredModel, ) -> ClinicalExtractionQualityAssessment: reasons = _collect_quality_reasons( document_type=str(document_type or "").strip().lower(), raw_text=str(raw_text or ""), analysis_model=analysis_model, ) low_confidence = bool(reasons) return ClinicalExtractionQualityAssessment( status="degraded" if low_confidence else "ok", low_confidence=low_confidence, should_retry=low_confidence, reason_codes=tuple(reasons), ) def _collect_quality_reasons( *, document_type: str, raw_text: str, analysis_model: ClinicalStructuredModel, ) -> list[str]: if document_type == "factura" and isinstance(analysis_model, FacturaStructured): return _factura_quality_reasons(raw_text, analysis_model) if document_type == "historia_clinica" and isinstance(analysis_model, HistoriaClinicaStructured): return _historia_quality_reasons(raw_text, analysis_model) if document_type == "quirurgico" and isinstance(analysis_model, DocumentoQuirurgicoStructured): return _quirurgico_quality_reasons(raw_text, analysis_model) return [] def _factura_quality_reasons(raw_text: str, model: FacturaStructured) -> list[str]: reasons: list[str] = [] patient_name = str(model.patient_name or "").strip() payer_name = str(model.pagador.aseguradora_eps or "").strip() provider_name = str(model.proveedor.nombre_institucion or "").strip() if not _is_valid_person_name(patient_name): reasons.append("identity_patient_name_invalid") if patient_name and patient_name in {payer_name, provider_name}: reasons.append("identity_patient_name_conflict") if str(model.paciente.numero_identificacion or "").strip() and not str( model.paciente.numero_identificacion or "" ).strip().replace(".", "").isdigit(): reasons.append("identity_patient_id_invalid") for field_name, value in { "fecha_emision": model.datos_factura.fecha_emision, "fecha_vencimiento": model.datos_factura.fecha_vencimiento, "fecha_ingreso": model.paciente.fecha_ingreso, "fecha_egreso": model.paciente.fecha_egreso, }.items(): if _looks_contaminated_date(value): reasons.append(f"{field_name}_contaminated") for field_name, value in { "total_servicios": model.resumen_financiero.total_servicios, "descuentos": model.resumen_financiero.descuentos, "copagos": model.resumen_financiero.copagos, "valor_total_factura": model.resumen_financiero.valor_total_factura, }.items(): if value and not _looks_money(value): reasons.append(f"{field_name}_invalid_money") has_table_signals = bool(_TABLE_SIGNAL_PATTERN.search(raw_text)) if has_table_signals and not model.lineas_canonicas: reasons.append("missing_canonical_lines") return reasons def _historia_quality_reasons(raw_text: str, model: HistoriaClinicaStructured) -> list[str]: reasons: list[str] = [] identity = extract_document_identity(raw_text, strategy="historia_clinica") if "patient_name" not in identity.redacted_identity_fields and not _is_valid_person_name( model.patient_name ): reasons.append("identity_patient_name_invalid") summary = str(model.resumen_clinico or "") if not is_valid_historia_resumen(summary): reasons.append("summary_low_quality") policy = calculate_historia_summary_policy(raw_text) summary_words = len(summary.split()) if policy.is_long and summary_words < policy.acceptable_min_words: reasons.append("summary_too_short_for_source") if policy.is_long and summary_words > policy.acceptable_max_words: reasons.append("summary_too_long_for_policy") if re.search(r"(?:^|\s)(?:y|con|de|para|o|e)[,.!?;:]?\s*$", summary, re.IGNORECASE): reasons.append("summary_incomplete_sentence") topic_patterns = { "presentation": r"motivo|consulta|ingresa|presenta|refiere|dolor", "diagnosis": r"diagn[oó]st|hallaz|fractura|lesi[oó]n", "diagnostics": r"laboratorio|radiograf|tomograf|resonancia|ecograf|ayuda diagn", "treatment": r"proced|cirug|tratamiento|manejo|medic|terapia", "evolution": r"evoluci[oó]n|egreso|alta|plan|seguimiento", } if policy.is_long: missing_topics = [ topic for topic, pattern in topic_patterns.items() if re.search(pattern, raw_text, re.IGNORECASE) and not re.search(pattern, summary, re.IGNORECASE) ] if len(missing_topics) >= 2: reasons.append("summary_missing_clinical_topics") for identity_value in (identity.patient_name, identity.patient_id, identity.case_number): if identity_value and identity_value.casefold() in summary.casefold(): reasons.append("summary_redundant_identity") break if _HISTORIA_SECTION_PATTERN.search(raw_text): if "diagnost" in raw_text.casefold() and not model.diagnosticos: reasons.append("diagnosticos_missing") if "proced" in raw_text.casefold() and not model.procedimientos: reasons.append("procedimientos_missing") if "medic" in raw_text.casefold() and not model.medicamentos: reasons.append("medicamentos_missing") reasons.extend(_postoperative_quality_reasons(raw_text, model.procedimientos, model.procedimientos_antecedentes)) antecedent_resolution = resolve_historia_antecedentes( raw_text, structured_candidates=model.antecedentes_estructurados, procedure_candidates=model.procedimientos_antecedentes, legacy_flat_candidates=model.antecedentes, ) reasons.extend(antecedent_resolution.reason_codes) if any(item.estado == "negado" for item in model.antecedentes_estructurados): reasons.append("antecedents_negative_not_allowed") source_antecedents = extract_historia_antecedentes_structured(raw_text) positive_source_antecedents = [ item for item in source_antecedents if item.estado == "presente" ] if positive_source_antecedents and not antecedent_resolution.items: reasons.append("antecedentes_missing") if any(item.categoria == "alergico" for item in positive_source_antecedents) and not any( item.categoria == "alergico" for item in antecedent_resolution.items ): reasons.append("alergias_missing") has_extracted_recommendations = any( str(item.indicacion or item.evidencia or "").strip() for item in model.recomendaciones_medicas ) if has_concrete_historia_recommendations(raw_text) and not has_extracted_recommendations: reasons.append("recomendaciones_medicas_missing") return list(dict.fromkeys(reasons)) def _quirurgico_quality_reasons(raw_text: str, model: DocumentoQuirurgicoStructured) -> list[str]: reasons: list[str] = [] if not _is_valid_person_name(model.patient_name): reasons.append("identity_patient_name_invalid") if not str(model.resumen_clinico or "").strip(): reasons.append("summary_missing") if _QX_SIGNAL_PATTERN.search(raw_text): if not model.diagnosticos: reasons.append("diagnosticos_missing") if not model.procedimientos: reasons.append("procedimientos_missing") if not str(model.hallazgos or "").strip(): reasons.append("hallazgos_missing") if not str(model.descripcion_procedimiento or model.procedimiento_principal or "").strip(): reasons.append("descripcion_missing") reasons.extend(_postoperative_quality_reasons(raw_text, model.procedimientos, [])) return reasons def _postoperative_quality_reasons( raw_text: str, current_procedures: list[Any], historical_procedures: list[Any], ) -> list[str]: signals = extract_deterministic_signals(raw_text, "historia_clinica").postoperative_signals if not signals: return [] if not current_procedures: return [ "postoperatorio_procedure_only_historical" if historical_procedures else "postoperatorio_procedure_missing" ] snapshot = extract_deterministic_signals(raw_text, "historia_clinica") unresolved = [] for signal in signals: context = resolve_postoperative_procedure(signal, snapshot.procedure_contexts) if context is None: unresolved.append("postoperatorio_procedure_context_unresolved") continue context_text = str(context.get("description") or "").casefold() context_code = str(context.get("codigo_cups") or "").strip() if not any( ( context_code and context_code == str(getattr(procedure, "codigo", "") or "").strip() ) or ( context_text and ( context_text in str(getattr(procedure, "descripcion", "") or "").casefold() or str(getattr(procedure, "descripcion", "") or "").casefold() in context_text ) ) for procedure in current_procedures ): unresolved.append("postoperatorio_procedure_missing") return list(dict.fromkeys(unresolved)) def _is_valid_person_name(value: Any) -> bool: sanitized = sanitize_patient_name_candidate(value) return bool(sanitized and sanitized.lower() != "desconocido") def _looks_contaminated_date(value: Any) -> bool: text = str(value or "").strip() if not text: return False if not _DATE_TIME_PATTERN.search(text): return True if _FACTURA_FIELD_LABEL_PATTERN.search(text): return True return len(text.split()) > 8 and not _DATE_PATTERN.fullmatch(text) def _looks_money(value: Any) -> bool: text = str(value or "").strip() if not text: return False return bool(_MONEY_PATTERN.fullmatch(text.replace(" ", "")))