"""Helpers puros para reducir complejidad accidental en casos de uso batch.""" from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass from typing import Any from app.batch_processing.application.models import BatchFileUpdatePayload, BatchUploadAcceptedView from app.batch_processing.application.utils import ( build_associated_user, build_case_key, normalize_case_key_list, ) from app.batch_processing.domain.models import ( BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, BATCH_STATUS_RECIBIDO, CLINICAL_STATUS_COMPLETADO, EPICRISIS_STATUS_COMPLETADO, EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PENDIENTE, EPICRISIS_STATUS_PROCESANDO, FILE_STATUS_ASOCIADO, FILE_STATUS_CLASIFICADO, FILE_STATUS_FALLIDO, FILE_STATUS_PENDIENTE_VALIDACION, ) from app.services.clinical_document_service import ClinicalDocumentRequest def validate_zip_upload(*, filename: str, contents: bytes) -> None: if not filename or not filename.lower().endswith(".zip"): raise ValueError("Solo se permiten archivos ZIP.") if not contents: raise ValueError("El archivo ZIP está vacío.") def validate_pdf_upload(*, filename: str, contents: bytes, empty_message: str, invalid_message: str) -> str: normalized_filename = str(filename or "").strip() if not normalized_filename or not normalized_filename.lower().endswith(".pdf"): raise ValueError(invalid_message) if not contents: raise ValueError(empty_message) return normalized_filename def build_upload_accepted_response( *, batch_repository: Any, batch_id: str, created_at: str, ) -> dict[str, Any]: batch = batch_repository.get_batch(batch_id) or {} return BatchUploadAcceptedView( batch_id=batch_id, status=str(batch.get("status") or BATCH_STATUS_RECIBIDO), created_at=str(batch.get("created_at") or created_at), poll_url=f"/api/lotes/{batch_id}", ).to_document() def resolve_case_context( *, case_context_resolver: Callable[[str, str], Any] | None, username: str, case_key: str, ) -> dict[str, Any]: if not case_key or case_context_resolver is None: return {} context = case_context_resolver(username, case_key) return dict(context or {}) def merge_unique_values(values: list[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 should_skip_file_processing(*, file_record: dict[str, Any] | None, batch_id: str) -> bool: if not file_record or file_record.get("batch_id") != batch_id: return True current_status = file_record.get("status") if current_status == FILE_STATUS_FALLIDO and file_record.get("error"): return True return current_status in { FILE_STATUS_CLASIFICADO, FILE_STATUS_ASOCIADO, FILE_STATUS_PENDIENTE_VALIDACION, } and bool(file_record.get("text_preview")) def resolve_batch_file_detected_type( *, file_record: dict[str, Any], extracted_text: str, document_classifier: Any, ) -> str: existing_detected_type = str(file_record.get("detected_type") or "").strip() is_prefactura_segment = bool(file_record.get("parent_prefactura_batch_id")) if is_prefactura_segment and existing_detected_type: return existing_detected_type return document_classifier.classify(file_record.get("original_name", ""), extracted_text) def build_batch_file_processing_payload( *, file_record: dict[str, Any], extracted_text: str, document_classifier: Any, association_service: Any, updated_at: str, extraction_metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: is_prefactura_segment = bool(file_record.get("parent_prefactura_batch_id")) detected_type = resolve_batch_file_detected_type( file_record=file_record, extracted_text=extracted_text, document_classifier=document_classifier, ) if not extracted_text.strip(): return BatchFileUpdatePayload( status=FILE_STATUS_PENDIENTE_VALIDACION, detected_type=detected_type, text_preview="", extracted_text="", extraction_metadata=dict(extraction_metadata or {}), evidence=["texto_no_extraible: ReadPdf.read no extrajo contenido textual"], review_required=True, review_messages=["El PDF no tiene texto extraíble. Requiere revisión manual."], error="texto_no_extraible", updated_at=updated_at, ).to_document() signals = association_service.extract_signals( "" if is_prefactura_segment else file_record.get("original_name", ""), extracted_text, detected_type, ) patient_name = signals.patient_name or str(file_record.get("patient_name") or "").strip() patient_id = signals.patient_id or str(file_record.get("patient_id") or "").strip() return BatchFileUpdatePayload( status=FILE_STATUS_CLASIFICADO, detected_type=detected_type, text_preview=extracted_text[:2500], extracted_text=extracted_text, extraction_metadata=dict(extraction_metadata or {}), patient_name=patient_name, patient_id=patient_id, case_number=signals.case_number or str(file_record.get("case_number") or "").strip(), service_date=signals.service_date, procedure_code=signals.procedure_code, procedure_description=signals.procedure_description, redacted_identity_fields=list(getattr(signals, "redacted_identity_fields", []) or []), associated_user=build_associated_user(patient_id, patient_name), association_source="auto", evidence=merge_unique_values([*list(file_record.get("evidence") or []), *list(signals.evidence)]), score_breakdown={}, top_candidates=[], manual_resolution={}, review_required=bool(file_record.get("review_required", False)), review_messages=merge_unique_values(list(file_record.get("review_messages") or [])), error="", updated_at=updated_at, ).to_document() def build_clinical_document_request( *, batch: dict[str, Any], file_record: dict[str, Any], batch_id: str, file_id: str, ) -> ClinicalDocumentRequest: return ClinicalDocumentRequest( raw_text=file_record.get("extracted_text", "") or file_record.get("text_preview", ""), detected_type=file_record.get("detected_type", "generico"), username=batch.get("usuario", ""), original_name=file_record.get("original_name", ""), case_key=file_record.get("case_key", ""), case_number=file_record.get("case_number", ""), patient_id=file_record.get("patient_id", ""), provided_patient_name=file_record.get("patient_name", ""), provided_patient_id=file_record.get("patient_id", ""), provided_case_number=file_record.get("case_number", ""), case_resolution_status="provisional" if bool(file_record.get("review_required", False)) else "confirmed", case_resolution_evidence=list(file_record.get("evidence") or []), redacted_identity_fields=list(file_record.get("redacted_identity_fields") or []), review_required=bool(file_record.get("review_required", False)), review_messages=list(file_record.get("review_messages") or []), selected_document_type=file_record.get("detected_type", "generico"), detected_document_type=file_record.get("detected_type", "generico"), source_file_hash=str(file_record.get("source_file_hash") or ""), extraction_metadata=dict(file_record.get("extraction_metadata") or {}), document_title=str(file_record.get("document_title") or ""), document_key=str(file_record.get("document_key") or ""), document_reference=str(file_record.get("document_reference") or ""), parent_prefactura_batch_id=str(file_record.get("parent_prefactura_batch_id") or ""), source_page_start=int(file_record.get("source_page_start") or 0), source_page_end=int(file_record.get("source_page_end") or 0), batch_id=batch_id, batch_file_id=file_id, ingestion_source="prefactura" if str(batch.get("ingestion_mode") or "") == "prefactura_pdf" else "batch", ) def should_short_circuit_materialization(*, file_record: dict[str, Any] | None, batch_id: str) -> bool: if not file_record or file_record.get("batch_id") != batch_id: return True return file_record.get("status") != FILE_STATUS_ASOCIADO def is_already_materialized(file_record: dict[str, Any]) -> bool: return file_record.get("clinical_status") == CLINICAL_STATUS_COMPLETADO and bool( file_record.get("analysis_document_id") ) def ensure_batch_ready_for_bulk_epicrisis(batch: dict[str, Any]) -> None: batch_status = str(batch.get("status") or "") if batch_status not in {BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES}: raise ValueError("El lote aún no ha finalizado su procesamiento.") if int(batch.get("pending_validation_files") or 0) > 0: raise ValueError("El lote aún tiene documentos pendientes de validación.") if int(batch.get("clinical_pending_files") or 0) > 0: raise ValueError("El procesamiento clínico del lote aún no ha finalizado.") @dataclass(frozen=True) class BulkEpicrisisQueuePlan: queued_case_keys: list[str] skipped_completed_count: int skipped_inflight_count: int skipped_not_ready_count: int def plan_bulk_epicrisis_queue( *, cases: list[dict[str, Any]], selected_case_keys: list[str] | None, ) -> BulkEpicrisisQueuePlan: queued_case_keys: list[str] = [] preselected_keys = set(normalize_case_key_list(selected_case_keys)) skipped_completed_count = 0 skipped_inflight_count = 0 skipped_not_ready_count = 0 for case in cases: case_key = str(case.get("case_key") or "").strip() if not case_key: continue if not bool(case.get("ready_for_epicrisis")): skipped_not_ready_count += 1 continue epicrisis_status = str(case.get("epicrisis_status") or EPICRISIS_STATUS_PENDIENTE) if preselected_keys: if case_key in preselected_keys: queued_case_keys.append(case_key) continue else: if epicrisis_status == EPICRISIS_STATUS_COMPLETADO: skipped_completed_count += 1 continue if epicrisis_status in {EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PROCESANDO}: skipped_inflight_count += 1 continue queued_case_keys.append(case_key) continue if epicrisis_status == EPICRISIS_STATUS_COMPLETADO: skipped_completed_count += 1 continue if epicrisis_status in {EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PROCESANDO}: skipped_inflight_count += 1 continue return BulkEpicrisisQueuePlan( queued_case_keys=queued_case_keys, skipped_completed_count=skipped_completed_count, skipped_inflight_count=skipped_inflight_count, skipped_not_ready_count=skipped_not_ready_count, ) def ensure_batch_ready_for_excel_export(batch: dict[str, Any], cases: list[dict[str, Any]]) -> list[str]: ensure_batch_ready_for_bulk_epicrisis(batch) if any( str(item.get("epicrisis_status") or "") in {EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PROCESANDO} for item in cases ): raise ValueError("Aún hay epicrisis del lote en curso.") eligible_case_keys = [ str(item.get("case_key") or "").strip() for item in cases if str(item.get("case_key") or "").strip() and str(item.get("epicrisis_status") or "") == EPICRISIS_STATUS_COMPLETADO ] if not eligible_case_keys: raise ValueError("No hay casos con epicrisis completada para exportar.") return eligible_case_keys @dataclass(frozen=True) class ManualAssociationResolution: selected_case: dict[str, Any] | None resolved_case_key: str resolved_case_number: str resolved_patient_name: str resolved_patient_id: str resolved_procedure_code: str evidence: list[str] def resolve_manual_association( *, batch_id: str, case_key: str, patient_name: str, patient_id: str, procedure_code: str, file_record: dict[str, Any], cases: list[dict[str, Any]], ) -> ManualAssociationResolution: normalized_case_key = case_key.strip() selected_case = _find_selected_case(cases=cases, case_key=normalized_case_key, batch_id=batch_id) resolved_patient_name = ( patient_name.strip() or (selected_case.get("patient_name", "") if selected_case else "") or file_record.get("patient_name", "") ) resolved_patient_id = ( patient_id.strip() or (selected_case.get("patient_id", "") if selected_case else "") or file_record.get("patient_id", "") ) resolved_case_number = (selected_case.get("case_number", "") if selected_case else "") or file_record.get( "case_number", "" ) resolved_procedure_code = ( procedure_code.strip() or (selected_case.get("procedure_code", "") if selected_case else "") or file_record.get("procedure_code", "") ) resolved_case_key = ( normalized_case_key if normalized_case_key and normalized_case_key != "__new__" else build_case_key( patient_id=resolved_patient_id, case_number=resolved_case_number, patient_name=resolved_patient_name, fallback_name=file_record.get("original_name", ""), ) ) if not resolved_case_key: raise ValueError("No fue posible construir un case_key para la resolución manual.") evidence = list(file_record.get("evidence", [])) if "validacion_manual" not in evidence: evidence.append("validacion_manual") return ManualAssociationResolution( selected_case=selected_case, resolved_case_key=resolved_case_key, resolved_case_number=str(resolved_case_number or ""), resolved_patient_name=str(resolved_patient_name or ""), resolved_patient_id=str(resolved_patient_id or ""), resolved_procedure_code=str(resolved_procedure_code or ""), evidence=evidence, ) def validate_manual_association_target( *, file_record: dict[str, Any] | None, batch_id: str, reason: str, ) -> dict[str, Any]: if not file_record or file_record.get("batch_id") != batch_id: raise ValueError("Documento no encontrado para el lote indicado.") if file_record.get("status") == FILE_STATUS_FALLIDO: raise ValueError("No se puede resolver manualmente un archivo fallido.") if not reason.strip(): raise ValueError("El motivo de validación manual es obligatorio.") return file_record def _find_selected_case( *, cases: list[dict[str, Any]], case_key: str, batch_id: str, ) -> dict[str, Any] | None: if not case_key or case_key == "__new__": return None selected_case = next((item for item in cases if item.get("case_key") == case_key), None) if not selected_case: raise ValueError("El caso seleccionado no existe en el lote.") return selected_case @dataclass(frozen=True) class ExcelCaseExportPlan: case_key: str epicrisis_status: str included: bool omission_reason: str context: dict[str, Any] | None def plan_excel_case_export( *, case: dict[str, Any], username: str, case_epicrisis_service: Any, ) -> ExcelCaseExportPlan: case_key = str(case.get("case_key") or "").strip() epicrisis_status = str(case.get("epicrisis_status") or EPICRISIS_STATUS_PENDIENTE) if epicrisis_status != EPICRISIS_STATUS_COMPLETADO: return ExcelCaseExportPlan( case_key=case_key, epicrisis_status=epicrisis_status, included=False, omission_reason=f"Estado de epicrisis no exportable: {epicrisis_status or 'desconocido'}.", context=None, ) try: cached = case_epicrisis_service.get_cached_case_context(username, case_key) if cached and isinstance(cached.get("contexto"), dict): context = dict(cached["contexto"]) else: context = case_epicrisis_service.cache_case_context(username, case_key, regen=False) included = isinstance(context, dict) and bool(context) omission_reason = "" if included else "No se encontró contexto de epicrisis." return ExcelCaseExportPlan( case_key=case_key, epicrisis_status=epicrisis_status, included=included, omission_reason=omission_reason, context=context, ) except Exception as exc: return ExcelCaseExportPlan( case_key=case_key, epicrisis_status=epicrisis_status, included=False, omission_reason=str(exc) or "No se pudo reconstruir la epicrisis.", context=None, )