"""Helpers puros para orquestar la ingestión de PDFs compuestos de prefactura.""" from __future__ import annotations import re import unicodedata from collections.abc import Iterable, Sequence from dataclasses import dataclass from pathlib import Path from types import SimpleNamespace from typing import Any from app.batch_processing.application.models import BatchFileCreatePayload from app.batch_processing.application.utils import build_associated_user, build_case_key, now_iso, slugify from app.batch_processing.domain.models import ( PREFERRED_ASSOCIATION_MODE_VALIDATE_AND_WARN, PrefacturaDocumentSegment, PrefacturaPageClassification, ) from app.batch_processing.domain.ports import ( BatchFileRepository, BatchWorkingFileStore, CaseAssociationService, DocumentClassifier, PrefacturaClassificationAdvisor, PrefacturaPdfExtractor, ) from app.clinical_pipeline.domain.models import DocumentClassificationDecision DEFAULT_GENERAL_DOCUMENT_TITLE = "Documento general" PREFACTURA_SEGMENT_ORDER = ( "prefactura", "historia_clinica", "quirurgico", "radiologia", "laboratorio", ) PREFACTURA_ADDITIONAL_SEGMENT_ORDER = ( "quirurgico", "radiologia", "laboratorio", ) PREFACTURA_SEGMENT_TITLES = { "prefactura": "Prefactura", "historia_clinica": "Historia clínica", "quirurgico": "Documento quirúrgico", "radiologia": "Radiología", "laboratorio": "Laboratorio", } _ADVISOR_ACCEPTANCE_THRESHOLD = 0.60 _ADVISOR_CANDIDATE_TYPES = ["historia_clinica", "quirurgico", "laboratorio", "radiologia"] @dataclass(frozen=True) class PrefacturaUploadContext: created_at: str preferred_case_key: str prefactura_case_number: str patient_name: str patient_id: str review_required: bool review_messages: list[str] @dataclass(frozen=True) class PrefacturaCompoundOrchestrator: """Orquesta la clasificación determinista y el fallback Gemini para prefacturas compuestas.""" document_classifier: DocumentClassifier association_service: CaseAssociationService classification_advisor: PrefacturaClassificationAdvisor | None = None def build_segments( self, *, pages: Sequence[Any], ) -> list[PrefacturaDocumentSegment]: classifications = self.classify_pages(pages=pages) if not classifications: return [] segments: list[PrefacturaDocumentSegment] = [] prefactura_classifications = [item for item in classifications if item.document_type == "prefactura"] if not prefactura_classifications: prefactura_classifications = [classifications[0]] prefactura_page_numbers = {item.page_number for item in prefactura_classifications} prefactura_segment = build_prefactura_segment_from_classifications( document_type="prefactura", classifications=prefactura_classifications, ) if prefactura_segment is not None: segments.append(prefactura_segment) clinical_classifications = [ item for item in classifications if item.page_number not in prefactura_page_numbers ] clinical_segment = build_prefactura_segment_from_classifications( document_type="historia_clinica", classifications=clinical_classifications, ) if clinical_segment is not None: segments.append(clinical_segment) for document_type in PREFACTURA_ADDITIONAL_SEGMENT_ORDER: additional_classifications = [ item for item in clinical_classifications if item.document_type == document_type and bool(item.is_autonomous) ] additional_segment = build_prefactura_segment_from_classifications( document_type=document_type, classifications=additional_classifications, ) if additional_segment is not None: segments.append(additional_segment) return segments def classify_pages(self, *, pages: Sequence[Any]) -> list[PrefacturaPageClassification]: classifications: list[PrefacturaPageClassification] = [] for page in pages: classifications.append(self.classify_page(page=page, total_pages=len(pages))) return classifications def classify_page( self, *, page: Any, total_pages: int, ) -> PrefacturaPageClassification: page_number = int(getattr(page, "page_number", 0) or 0) page_text = str(getattr(page, "text", "") or "") detected_type, classifier_title = self.document_classifier.describe( f"pagina_{page_number}.pdf", page_text, ) if page_number <= 1: return PrefacturaPageClassification( page_number=page_number, document_type="prefactura", document_title=macrofamily_title("prefactura"), confidence=1.0, evidence=_build_page_evidence( detected_type=detected_type, classifier_title=classifier_title, text=page_text, document_type="prefactura", page_number=page_number, ), reason="primera_pagina_anclada_como_prefactura", source="deterministic", is_autonomous=False, page_text=page_text, ) document_type = collapse_prefactura_macrofamily( detected_type=detected_type, document_title=classifier_title, text=page_text, ) is_autonomous = should_materialize_prefactura_additional_document( detected_type=detected_type, document_title=classifier_title, text=page_text, ) confidence = _deterministic_confidence( document_type=document_type, detected_type=detected_type, classifier_title=classifier_title, text=page_text, is_autonomous=is_autonomous, ) evidence = _build_page_evidence( detected_type=detected_type, classifier_title=classifier_title, text=page_text, document_type=document_type, page_number=page_number, ) reason = _deterministic_reason( document_type=document_type, detected_type=detected_type, classifier_title=classifier_title, ) deterministic_result = PrefacturaPageClassification( page_number=page_number, document_type=document_type, document_title=macrofamily_title(document_type), confidence=confidence, evidence=evidence, reason=reason, source="deterministic", is_autonomous=is_autonomous, page_text=page_text, ) if not self._should_escalate(deterministic_result, detected_type=detected_type, classifier_title=classifier_title, text=page_text): return deterministic_result advisor_result = self.classification_advisor.classify_page( filename=f"pagina_{page_number}.pdf", page_number=page_number, page_text=page_text, detected_type=detected_type, classifier_title=classifier_title, candidate_types=list(_ADVISOR_CANDIDATE_TYPES), ) if self.classification_advisor is not None else None if advisor_result is None: return deterministic_result if advisor_result.confidence < _ADVISOR_ACCEPTANCE_THRESHOLD: return deterministic_result return advisor_result def build_upload_context( self, *, filename: str, segments: Sequence[PrefacturaDocumentSegment], colombia_tz: Any, ) -> PrefacturaUploadContext: anchor_identity = resolve_prefactura_identity( segments=segments, association_service=self.association_service, ) prefactura_case_number = anchor_identity["case_number"] return PrefacturaUploadContext( created_at=now_iso(colombia_tz), preferred_case_key=build_case_key( patient_id="", case_number=prefactura_case_number, patient_name="", fallback_name=filename, ), prefactura_case_number=prefactura_case_number, patient_name=anchor_identity["patient_name"], patient_id=anchor_identity["patient_id"], review_required=not bool(prefactura_case_number), review_messages=( ["No se identificó número de caso en la prefactura. Requiere validación manual."] if not prefactura_case_number else [] ), ) def _should_escalate( self, result: PrefacturaPageClassification, *, detected_type: str, classifier_title: str, text: str, ) -> bool: if result.document_type == "prefactura": return False if result.confidence >= _ADVISOR_ACCEPTANCE_THRESHOLD: return False if result.document_type in {"quirurgico", "laboratorio", "radiologia"} and result.is_autonomous: return False normalized_title = normalize_prefactura_value(classifier_title) normalized_text = normalize_prefactura_value(text) if detected_type in {"generico", "prescripcion"}: return True if result.document_type == "historia_clinica": return not any( marker in normalized_title or marker in normalized_text for marker in ( "historia clinica", "evolucion", "nota", "urgencias", "curaciones", "observacion", "observacion medica", "admision", ) ) return True def build_prefactura_segments( *, pages: Sequence[Any], document_classifier: DocumentClassifier, classification_advisor: PrefacturaClassificationAdvisor | None = None, association_service: CaseAssociationService | None = None, ) -> list[PrefacturaDocumentSegment]: orchestrator = PrefacturaCompoundOrchestrator( document_classifier=document_classifier, association_service=association_service or _NullCaseAssociationService(), classification_advisor=classification_advisor, ) return orchestrator.build_segments(pages=pages) def build_prefactura_segment_from_classifications( *, document_type: str, classifications: Sequence[PrefacturaPageClassification], ) -> PrefacturaDocumentSegment | None: if not classifications: return None ordered_classifications = sorted(classifications, key=lambda item: int(item.page_number)) page_numbers = [int(item.page_number) for item in ordered_classifications] text = "\n".join(str(item.page_text or "") for item in ordered_classifications).strip() return PrefacturaDocumentSegment( document_type=document_type, document_title=macrofamily_title(document_type), document_key=document_type, document_reference="", page_start=page_numbers[0], page_end=page_numbers[-1], page_numbers=page_numbers, text=text, ) def resolve_prefactura_identity( *, segments: Sequence[PrefacturaDocumentSegment], association_service: CaseAssociationService, ) -> dict[str, str]: resolved_case_number = "" resolved_patient_name = "" resolved_patient_id = "" ordered = sorted(segments, key=lambda item: 0 if item.document_type == "prefactura" else 1) for segment in ordered: signals = association_service.extract_signals("", segment.text, segment.document_type) if not resolved_case_number: resolved_case_number = str(signals.case_number or "").strip() if not resolved_patient_name: resolved_patient_name = str(signals.patient_name or "").strip() if not resolved_patient_id: resolved_patient_id = str(signals.patient_id or "").strip() if resolved_case_number and resolved_patient_name and resolved_patient_id: break return { "case_number": resolved_case_number, "patient_name": resolved_patient_name, "patient_id": resolved_patient_id, } def build_prefactura_upload_context( *, filename: str, segments: Sequence[PrefacturaDocumentSegment], association_service: CaseAssociationService, colombia_tz: Any, ) -> PrefacturaUploadContext: orchestrator = PrefacturaCompoundOrchestrator( document_classifier=_NullDocumentClassifier(), association_service=association_service, ) return orchestrator.build_upload_context( filename=filename, segments=segments, colombia_tz=colombia_tz, ) def persist_prefactura_segments( *, batch_id: str, normalized_filename: str, contents: bytes, segments: Sequence[PrefacturaDocumentSegment], upload_context: PrefacturaUploadContext, prefactura_pdf_extractor: PrefacturaPdfExtractor, working_file_store: BatchWorkingFileStore, batch_file_repository: BatchFileRepository, association_service: CaseAssociationService, ) -> None: for index, segment in enumerate(segments, start=1): page_numbers = segment.page_numbers or list(range(segment.page_start, segment.page_end + 1)) range_pdf = prefactura_pdf_extractor.build_pages_pdf(contents, page_numbers) segment_filename = build_prefactura_segment_filename( original_filename=normalized_filename, index=index, segment=segment, ) stored_path = working_file_store.save_file(batch_id, segment_filename, range_pdf) signals = association_service.extract_signals("", segment.text, segment.document_type) case_number = str(signals.case_number or upload_context.prefactura_case_number).strip() patient_name = str(signals.patient_name or upload_context.patient_name or "").strip() patient_id = str(signals.patient_id or upload_context.patient_id or "").strip() evidence = list(signals.evidence or []) if segment.document_type == "prefactura" and case_number and not patient_id: evidence.append("prefactura_sin_identificacion") if upload_context.prefactura_case_number: evidence.append("prefactura_numero_caso_ancla") batch_file_repository.create_file( BatchFileCreatePayload( batch_id=batch_id, original_name=segment_filename, relative_path=segment_filename, stored_path=stored_path, status="pendiente", detected_type=segment.document_type, patient_name=patient_name, patient_id=patient_id, case_number=case_number, service_date=str(signals.service_date or "").strip(), procedure_code=str(signals.procedure_code or "").strip(), procedure_description=str(signals.procedure_description or "").strip(), redacted_identity_fields=list(getattr(signals, "redacted_identity_fields", []) or []), associated_user=build_associated_user(patient_id, patient_name), document_title=segment.document_title, document_key=segment.document_key, document_reference=segment.document_reference, preferred_case_key=upload_context.preferred_case_key if upload_context.prefactura_case_number else "", preferred_case_number=upload_context.prefactura_case_number, preferred_patient_id=upload_context.patient_id, preferred_patient_name=upload_context.patient_name, preferred_association_mode=PREFERRED_ASSOCIATION_MODE_VALIDATE_AND_WARN, evidence=unique_text_values(evidence), text_preview=segment.text[:2500], extracted_text=segment.text, review_required=upload_context.review_required, review_messages=upload_context.review_messages, parent_prefactura_batch_id=batch_id, source_page_start=segment.page_start, source_page_end=segment.page_end, created_at=upload_context.created_at, updated_at=upload_context.created_at, ).to_document() ) def build_prefactura_segment_filename( *, original_filename: str, index: int, segment: PrefacturaDocumentSegment, ) -> str: stem = Path(original_filename).stem or "prefactura" segment_label = slugify(segment.document_title or segment.document_type) or segment.document_type return f"{stem}_{index:02d}_{segment_label}_p{segment.page_start}-{segment.page_end}.pdf" def unique_text_values(values: Iterable[Any]) -> list[str]: seen: set[str] = set() result: list[str] = [] for value in values: normalized = str(value or "").strip() if not normalized or normalized in seen: continue seen.add(normalized) result.append(normalized) return result def describe_prefactura_page(*, page: Any, document_classifier: DocumentClassifier) -> dict[str, Any]: page_text = str(page.text or "") detected_type, classifier_title = document_classifier.describe( f"pagina_{page.page_number}.pdf", page_text, ) document_type = collapse_prefactura_macrofamily( detected_type=detected_type, document_title=classifier_title, text=page_text, ) return { "page_number": int(page.page_number), "document_type": document_type, "document_title": macrofamily_title(document_type), "page_text": page_text, "document_key": document_type, "is_autonomous": should_materialize_prefactura_additional_document( detected_type=detected_type, document_title=classifier_title, text=page_text, ), } def collapse_prefactura_macrofamily( *, detected_type: str, document_title: str, text: str, ) -> str: normalized_title = normalize_prefactura_value(document_title) normalized_text = normalize_prefactura_value(text) if detected_type == "prefactura" or "prefactura" in normalized_text or "extracto de cuenta" in normalized_text: return "prefactura" if detected_type == "quirurgico" and is_real_quirurgico_content(normalized_title, normalized_text): return "quirurgico" if detected_type == "laboratorio" and is_real_laboratory_content(normalized_title, normalized_text): return "laboratorio" if detected_type == "radiologia" and is_real_radiology_content(normalized_title, normalized_text): return "radiologia" return "historia_clinica" def should_materialize_prefactura_additional_document( *, detected_type: str, document_title: str, text: str, ) -> bool: normalized_title = normalize_prefactura_value(document_title) normalized_text = normalize_prefactura_value(text) if detected_type == "quirurgico": return is_real_quirurgico_content(normalized_title, normalized_text) if detected_type == "radiologia": return is_real_radiology_content(normalized_title, normalized_text) if detected_type == "laboratorio": return is_real_laboratory_content(normalized_title, normalized_text) return False def is_real_radiology_content(normalized_title: str, normalized_text: str) -> bool: if "ordenes de paraclinicos" in normalized_text: return False return any( marker in normalized_title or marker in normalized_text for marker in ("resultado radiologia", "informe radiologia", "lectura radiologia", "radiografia") ) def is_real_laboratory_content(normalized_title: str, normalized_text: str) -> bool: if "ordenes de paraclinicos" in normalized_text: return False return any( marker in normalized_title or marker in normalized_text for marker in ("resultado de laboratorio", "resultado laboratorio", "hemograma", "bioquimica") ) def is_real_quirurgico_content(normalized_title: str, normalized_text: str) -> bool: return any( marker in normalized_title or marker in normalized_text for marker in ( "descripcion quirurgica", "descripción quirúrgica", "reporte quirurgico", "reporte quirúrgico", "nota quirurgica", "nota quirúrgica", "procedimiento quirurgico", "procedimiento quirúrgico", ) ) def macrofamily_title(document_type: str) -> str: return PREFACTURA_SEGMENT_TITLES.get(document_type, DEFAULT_GENERAL_DOCUMENT_TITLE) def normalize_prefactura_value(value: str) -> str: normalized = unicodedata.normalize("NFKD", str(value or "")) normalized = normalized.encode("ascii", "ignore").decode("ascii").lower() normalized = re.sub(r"\s+", " ", normalized) return normalized.strip() def _build_page_evidence( *, detected_type: str, classifier_title: str, text: str, document_type: str, page_number: int, ) -> list[str]: evidence: list[str] = [f"pagina_{page_number}"] normalized_title = normalize_prefactura_value(classifier_title) normalized_text = normalize_prefactura_value(text) if "prefactura" in normalized_text or "extracto de cuenta" in normalized_text: evidence.append("marcador_prefactura") if "historia clinica" in normalized_text or "historia clinica" in normalized_title: evidence.append("marcador_historia_clinica") if detected_type in {"quirurgico", "laboratorio", "radiologia"}: evidence.append(f"clasificador_{detected_type}") if document_type in {"quirurgico", "laboratorio", "radiologia"}: evidence.append("documento_autonomo") if detected_type == "generico": evidence.append("clasificador_generico") return unique_text_values(evidence) def _deterministic_confidence( *, document_type: str, detected_type: str, classifier_title: str, text: str, is_autonomous: bool, ) -> float: normalized_title = normalize_prefactura_value(classifier_title) normalized_text = normalize_prefactura_value(text) if document_type == "prefactura": return 1.0 if is_autonomous: return 0.93 if document_type == "historia_clinica": if detected_type == "historia_clinica": return 0.82 if any(marker in normalized_title or marker in normalized_text for marker in ("nota", "evolucion", "urgencias", "curaciones")): return 0.74 return 0.56 return 0.52 def _deterministic_reason( *, document_type: str, detected_type: str, classifier_title: str, ) -> str: normalized_title = normalize_prefactura_value(classifier_title) if document_type == "prefactura": return "primera_pagina_anclada_como_prefactura" if detected_type == document_type: return f"coincide_con_clasificador_{detected_type}" if document_type == "historia_clinica" and normalized_title: return f"consolida_en_historia_clinica_desde_{detected_type}" if document_type in {"quirurgico", "laboratorio", "radiologia"}: return f"segmento_autonomo_{document_type}" return "clasificacion_deterministica" class _NullDocumentClassifier: def describe(self, filename: str, text: str) -> tuple[str, str]: _ = filename, text return "generico", "Documento general" def classify(self, filename: str, text: str) -> str: _ = filename, text return "generico" def inspect(self, filename: str, text: str) -> DocumentClassificationDecision: _ = filename, text return DocumentClassificationDecision( document_type="generico", title=DEFAULT_GENERAL_DOCUMENT_TITLE, confidence=0.0, reasons=["null_classifier"], ) class _NullCaseAssociationService: def extract_signals(self, filename: str, text: str, detected_type: str) -> Any: _ = filename, text, detected_type return SimpleNamespace( patient_name="", patient_id="", case_number="", service_date="", procedure_code="", procedure_description="", evidence=[], ) def associate(self, file_records: list[dict]) -> Any: _ = file_records return SimpleNamespace(decisions=[], cases=[])