"""Casos de uso de escritura y orquestación del flujo batch.""" from __future__ import annotations import logging from dataclasses import dataclass from datetime import tzinfo from pathlib import Path from time import monotonic, sleep from typing import TYPE_CHECKING, Any from uuid import uuid4 from app.batch_processing.application.batch_command_helpers import ( build_batch_file_processing_payload, build_clinical_document_request, build_upload_accepted_response, ensure_batch_ready_for_bulk_epicrisis, ensure_batch_ready_for_excel_export, is_already_materialized, plan_bulk_epicrisis_queue, plan_excel_case_export, resolve_case_context, resolve_manual_association, should_short_circuit_materialization, should_skip_file_processing, validate_manual_association_target, validate_pdf_upload, validate_zip_upload, ) from app.batch_processing.application.builders import ( build_batch_totals_update, build_bulk_epicrisis_payload, build_cases_from_files, ) from app.batch_processing.application.models import ( BatchBulkEpicrisisPayload, BatchCaseUpdatePayload, BatchCreatePayload, BatchFileCreatePayload, BatchFileUpdatePayload, BatchUpdatePayload, ManualAssociationResultView, ManualResolutionMetadata, MaterializationResultView, QueueBatchEpicrisisExcelResult, QueueBatchEpicrisisResult, ) from app.batch_processing.application.prefactura_compound import ( PrefacturaCompoundOrchestrator, _NullCaseAssociationService, _NullDocumentClassifier, persist_prefactura_segments, ) from app.batch_processing.application.utils import ( build_associated_user, now_iso, ) from app.batch_processing.domain.models import ( ASSOCIATION_SOURCE_MANUAL, BATCH_STATUS_CANCELADO, BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, BATCH_STATUS_ELIMINANDO, BATCH_STATUS_FALLIDO, BATCH_STATUS_PREPARANDO, BATCH_STATUS_PROCESANDO, CLINICAL_STATUS_COMPLETADO, CLINICAL_STATUS_EN_COLA, CLINICAL_STATUS_FALLIDO, CLINICAL_STATUS_PENDIENTE, CLINICAL_STATUS_PROCESANDO, EPICRISIS_STATUS_COMPLETADO, EPICRISIS_STATUS_COMPLETADO_CON_ERRORES, EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_FALLIDO, EPICRISIS_STATUS_PROCESANDO, FILE_STATUS_ASOCIADO, FILE_STATUS_CLASIFICADO, FILE_STATUS_FALLIDO, FILE_STATUS_PENDIENTE, FILE_STATUS_PENDIENTE_VALIDACION, FILE_STATUS_PROCESANDO, INGESTION_MODE_MANUAL_HISTORIA, INGESTION_MODE_MANUAL_SOPORTE, INGESTION_MODE_PREFACTURA_PDF, PREFERRED_ASSOCIATION_MODE_VALIDATE_AND_WARN, ) from app.batch_processing.domain.ports import ( ArchiveExtractor, BatchArchiveStore, BatchArtifactCleaner, BatchCaseRepository, BatchFileRepository, BatchJobDispatcher, BatchReportStore, BatchRepository, BatchWorkingFileStore, CaseAssociationService, DocumentClassifier, PrefacturaPdfExtractor, TextExtractor, ) from app.core.logging import bind_log_context, get_audit_logger if TYPE_CHECKING: from app.batch_processing.infrastructure.batch_epicrisis_excel import ( BatchEpicrisisExcelWorkbookBuilder, ) from app.services.case_epicrisis_service import CaseEpicrisisService from app.services.clinical_document_service import ClinicalDocumentService from app.services.clinical_document_service import ClinicalDocumentRequest logger = logging.getLogger(__name__) audit_logger = get_audit_logger() _BATCH_NOT_FOUND_MESSAGE = "Lote no encontrado." _BATCH_DELETING_MESSAGE = "El lote está siendo eliminado y no admite más operaciones." _BATCH_CANCELLED_MESSAGE = "El lote fue cancelado y no admite más operaciones." # Alias de compatibilidad para consumidores y pruebas que todavía importan # el payload histórico desde este módulo. ClinicalDocumentPayload = ClinicalDocumentRequest class BatchDeletionError(Exception): def __init__(self, message: str, *, status_code: int) -> None: super().__init__(message) self.status_code = status_code def _default_excel_filename(batch_id: str) -> str: return f"epicrisis_lote_{batch_id}.xlsx" def _is_batch_deleting(batch: dict[str, Any] | None) -> bool: if not batch: return False return ( str(batch.get("status") or "").strip() == BATCH_STATUS_ELIMINANDO or str(batch.get("deletion_status") or "").strip() == "requested" ) def _is_batch_cancelled(batch: dict[str, Any] | None) -> bool: if not batch: return False return str(batch.get("status") or "").strip() == BATCH_STATUS_CANCELADO @dataclass class RecomputeBatchTotalsUseCase: """Recalcula contadores y estado del lote desde el estado real de sus archivos.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository colombia_tz: tzinfo def execute(self, batch_id: str, *, force_status: str | None = None) -> None: batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return files = self.batch_file_repository.list_files(batch_id) payload = build_batch_totals_update( files, updated_at=now_iso(self.colombia_tz), force_status=force_status, ) self.batch_repository.update_batch(batch_id, payload.to_document()) @dataclass class CreateBatchUploadUseCase: """Registra el lote, persiste el ZIP y dispara su procesamiento asíncrono.""" batch_repository: BatchRepository archive_store: BatchArchiveStore job_dispatcher: BatchJobDispatcher colombia_tz: tzinfo def execute( self, *, filename: str, contents: bytes, username: str, ) -> dict[str, Any]: validate_zip_upload(filename=filename, contents=contents) created_at = now_iso(self.colombia_tz) create_payload = BatchCreatePayload( usuario=username, nombre_archivo=filename, created_at=created_at, updated_at=created_at, ) batch_id = self.batch_repository.create_batch(create_payload.to_document()) archive_path = self.archive_store.save_archive(batch_id, filename, contents) self.batch_repository.update_batch( batch_id, BatchUpdatePayload( archive_path=archive_path, updated_at=now_iso(self.colombia_tz), ).to_document(), ) self.job_dispatcher.dispatch(batch_id) return build_upload_accepted_response( batch_repository=self.batch_repository, batch_id=batch_id, created_at=created_at, ) @dataclass class CreateManualBatchUploadUseCase: """Crea un mini-lote manual de un PDF y lo despacha al pipeline batch.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository working_file_store: BatchWorkingFileStore job_dispatcher: BatchJobDispatcher colombia_tz: tzinfo case_context_resolver: Any | None = None def execute( self, *, filename: str, contents: bytes, username: str, ingestion_mode: str, detected_type: str = "", case_key: str = "", ) -> dict[str, Any]: normalized_filename = validate_pdf_upload( filename=filename, contents=contents, empty_message="El archivo PDF está vacío.", invalid_message="Solo se permiten archivos PDF.", ) normalized_case_key = str(case_key or "").strip() if ingestion_mode not in {INGESTION_MODE_MANUAL_HISTORIA, INGESTION_MODE_MANUAL_SOPORTE}: raise ValueError("Modo de cargue manual no soportado.") if ingestion_mode == INGESTION_MODE_MANUAL_SOPORTE and not str(detected_type or "").strip(): raise ValueError("Debes indicar el tipo de documento para el soporte.") created_at = now_iso(self.colombia_tz) create_payload = BatchCreatePayload( usuario=username, nombre_archivo=normalized_filename, ingestion_mode=ingestion_mode, created_at=created_at, updated_at=created_at, total_files=1, ) batch_id = self.batch_repository.create_batch(create_payload.to_document()) stored_path = self.working_file_store.save_file(batch_id, normalized_filename, contents) preferred_context = resolve_case_context( case_context_resolver=self.case_context_resolver, username=username, case_key=normalized_case_key, ) file_payload = BatchFileCreatePayload( batch_id=batch_id, original_name=normalized_filename, relative_path=Path(stored_path).name, stored_path=stored_path, status=FILE_STATUS_PENDIENTE, detected_type=str(detected_type or "").strip(), preferred_case_key=normalized_case_key, preferred_case_number=str(preferred_context.get("case_number") or "").strip(), preferred_patient_id=str(preferred_context.get("patient_id") or "").strip(), preferred_patient_name=str(preferred_context.get("patient_name") or "").strip(), preferred_association_mode=PREFERRED_ASSOCIATION_MODE_VALIDATE_AND_WARN, created_at=created_at, updated_at=created_at, ) self.batch_file_repository.create_file(file_payload.to_document()) self.job_dispatcher.dispatch(batch_id) return build_upload_accepted_response( batch_repository=self.batch_repository, batch_id=batch_id, created_at=created_at, ) @dataclass class CreatePrefacturaBatchUploadUseCase: """Crea un lote desde un PDF compuesto de prefactura y sus soportes.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository working_file_store: BatchWorkingFileStore prefactura_pdf_extractor: PrefacturaPdfExtractor job_dispatcher: BatchJobDispatcher colombia_tz: tzinfo document_classifier: DocumentClassifier | None = None association_service: CaseAssociationService | None = None prefactura_orchestrator: PrefacturaCompoundOrchestrator | None = None def execute( self, *, filename: str, contents: bytes, username: str, ) -> dict[str, Any]: normalized_filename = validate_pdf_upload( filename=filename, contents=contents, empty_message="El PDF de prefactura está vacío.", invalid_message="Solo se permiten archivos PDF para prefactura.", ) orchestrator = self._resolve_orchestrator() pages = self.prefactura_pdf_extractor.extract_pages(contents) if not pages: raise ValueError("El PDF de prefactura no contiene páginas procesables.") segments = orchestrator.build_segments(pages=pages) if not segments: raise ValueError("No fue posible identificar documentos dentro de la prefactura.") upload_context = orchestrator.build_upload_context( filename=normalized_filename, segments=segments, colombia_tz=self.colombia_tz, ) batch_id = self.batch_repository.create_batch( BatchCreatePayload( usuario=username, nombre_archivo=normalized_filename, ingestion_mode=INGESTION_MODE_PREFACTURA_PDF, created_at=upload_context.created_at, updated_at=upload_context.created_at, total_files=len(segments), ).to_document() ) persist_prefactura_segments( batch_id=batch_id, normalized_filename=normalized_filename, contents=contents, segments=segments, upload_context=upload_context, prefactura_pdf_extractor=self.prefactura_pdf_extractor, working_file_store=self.working_file_store, batch_file_repository=self.batch_file_repository, association_service=orchestrator.association_service, ) self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=BATCH_STATUS_PROCESANDO, updated_at=now_iso(self.colombia_tz), ).to_document(), ) self.job_dispatcher.dispatch(batch_id) return build_upload_accepted_response( batch_repository=self.batch_repository, batch_id=batch_id, created_at=upload_context.created_at, ) def _resolve_orchestrator(self) -> PrefacturaCompoundOrchestrator: if self.prefactura_orchestrator is not None: return self.prefactura_orchestrator return PrefacturaCompoundOrchestrator( document_classifier=self.document_classifier or _NullDocumentClassifier(), association_service=self.association_service or _NullCaseAssociationService(), ) @dataclass class PrepareBatchUseCase: """Extrae el ZIP de forma segura y crea los registros iniciales por archivo.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository archive_extractor: ArchiveExtractor artifact_cleaner: BatchArtifactCleaner totals_use_case: RecomputeBatchTotalsUseCase colombia_tz: tzinfo def execute(self, batch_id: str) -> list[str]: batch = self.batch_repository.get_batch(batch_id) if not batch: logger.warning("Lote %s no encontrado para preparación", batch_id) return [] if _is_batch_deleting(batch) or _is_batch_cancelled(batch): logger.info("Lote %s omitido en preparación por eliminación en curso", batch_id) return [] existing_files = self.batch_file_repository.list_files(batch_id) if existing_files: self.totals_use_case.execute(batch_id, force_status=BATCH_STATUS_PROCESANDO) return [item["_id"] for item in existing_files if item.get("status") != FILE_STATUS_FALLIDO] self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=BATCH_STATUS_PREPARANDO, error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) archive_path = str(batch.get("archive_path") or "") try: extraction = self.archive_extractor.extract_archive(archive_path, batch_id) except Exception as exc: logger.exception("No se pudo extraer el lote %s", batch_id) self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=BATCH_STATUS_FALLIDO, error=f"No se pudo descomprimir el ZIP: {exc}", updated_at=now_iso(self.colombia_tz), ).to_document(), ) return [] finally: self._cleanup_archive(batch_id=batch_id, archive_path=archive_path) if not extraction.entries and not extraction.rejected_entries: self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=BATCH_STATUS_FALLIDO, error="El ZIP no contiene PDFs procesables.", updated_at=now_iso(self.colombia_tz), ).to_document(), ) return [] file_ids: list[str] = [] for rejected in extraction.rejected_entries: payload = self._build_file_payload( batch_id=batch_id, original_name=rejected.original_name, relative_path=rejected.relative_path, stored_path=rejected.extracted_path or "", status=FILE_STATUS_FALLIDO, error=rejected.error or "Archivo no soportado", ) self.batch_file_repository.create_file(payload.to_document()) for entry in extraction.entries: payload = self._build_file_payload( batch_id=batch_id, original_name=entry.original_name, relative_path=entry.relative_path, stored_path=entry.extracted_path or "", status=FILE_STATUS_PENDIENTE, ) file_id = self.batch_file_repository.create_file(payload.to_document()) file_ids.append(file_id) self.totals_use_case.execute(batch_id, force_status=BATCH_STATUS_PROCESANDO) return file_ids def _build_file_payload( self, *, batch_id: str, original_name: str, relative_path: str, stored_path: str, status: str, error: str = "", ) -> BatchFileCreatePayload: timestamp = now_iso(self.colombia_tz) return BatchFileCreatePayload( batch_id=batch_id, original_name=original_name, relative_path=relative_path, stored_path=stored_path, status=status, error=error, created_at=timestamp, updated_at=timestamp, ) def _cleanup_archive(self, *, batch_id: str, archive_path: str) -> None: if not archive_path: return try: self.artifact_cleaner.delete_archive(archive_path) except Exception as exc: logger.warning( "No se pudo eliminar ZIP temporal del lote %s (%s): %s", batch_id, archive_path, exc, ) return self.batch_repository.update_batch( batch_id, BatchUpdatePayload( archive_path="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) @dataclass class ProcessBatchFileUseCase: """Extrae texto, clasifica y deja señales normalizadas sin bloquear el lote.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository text_extractor: TextExtractor document_classifier: DocumentClassifier association_service: CaseAssociationService colombia_tz: tzinfo def execute(self, batch_id: str, file_id: str) -> None: batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return file_record = self.batch_file_repository.get_file(file_id) if should_skip_file_processing(file_record=file_record, batch_id=batch_id): return if file_record is None: return self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( status=FILE_STATUS_PROCESANDO, updated_at=now_iso(self.colombia_tz), ).to_document(), ) try: extract_method = getattr(self.text_extractor, "extract", None) if callable(extract_method): extraction_result = extract_method(file_record["stored_path"]) text = extraction_result.text extraction_metadata = extraction_result.metadata.to_document() else: text = self.text_extractor.extract_text(file_record["stored_path"]) extraction_metadata = {"character_count": len(text), "warnings": []} if not text.strip(): raise ValueError("El PDF no contiene texto extraíble y requiere revisión manual.") self.batch_file_repository.update_file( file_id, build_batch_file_processing_payload( file_record=file_record, extracted_text=text, document_classifier=self.document_classifier, association_service=self.association_service, updated_at=now_iso(self.colombia_tz), extraction_metadata=extraction_metadata, ), ) except Exception as exc: logger.warning("Fallo procesando archivo %s del lote %s: %s", file_id, batch_id, exc) self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( status=FILE_STATUS_FALLIDO, error=str(exc), updated_at=now_iso(self.colombia_tz), ).to_document(), ) @dataclass class RefreshBatchCasesUseCase: """Reconstruye y persiste el agregado por caso desde el estado actual del lote.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository batch_case_repository: BatchCaseRepository lock_ttl_seconds: int = 30 lock_retry_interval_seconds: float = 0.05 lock_wait_timeout_seconds: float = 5.0 def execute(self, batch_id: str) -> list[dict[str, Any]]: owner = f"refresh-{uuid4().hex}" self._acquire_refresh_lock(batch_id, owner) try: batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return [] files = self.batch_file_repository.list_files(batch_id) existing_cases = { item.get("case_key", ""): item for item in self.batch_case_repository.list_cases(batch_id) if item.get("case_key") } cases = build_cases_from_files( batch_id, files, username=str(batch.get("usuario") or ""), batch_status=str(batch.get("status") or ""), pending_validation_files=int(batch.get("pending_validation_files") or 0), existing_cases=existing_cases, ) serialized_cases = [item.to_document() for item in cases] self.batch_case_repository.replace_cases(batch_id, serialized_cases) return serialized_cases finally: self.batch_repository.release_cases_refresh_lock(batch_id, owner) def _acquire_refresh_lock(self, batch_id: str, owner: str) -> None: deadline = monotonic() + max(float(self.lock_wait_timeout_seconds or 0), 0.0) retry_interval = max(float(self.lock_retry_interval_seconds or 0), 0.01) while True: acquired = self.batch_repository.acquire_cases_refresh_lock( batch_id, owner, max(int(self.lock_ttl_seconds or 0), 1), ) if acquired: return if monotonic() >= deadline: raise TimeoutError( f"No fue posible adquirir el lock de refresco de casos para el lote {batch_id}." ) sleep(retry_interval) @dataclass class FinalizeBatchUseCase: """Ejecuta la asociación global y deja el lote listo para el paso clínico.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository association_service: CaseAssociationService totals_use_case: RecomputeBatchTotalsUseCase refresh_cases_use_case: RefreshBatchCasesUseCase colombia_tz: tzinfo def execute(self, batch_id: str) -> list[str]: batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return [] files = self.batch_file_repository.list_files(batch_id) candidates = [ item for item in files if item.get("status") in { FILE_STATUS_CLASIFICADO, FILE_STATUS_ASOCIADO, FILE_STATUS_PENDIENTE_VALIDACION, } and item.get("association_source") != ASSOCIATION_SOURCE_MANUAL ] association_result = self.association_service.associate(candidates) for decision in association_result.decisions: payload = BatchFileUpdatePayload.from_association_decision( decision, updated_at=now_iso(self.colombia_tz), ) self.batch_file_repository.update_file(decision.file_id, payload.to_document()) self.totals_use_case.execute(batch_id) self.refresh_cases_use_case.execute(batch_id) refreshed_files = self.batch_file_repository.list_files(batch_id) return [ item["_id"] for item in refreshed_files if item.get("status") == FILE_STATUS_ASOCIADO and item.get("clinical_status") != CLINICAL_STATUS_COMPLETADO ] @dataclass class MaterializeBatchFileUseCase: """Convierte un archivo asociado en un documento clínico persistido.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository totals_use_case: RecomputeBatchTotalsUseCase refresh_cases_use_case: RefreshBatchCasesUseCase clinical_document_service: ClinicalDocumentService case_deletion_service: Any artifact_cleaner: BatchArtifactCleaner colombia_tz: tzinfo def execute(self, batch_id: str, file_id: str) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return {} file_record = self.batch_file_repository.get_file(file_id) if should_short_circuit_materialization(file_record=file_record, batch_id=batch_id): return {} if not file_record: return {} if is_already_materialized(file_record): self._cleanup_extracted_file( batch_id=batch_id, file_id=file_id, stored_path=str(file_record.get("stored_path") or ""), ) self.totals_use_case.execute(batch_id) self.refresh_cases_use_case.execute(batch_id) return MaterializationResultView( file_id=file_id, analysis_document_id=str(file_record.get("analysis_document_id") or ""), clinical_status=CLINICAL_STATUS_COMPLETADO, ).to_document() self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( clinical_status=CLINICAL_STATUS_PROCESANDO, clinical_error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) try: result = self.clinical_document_service.process_and_persist( build_clinical_document_request( batch=batch, file_record=file_record, batch_id=batch_id, file_id=file_id, ) ) refreshed_batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(refreshed_batch) or _is_batch_cancelled(refreshed_batch): self.case_deletion_service.purge_document_record( username=str(refreshed_batch.get("usuario") or batch.get("usuario") or ""), document_id=str(result.get("id_documento") or ""), ) return {} self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( clinical_status=CLINICAL_STATUS_COMPLETADO, analysis_document_id=result.get("id_documento", ""), clinical_error="", clinical_processed_at=now_iso(self.colombia_tz), updated_at=now_iso(self.colombia_tz), ).to_document(), ) self._cleanup_extracted_file( batch_id=batch_id, file_id=file_id, stored_path=str(file_record.get("stored_path") or ""), ) response = MaterializationResultView( file_id=file_id, analysis_document_id=str(result.get("id_documento", "")), clinical_status=CLINICAL_STATUS_COMPLETADO, ) except Exception as exc: logger.exception( "Fallo materializando clínicamente archivo %s del lote %s", file_id, batch_id, ) self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( clinical_status=CLINICAL_STATUS_FALLIDO, clinical_error=str(exc), clinical_processed_at=now_iso(self.colombia_tz), updated_at=now_iso(self.colombia_tz), ).to_document(), ) response = MaterializationResultView( file_id=file_id, clinical_status=CLINICAL_STATUS_FALLIDO, clinical_error=str(exc), ) self.totals_use_case.execute(batch_id) self.refresh_cases_use_case.execute(batch_id) return response.to_document() def _cleanup_extracted_file( self, *, batch_id: str, file_id: str, stored_path: str, ) -> None: if not stored_path: return try: self.artifact_cleaner.delete_extracted_file(stored_path) except Exception as exc: logger.warning( "No se pudo eliminar PDF temporal %s del archivo %s del lote %s: %s", stored_path, file_id, batch_id, exc, ) return self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( stored_path="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) @dataclass class QueueBatchFileClinicalUseCase: """Marca un archivo asociado como encolado para materialización clínica.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository totals_use_case: RecomputeBatchTotalsUseCase refresh_cases_use_case: RefreshBatchCasesUseCase colombia_tz: tzinfo def execute(self, batch_id: str, file_id: str, *, job_id: str = "") -> None: batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return file_record = self.batch_file_repository.get_file(file_id) if not file_record or file_record.get("batch_id") != batch_id: return if file_record.get("status") != FILE_STATUS_ASOCIADO: return self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( clinical_status=CLINICAL_STATUS_EN_COLA, clinical_job_id=job_id, clinical_error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) self.totals_use_case.execute(batch_id) self.refresh_cases_use_case.execute(batch_id) @dataclass class RecomputeBatchBulkEpicrisisUseCase: """Sincroniza el agregado masivo de epicrisis del lote con sus casos.""" batch_repository: BatchRepository batch_case_repository: BatchCaseRepository def execute(self, batch_id: str) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) if not batch: return {} if _is_batch_deleting(batch) or _is_batch_cancelled(batch): return {} cases = self.batch_case_repository.list_cases(batch_id) payload = build_bulk_epicrisis_payload(batch, cases) current = BatchBulkEpicrisisPayload.from_batch_record(batch) if current.to_document() != payload.to_document(): self.batch_repository.update_batch(batch_id, payload.to_document()) return payload.to_document() @dataclass class QueueBatchEpicrisisUseCase: """Selecciona casos elegibles para epicrisis masiva y persiste el agregado.""" batch_repository: BatchRepository batch_case_repository: BatchCaseRepository recompute_bulk_epicrisis_use_case: RecomputeBatchBulkEpicrisisUseCase colombia_tz: tzinfo def execute( self, batch_id: str, *, job_id: str = "", selected_case_keys: list[str] | None = None, persist: bool = True, ) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) or {} if not batch: raise ValueError(_BATCH_NOT_FOUND_MESSAGE) if _is_batch_deleting(batch): raise ValueError(_BATCH_DELETING_MESSAGE) if _is_batch_cancelled(batch): raise ValueError(_BATCH_CANCELLED_MESSAGE) ensure_batch_ready_for_bulk_epicrisis(batch) queue_plan = plan_bulk_epicrisis_queue( cases=self.batch_case_repository.list_cases(batch_id), selected_case_keys=selected_case_keys, ) timestamp = now_iso(self.colombia_tz) skipped_total = ( queue_plan.skipped_completed_count + queue_plan.skipped_inflight_count + queue_plan.skipped_not_ready_count ) initial_status = ( EPICRISIS_STATUS_EN_COLA if queue_plan.queued_case_keys else EPICRISIS_STATUS_COMPLETADO ) payload = BatchBulkEpicrisisPayload( bulk_epicrisis_status=initial_status, bulk_epicrisis_job_id=job_id, bulk_epicrisis_requested_at=timestamp, bulk_epicrisis_total_target=len(queue_plan.queued_case_keys), bulk_epicrisis_completed_count=0, bulk_epicrisis_failed_count=0, bulk_epicrisis_skipped_count=skipped_total, bulk_epicrisis_case_keys=queue_plan.queued_case_keys, ) if persist: for case_key in queue_plan.queued_case_keys: self.batch_case_repository.update_case( batch_id, case_key, BatchCaseUpdatePayload( epicrisis_status=EPICRISIS_STATUS_EN_COLA, epicrisis_job_id=job_id, epicrisis_error="", epicrisis_url=f"/epicrisis?case_key={case_key}", updated_at=timestamp, ).to_document(), ) self.batch_repository.update_batch(batch_id, payload.to_document()) if not queue_plan.queued_case_keys: payload = BatchBulkEpicrisisPayload( **self.recompute_bulk_epicrisis_use_case.execute(batch_id) ) return QueueBatchEpicrisisResult( batch_id=batch_id, job_id=job_id, status=payload.bulk_epicrisis_status, queued_count=len(queue_plan.queued_case_keys), skipped_completed_count=queue_plan.skipped_completed_count, skipped_inflight_count=queue_plan.skipped_inflight_count, skipped_not_ready_count=queue_plan.skipped_not_ready_count, case_keys=queue_plan.queued_case_keys, ).to_document() @dataclass class QueueBatchEpicrisisExcelUseCase: """Valida y marca el lote como listo para generar el workbook Excel.""" batch_repository: BatchRepository batch_case_repository: BatchCaseRepository colombia_tz: tzinfo def execute( self, batch_id: str, *, job_id: str = "", persist: bool = True, ) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) or {} if not batch: raise ValueError(_BATCH_NOT_FOUND_MESSAGE) if _is_batch_deleting(batch): raise ValueError(_BATCH_DELETING_MESSAGE) if _is_batch_cancelled(batch): raise ValueError(_BATCH_CANCELLED_MESSAGE) cases = self.batch_case_repository.list_cases(batch_id) eligible_case_keys = ensure_batch_ready_for_excel_export(batch, cases) filename = _default_excel_filename(batch_id) if persist: now = now_iso(self.colombia_tz) self.batch_repository.update_batch( batch_id, BatchUpdatePayload( excel_epicrisis_status=EPICRISIS_STATUS_EN_COLA, excel_epicrisis_job_id=job_id, excel_epicrisis_requested_at=now, excel_epicrisis_generated_at="", excel_epicrisis_error="", excel_epicrisis_filename=filename, excel_epicrisis_download_url="", excel_epicrisis_included_count=0, excel_epicrisis_omitted_count=0, excel_epicrisis_path="", updated_at=now, ).to_document(), ) return QueueBatchEpicrisisExcelResult( batch_id=batch_id, job_id=job_id, status=EPICRISIS_STATUS_EN_COLA, eligible_count=len(eligible_case_keys), filename=filename, ).to_document() @dataclass class GenerateBatchEpicrisisExcelUseCase: """Genera el workbook Excel del lote con una hoja por cada case_key exportable.""" batch_repository: BatchRepository batch_case_repository: BatchCaseRepository report_store: BatchReportStore workbook_builder: BatchEpicrisisExcelWorkbookBuilder case_epicrisis_service: CaseEpicrisisService colombia_tz: tzinfo def execute(self, batch_id: str, *, job_id: str = "") -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) or {} if not batch: raise ValueError(_BATCH_NOT_FOUND_MESSAGE) if _is_batch_deleting(batch): raise ValueError(_BATCH_DELETING_MESSAGE) if _is_batch_cancelled(batch): raise ValueError(_BATCH_CANCELLED_MESSAGE) username = str(batch.get("usuario") or "").strip() if not username: raise ValueError("El lote no tiene usuario asociado.") self.batch_repository.update_batch( batch_id, BatchUpdatePayload( excel_epicrisis_status=EPICRISIS_STATUS_PROCESANDO, excel_epicrisis_job_id=job_id or None, excel_epicrisis_error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) cases = self.batch_case_repository.list_cases(batch_id) rows: list[Any] = [] included_count = 0 omitted_count = 0 filename = _default_excel_filename(batch_id) try: for case in cases: export_plan = plan_excel_case_export( case=case, username=username, case_epicrisis_service=self.case_epicrisis_service, ) case_key = export_plan.case_key if not case_key: continue if export_plan.included: included_count += 1 else: omitted_count += 1 rows.append( self.workbook_builder_row( case=case, case_key=case_key, epicrisis_status=export_plan.epicrisis_status, included=export_plan.included, omission_reason=export_plan.omission_reason, context=export_plan.context, ) ) if included_count == 0: raise ValueError("No se pudo incluir ningún caso en el Excel del lote.") workbook_bytes = self.workbook_builder.build(batch_id=batch_id, rows=rows) report_path = self.report_store.save_excel_report(batch_id, filename, workbook_bytes) final_status = ( EPICRISIS_STATUS_COMPLETADO if omitted_count == 0 else EPICRISIS_STATUS_COMPLETADO_CON_ERRORES ) download_url = f"/api/lotes/{batch_id}/excel-epicrisis/descarga" self.batch_repository.update_batch( batch_id, BatchUpdatePayload( excel_epicrisis_status=final_status, excel_epicrisis_job_id=job_id or None, excel_epicrisis_generated_at=now_iso(self.colombia_tz), excel_epicrisis_error="", excel_epicrisis_filename=filename, excel_epicrisis_download_url=download_url, excel_epicrisis_included_count=included_count, excel_epicrisis_omitted_count=omitted_count, excel_epicrisis_path=str(Path(report_path)), updated_at=now_iso(self.colombia_tz), ).to_document(), ) except Exception as exc: self.batch_repository.update_batch( batch_id, BatchUpdatePayload( excel_epicrisis_status=EPICRISIS_STATUS_FALLIDO, excel_epicrisis_job_id=job_id or None, excel_epicrisis_error=str(exc), excel_epicrisis_download_url="", excel_epicrisis_included_count=included_count, excel_epicrisis_omitted_count=omitted_count, updated_at=now_iso(self.colombia_tz), ).to_document(), ) raise return { "batch_id": batch_id, "status": final_status, "filename": filename, "included_count": included_count, "omitted_count": omitted_count, } def workbook_builder_row( self, *, case: dict[str, Any], case_key: str, epicrisis_status: str, included: bool, omission_reason: str, context: dict[str, Any] | None, ) -> Any: from app.batch_processing.infrastructure.batch_epicrisis_excel import ( BatchEpicrisisExcelCaseRow, ) return BatchEpicrisisExcelCaseRow( case_key=case_key, patient_name=str(case.get("patient_name") or ""), patient_id=str(case.get("patient_id") or ""), case_number=str(case.get("case_number") or ""), procedure_description=str(case.get("procedure_description") or ""), epicrisis_status=epicrisis_status, included_in_excel=included, omission_reason=omission_reason, context=context, ) @dataclass class ResolveFileAssociationUseCase: """Resuelve manualmente documentos ambiguos sin alterar el flujo automático.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository totals_use_case: RecomputeBatchTotalsUseCase refresh_cases_use_case: RefreshBatchCasesUseCase colombia_tz: tzinfo def execute( self, *, batch_id: str, file_id: str, resolved_by: str, case_key: str, patient_name: str, patient_id: str, procedure_code: str, reason: str, ) -> dict[str, Any]: bind_log_context(batch_id=batch_id, file_id=file_id, case_key=case_key, username=resolved_by) batch = self.batch_repository.get_batch(batch_id) or {} if _is_batch_deleting(batch): raise ValueError(_BATCH_DELETING_MESSAGE) if _is_batch_cancelled(batch): raise ValueError(_BATCH_CANCELLED_MESSAGE) file_record = self.batch_file_repository.get_file(file_id) file_record = validate_manual_association_target( file_record=file_record, batch_id=batch_id, reason=reason, ) resolution = resolve_manual_association( batch_id=batch_id, case_key=case_key, patient_name=patient_name, patient_id=patient_id, procedure_code=procedure_code, file_record=file_record, cases=self.refresh_cases_use_case.batch_case_repository.list_cases(batch_id), ) manual_resolution = ManualResolutionMetadata( resolved_by=resolved_by, resolved_at=now_iso(self.colombia_tz), reason=reason.strip(), ) self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( status=FILE_STATUS_ASOCIADO, case_key=resolution.resolved_case_key, patient_name=resolution.resolved_patient_name, patient_id=resolution.resolved_patient_id, procedure_code=resolution.resolved_procedure_code, associated_user=build_associated_user( resolution.resolved_patient_id, resolution.resolved_patient_name, ), association_source=ASSOCIATION_SOURCE_MANUAL, confidence=max(float(file_record.get("confidence", 0.0)), 1.0), evidence=resolution.evidence, manual_resolution=manual_resolution.to_document(), updated_at=now_iso(self.colombia_tz), error="", ).to_document(), ) self.totals_use_case.execute(batch_id) self.refresh_cases_use_case.execute(batch_id) updated_file = self.batch_file_repository.get_file(file_id) or {} audit_logger.business_event( event_type="audit.manual_resolution", action="resolve_file_association", outcome="success", service="resolve_file_association_use_case", resource={ "batch_id": batch_id, "file_id": file_id, "case_key": resolution.resolved_case_key, "resolved_by": resolved_by, }, ) return ManualAssociationResultView.from_record(updated_file).to_document() @dataclass class CancelBatchProcessingUseCase: """Marca un lote no terminal como cancelado para detener trabajo futuro.""" batch_repository: BatchRepository colombia_tz: tzinfo def execute(self, *, batch_id: str, username: str) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) if not batch or str(batch.get("usuario") or "").strip() != username: raise BatchDeletionError("Lote no encontrado.", status_code=404) if _is_batch_deleting(batch): raise BatchDeletionError(_BATCH_DELETING_MESSAGE, status_code=409) if _is_batch_cancelled(batch): return { "message": "El lote ya estaba cancelado.", "batch_id": batch_id, "status": BATCH_STATUS_CANCELADO, } if str(batch.get("status") or "").strip() in { BATCH_STATUS_FALLIDO, BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, }: raise BatchDeletionError("El lote ya está finalizado y no se puede cancelar.", status_code=409) self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=BATCH_STATUS_CANCELADO, error="Cancelado por el usuario.", updated_at=now_iso(self.colombia_tz), ).to_document(), ) return { "message": "Lote cancelado correctamente.", "batch_id": batch_id, "status": BATCH_STATUS_CANCELADO, } @dataclass class RetryBatchProcessingUseCase: """Reencola solo trabajo fallido o incompleto recuperable del lote.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository process_batch_file_use_case: ProcessBatchFileUseCase queue_batch_file_clinical_use_case: QueueBatchFileClinicalUseCase totals_use_case: RecomputeBatchTotalsUseCase refresh_cases_use_case: RefreshBatchCasesUseCase colombia_tz: tzinfo def execute(self, *, batch_id: str, username: str) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) if not batch or str(batch.get("usuario") or "").strip() != username: raise BatchDeletionError("Lote no encontrado.", status_code=404) if _is_batch_deleting(batch): raise BatchDeletionError(_BATCH_DELETING_MESSAGE, status_code=409) retried_files = 0 requeued_clinical = 0 files = self.batch_file_repository.list_files(batch_id) for item in files: file_id = str(item.get("_id") or "").strip() if not file_id: continue file_status = str(item.get("status") or "").strip() clinical_status = str(item.get("clinical_status") or "").strip() if file_status == FILE_STATUS_FALLIDO and str(item.get("stored_path") or "").strip(): self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( status=FILE_STATUS_PENDIENTE, error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) self.process_batch_file_use_case.execute(batch_id, file_id) retried_files += 1 continue if file_status == FILE_STATUS_ASOCIADO and clinical_status in { "", CLINICAL_STATUS_PENDIENTE, CLINICAL_STATUS_FALLIDO, }: self.queue_batch_file_clinical_use_case.execute(batch_id, file_id) requeued_clinical += 1 if retried_files == 0 and requeued_clinical == 0: raise BatchDeletionError("El lote no tiene trabajo recuperable para reintentar.", status_code=409) next_status = BATCH_STATUS_PROCESANDO self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=next_status, error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) self.totals_use_case.execute(batch_id, force_status=next_status) self.refresh_cases_use_case.execute(batch_id) return { "message": "Trabajo recuperable reencolado correctamente.", "batch_id": batch_id, "status": next_status, "retried_files": retried_files, "requeued_clinical_files": requeued_clinical, } @dataclass class DeleteBatchUseCase: """Elimina un lote batch y todos los casos clínicos alcanzados por su cascada.""" batch_repository: BatchRepository batch_file_repository: BatchFileRepository batch_case_repository: BatchCaseRepository artifact_cleaner: BatchArtifactCleaner case_deletion_service: Any colombia_tz: tzinfo def execute( self, *, batch_id: str, username: str, confirmation_batch_id: str, confirmation_phrase: str, ) -> dict[str, Any]: batch = self.batch_repository.get_batch(batch_id) if not batch or str(batch.get("usuario") or "").strip() != username: raise BatchDeletionError("Lote no encontrado.", status_code=404) if str(confirmation_batch_id or "").strip() != batch_id: raise BatchDeletionError("El batch_id de confirmación no coincide.", status_code=400) if str(confirmation_phrase or "").strip() != "ELIMINAR LOTE": raise BatchDeletionError("La frase de confirmación no coincide.", status_code=400) timestamp = now_iso(self.colombia_tz) self.batch_repository.update_batch( batch_id, BatchUpdatePayload( status=BATCH_STATUS_ELIMINANDO, updated_at=timestamp, ).to_document() | {"deletion_status": "requested", "deletion_requested_at": timestamp}, ) files = self.batch_file_repository.list_files(batch_id) cases = self.batch_case_repository.list_cases(batch_id) case_keys = sorted( { str(value or "").strip() for value in ( [item.get("case_key") for item in files] + [item.get("case_key") for item in cases] ) if str(value or "").strip() } ) deleted_documents = 0 deleted_derived_artifacts = 0 deleted_case_keys = 0 for case_key in case_keys: try: payload = self.case_deletion_service.delete_case(username=username, case_key=case_key) except Exception as exc: if getattr(exc, "status_code", None) == 404: continue raise deleted_documents += int(payload.get("deleted_documents", 0) or 0) deleted_derived_artifacts += int(payload.get("deleted_derived_artifacts", 0) or 0) deleted_case_keys += 1 orphaned_batch_documents = 0 purge_batch_documents = getattr(self.case_deletion_service, "purge_documents_by_batch_file_ids", None) if callable(purge_batch_documents): orphaned_batch_documents = int( purge_batch_documents( username=username, batch_file_ids=[str(item.get("_id") or "") for item in files], ) or 0 ) deleted_files = self.batch_file_repository.delete_files_by_batch(batch_id) deleted_cases = self.batch_case_repository.delete_cases_by_batch(batch_id) self.artifact_cleaner.delete_batch_artifacts(batch_id) deleted_batches = self.batch_repository.delete_batch(batch_id) audit_logger.business_event( event_type="batch.deleted", action="delete_batch", outcome="success", service="delete_batch_use_case", resource={"batch_id": batch_id, "username": username}, metrics={ "deleted_batches": deleted_batches, "deleted_files": deleted_files, "deleted_runtime_cases": deleted_cases, "deleted_case_keys": deleted_case_keys, "deleted_documents": deleted_documents, "orphaned_batch_documents": orphaned_batch_documents, "deleted_derived_artifacts": deleted_derived_artifacts, }, ) return { "message": "Lote eliminado correctamente.", "batch_id": batch_id, "deleted_case_keys": deleted_case_keys, "deleted_documents": deleted_documents, "orphaned_batch_documents": orphaned_batch_documents, "deleted_derived_artifacts": deleted_derived_artifacts, "deleted_batch_files": deleted_files, "deleted_batch_cases": deleted_cases, "deleted_batch_records": deleted_batches, }