"""Casos de uso de escritura y orquestación del flujo batch.""" from __future__ import annotations from dataclasses import dataclass from datetime import tzinfo from typing import TYPE_CHECKING, Any import logging from pathlib import Path 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, BatchUploadAcceptedView, ManualAssociationResultView, ManualResolutionMetadata, MaterializationResultView, QueueBatchEpicrisisResult, QueueBatchEpicrisisExcelResult, ) from app.batch_processing.application.utils import ( build_associated_user, build_case_key, normalize_case_key_list, now_iso, ) from app.batch_processing.domain.models import ( ASSOCIATION_SOURCE_AUTO, ASSOCIATION_SOURCE_MANUAL, BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, BATCH_STATUS_FALLIDO, BATCH_STATUS_PREPARANDO, BATCH_STATUS_PROCESANDO, BATCH_STATUS_RECIBIDO, CLINICAL_STATUS_COMPLETADO, CLINICAL_STATUS_EN_COLA, CLINICAL_STATUS_FALLIDO, CLINICAL_STATUS_PROCESANDO, EPICRISIS_STATUS_COMPLETADO, EPICRISIS_STATUS_COMPLETADO_CON_ERRORES, EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PENDIENTE, EPICRISIS_STATUS_PROCESANDO, FILE_STATUS_ASOCIADO, FILE_STATUS_CLASIFICADO, FILE_STATUS_FALLIDO, FILE_STATUS_PENDIENTE, FILE_STATUS_PENDIENTE_VALIDACION, FILE_STATUS_PROCESANDO, ) from app.batch_processing.domain.ports import ( ArchiveExtractor, BatchArtifactCleaner, BatchArchiveStore, BatchCaseRepository, BatchFileRepository, BatchJobDispatcher, BatchReportStore, BatchRepository, CaseAssociationService, DocumentClassifier, TextExtractor, ) 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 logger = logging.getLogger(__name__) def _default_excel_filename(batch_id: str) -> str: return f"epicrisis_lote_{batch_id}.xlsx" @dataclass(frozen=True) class ClinicalDocumentPayload: """Payload mínimo que la capa de aplicación entrega al servicio clínico.""" raw_text: str detected_type: str username: str original_name: str case_key: str = "" case_number: str = "" patient_id: str = "" batch_id: str = "" batch_file_id: str = "" ingestion_source: str = "manual" provided_patient_name: str = "" @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: 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]: 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.") 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) batch = self.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() @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 [] 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_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: file_record = self.batch_file_repository.get_file(file_id) if not file_record or file_record.get("batch_id") != batch_id: return current_status = file_record.get("status") if current_status == FILE_STATUS_FALLIDO and file_record.get("error"): return if current_status in { FILE_STATUS_CLASIFICADO, FILE_STATUS_ASOCIADO, FILE_STATUS_PENDIENTE_VALIDACION, } and file_record.get("text_preview"): return self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( status=FILE_STATUS_PROCESANDO, updated_at=now_iso(self.colombia_tz), ).to_document(), ) try: text = self.text_extractor.extract_text(file_record["stored_path"]) if not text.strip(): raise ValueError("El PDF no contiene texto extraíble.") detected_type = self.document_classifier.classify( file_record.get("original_name", ""), text, ) signals = self.association_service.extract_signals( file_record.get("original_name", ""), text, detected_type, ) self.batch_file_repository.update_file( file_id, BatchFileUpdatePayload( status=FILE_STATUS_CLASIFICADO, detected_type=detected_type, text_preview=text[:2500], extracted_text=text, patient_name=signals.patient_name, patient_id=signals.patient_id, case_number=signals.case_number, service_date=signals.service_date, procedure_code=signals.procedure_code, procedure_description=signals.procedure_description, associated_user=build_associated_user( signals.patient_id, signals.patient_name, ), association_source=ASSOCIATION_SOURCE_AUTO, evidence=list(signals.evidence), score_breakdown={}, top_candidates=[], manual_resolution={}, error="", updated_at=now_iso(self.colombia_tz), ).to_document(), ) 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 def execute(self, batch_id: str) -> list[dict[str, Any]]: batch = self.batch_repository.get_batch(batch_id) or {} 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 @dataclass class FinalizeBatchUseCase: """Ejecuta la asociación global y deja el lote listo para el paso clínico.""" 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]: 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 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 {} 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 {} if ( file_record.get("clinical_status") == CLINICAL_STATUS_COMPLETADO and file_record.get("analysis_document_id") ): 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( ClinicalDocumentPayload( 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", ""), batch_id=batch_id, batch_file_id=file_id, ingestion_source="batch", ) ) 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_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: 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 {} 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("Lote no encontrado.") batch_status = str(batch.get("status") or "") is_terminal_batch = batch_status in { BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, } if not is_terminal_batch: 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.") 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 self.batch_case_repository.list_cases(batch_id): 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 timestamp = now_iso(self.colombia_tz) skipped_total = ( skipped_completed_count + skipped_inflight_count + skipped_not_ready_count ) initial_status = ( EPICRISIS_STATUS_EN_COLA if 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(queued_case_keys), bulk_epicrisis_completed_count=0, bulk_epicrisis_failed_count=0, bulk_epicrisis_skipped_count=skipped_total, bulk_epicrisis_case_keys=queued_case_keys, ) if persist: for case_key in 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 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(queued_case_keys), skipped_completed_count=skipped_completed_count, skipped_inflight_count=skipped_inflight_count, skipped_not_ready_count=skipped_not_ready_count, case_keys=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("Lote no encontrado.") batch_status = str(batch.get("status") or "") is_terminal_batch = batch_status in { BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, } if not is_terminal_batch: 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.") cases = self.batch_case_repository.list_cases(batch_id) 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.") 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("Lote no encontrado.") 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: case_key = str(case.get("case_key") or "").strip() if not case_key: continue epicrisis_status = str(case.get("epicrisis_status") or EPICRISIS_STATUS_PENDIENTE) omission_reason = "" context: dict[str, Any] | None = None included = False if epicrisis_status == EPICRISIS_STATUS_COMPLETADO: try: cached = self.case_epicrisis_service.get_cached_case_context(username, case_key) if cached and isinstance(cached.get("contexto"), dict): context = dict(cached["contexto"]) else: context = self.case_epicrisis_service.cache_case_context( username, case_key, regen=False, ) included = isinstance(context, dict) and bool(context) if not included: omission_reason = "No se encontró contexto de epicrisis." except Exception as exc: omission_reason = str(exc) or "No se pudo reconstruir la epicrisis." else: omission_reason = f"Estado de epicrisis no exportable: {epicrisis_status or 'desconocido'}." if included: included_count += 1 else: omitted_count += 1 rows.append( self.workbook_builder_row( case=case, case_key=case_key, epicrisis_status=epicrisis_status, included=included, omission_reason=omission_reason, context=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_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]: file_record = self.batch_file_repository.get_file(file_id) 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.") selected_case = None normalized_case_key = case_key.strip() if normalized_case_key and normalized_case_key != "__new__": selected_case = next( ( item for item in self.refresh_cases_use_case.batch_case_repository.list_cases(batch_id) if item.get("case_key") == normalized_case_key ), None, ) if not selected_case: raise ValueError("El caso seleccionado no existe en el lote.") 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.") previous_evidence = list(file_record.get("evidence", [])) if "validacion_manual" not in previous_evidence: previous_evidence.append("validacion_manual") 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=resolved_case_key, patient_name=resolved_patient_name, patient_id=resolved_patient_id, procedure_code=resolved_procedure_code, associated_user=build_associated_user( resolved_patient_id, resolved_patient_name, ), association_source=ASSOCIATION_SOURCE_MANUAL, confidence=max(float(file_record.get("confidence", 0.0)), 1.0), evidence=previous_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 {} return ManualAssociationResultView.from_record(updated_file).to_document()