"""Funciones puras para calcular agregados y payloads del módulo batch.""" from __future__ import annotations from collections.abc import Mapping, Sequence from typing import Any from app.batch_processing.application.models import ( BatchBulkEpicrisisPayload, BatchCaseRecord, BatchUpdatePayload, ) from app.batch_processing.application.utils import ( build_associated_user, normalize_case_key_list, slugify, ) from app.batch_processing.domain.models import ( BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, BATCH_STATUS_FALLIDO, BATCH_STATUS_PROCESANDO, CLINICAL_STATUS_COMPLETADO, CLINICAL_STATUS_EN_COLA, CLINICAL_STATUS_FALLIDO, CLINICAL_STATUS_PENDIENTE, CLINICAL_STATUS_PROCESANDO, EPICRISIS_STATUS_COMPLETADO, EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_FALLIDO, EPICRISIS_STATUS_PENDIENTE, EPICRISIS_STATUS_PROCESANDO, FILE_STATUS_ASOCIADO, FILE_STATUS_FALLIDO, FILE_STATUS_PENDIENTE_VALIDACION, ) def build_batch_totals_update( files: Sequence[Mapping[str, Any]], *, updated_at: str, force_status: str | None = None, ) -> BatchUpdatePayload: """Calcula el estado agregado del lote a partir de los estados de archivo.""" total = len(files) failed = sum(1 for item in files if item.get("status") == FILE_STATUS_FALLIDO) pending_validation = sum( 1 for item in files if item.get("status") == FILE_STATUS_PENDIENTE_VALIDACION ) associated = sum(1 for item in files if item.get("status") == FILE_STATUS_ASOCIADO) processed = sum( 1 for item in files if item.get("status") in { FILE_STATUS_ASOCIADO, FILE_STATUS_PENDIENTE_VALIDACION, FILE_STATUS_FALLIDO, } ) associated_files = [item for item in files if item.get("status") == FILE_STATUS_ASOCIADO] clinical_completed = sum( 1 for item in associated_files if item.get("clinical_status") == CLINICAL_STATUS_COMPLETADO ) clinical_failed = sum( 1 for item in associated_files if item.get("clinical_status") == CLINICAL_STATUS_FALLIDO ) clinical_pending = sum( 1 for item in associated_files if item.get("clinical_status") in { CLINICAL_STATUS_PENDIENTE, CLINICAL_STATUS_EN_COLA, CLINICAL_STATUS_PROCESANDO, } ) status = force_status if status is None: if total == 0: status = BATCH_STATUS_FALLIDO elif processed < total or clinical_pending > 0: status = BATCH_STATUS_PROCESANDO elif processed == total and failed == total: status = BATCH_STATUS_FALLIDO elif failed > 0 or pending_validation > 0: status = BATCH_STATUS_COMPLETADO_CON_ERRORES else: status = BATCH_STATUS_COMPLETADO return BatchUpdatePayload( status=status, total_files=total, processed_files=processed, failed_files=failed, pending_validation_files=pending_validation, associated_files=associated, clinical_processed_files=clinical_completed, clinical_failed_files=clinical_failed, clinical_pending_files=clinical_pending, updated_at=updated_at, ) def build_bulk_epicrisis_payload( batch: Mapping[str, Any] | None, cases: Sequence[Mapping[str, Any]], ) -> BatchBulkEpicrisisPayload: """Sincroniza el agregado de epicrisis del lote con el estado real por caso.""" batch_data = batch or {} current_status = str(batch_data.get("bulk_epicrisis_status") or EPICRISIS_STATUS_PENDIENTE) target_case_keys = normalize_case_key_list(batch_data.get("bulk_epicrisis_case_keys")) total_target = int(batch_data.get("bulk_epicrisis_total_target") or len(target_case_keys) or 0) skipped_count = int(batch_data.get("bulk_epicrisis_skipped_count") or 0) case_index = { str(item.get("case_key") or "").strip(): item for item in cases if str(item.get("case_key") or "").strip() } completed_count = 0 failed_count = 0 queued_count = 0 processing_count = 0 pending_count = 0 for case_key in target_case_keys: case_status = str( (case_index.get(case_key) or {}).get("epicrisis_status") or EPICRISIS_STATUS_PENDIENTE ) if case_status == EPICRISIS_STATUS_COMPLETADO: completed_count += 1 elif case_status == EPICRISIS_STATUS_FALLIDO: failed_count += 1 elif case_status == EPICRISIS_STATUS_PROCESANDO: processing_count += 1 elif case_status == EPICRISIS_STATUS_EN_COLA: queued_count += 1 else: pending_count += 1 if total_target <= 0: bulk_status = current_status if current_status else EPICRISIS_STATUS_PENDIENTE if bulk_status not in { EPICRISIS_STATUS_PENDIENTE, EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PROCESANDO, EPICRISIS_STATUS_COMPLETADO, EPICRISIS_STATUS_FALLIDO, }: bulk_status = EPICRISIS_STATUS_PENDIENTE elif processing_count > 0 or completed_count > 0 or failed_count > 0: bulk_status = ( EPICRISIS_STATUS_PROCESANDO if completed_count + failed_count < total_target else ( EPICRISIS_STATUS_FALLIDO if failed_count > 0 and completed_count == 0 else EPICRISIS_STATUS_COMPLETADO ) ) elif queued_count > 0: bulk_status = EPICRISIS_STATUS_EN_COLA elif pending_count > 0: bulk_status = ( current_status if current_status in {EPICRISIS_STATUS_EN_COLA, EPICRISIS_STATUS_PROCESANDO} else EPICRISIS_STATUS_PENDIENTE ) else: bulk_status = ( EPICRISIS_STATUS_FALLIDO if failed_count > 0 and completed_count == 0 else EPICRISIS_STATUS_COMPLETADO ) return BatchBulkEpicrisisPayload( bulk_epicrisis_status=bulk_status, bulk_epicrisis_job_id=str(batch_data.get("bulk_epicrisis_job_id") or ""), bulk_epicrisis_requested_at=str(batch_data.get("bulk_epicrisis_requested_at") or ""), bulk_epicrisis_total_target=total_target, bulk_epicrisis_completed_count=completed_count, bulk_epicrisis_failed_count=failed_count, bulk_epicrisis_skipped_count=skipped_count, bulk_epicrisis_case_keys=target_case_keys, ) def build_cases_from_files( batch_id: str, files: Sequence[Mapping[str, Any]], *, username: str = "", batch_status: str = "", pending_validation_files: int = 0, existing_cases: Mapping[str, Mapping[str, Any]] | None = None, ) -> list[BatchCaseRecord]: """Reconstruye el agregado por caso usando los documentos asociados del lote.""" existing_case_map = dict(existing_cases or {}) grouped: dict[str, BatchCaseRecord] = {} for file_record in files: if file_record.get("status") != FILE_STATUS_ASOCIADO: continue case_key = str(file_record.get("case_key") or "").strip() if not case_key: continue case = grouped.get(case_key) if case is None: case = BatchCaseRecord.from_existing( batch_id=batch_id, username=username, case_key=case_key, existing_case=existing_case_map.get(case_key), ) grouped[case_key] = case for field_name in ( "patient_name", "patient_id", "case_number", "service_date", "procedure_code", "procedure_description", "associated_user", ): current_value = getattr(case, field_name) candidate_value = str(file_record.get(field_name) or "").strip() if not current_value and candidate_value: setattr(case, field_name, candidate_value) file_id = str(file_record.get("_id") or "").strip() if file_id and file_id not in case.file_ids: case.file_ids.append(file_id) analysis_document_id = str(file_record.get("analysis_document_id") or "").strip() if analysis_document_id and analysis_document_id not in case.analysis_document_ids: case.analysis_document_ids.append(analysis_document_id) detected_type = str(file_record.get("detected_type") or "").strip() if detected_type and detected_type not in case.detected_types: case.detected_types.append(detected_type) case.document_count += 1 clinical_status = file_record.get("clinical_status") if clinical_status == CLINICAL_STATUS_COMPLETADO: case.processed_document_count += 1 elif clinical_status == CLINICAL_STATUS_FALLIDO: case.clinical_failed_count += 1 epicrisis_status = str(file_record.get("epicrisis_status") or "").strip() if epicrisis_status: case.epicrisis_status = epicrisis_status epicrisis_job_id = str(file_record.get("epicrisis_job_id") or "").strip() if epicrisis_job_id: case.epicrisis_job_id = epicrisis_job_id updated_at = str(file_record.get("updated_at") or "").strip() if updated_at and updated_at > case.updated_at: case.updated_at = updated_at for case in grouped.values(): if not case.associated_user: case.associated_user = build_associated_user(case.patient_id, case.patient_name) total_terminal = ( case.processed_document_count + case.clinical_failed_count ) >= case.document_count batch_terminal = batch_status in { BATCH_STATUS_COMPLETADO, BATCH_STATUS_COMPLETADO_CON_ERRORES, } case.ready_for_epicrisis = bool( case.analysis_document_ids and total_terminal and batch_terminal and int(pending_validation_files or 0) == 0 ) if not case.epicrisis_status: case.epicrisis_status = EPICRISIS_STATUS_PENDIENTE return list(grouped.values()) def build_case_slug_hint(patient_id: str, case_number: str, patient_name: str) -> str: """Expone una ayuda mínima para depuración de claves de caso.""" return "-".join( [part for part in (slugify(patient_id), slugify(case_number), slugify(patient_name)) if part] )