"""Modelos serializables para la capa de aplicación de carga individual.""" from __future__ import annotations from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field from app.individual_ingestion.domain.models import ( INGESTION_SOURCE_INDIVIDUAL, UPLOAD_STATUS_PRECHECK_EN_COLA, ) def _string(value: Any, *, strip: bool = True) -> str: if value is None: return "" text = str(value) return text.strip() if strip else text def _list_value(value: Any) -> list[Any]: return list(value or []) def _bool_value(value: Any) -> bool: return bool(value) class ApplicationModel(BaseModel): model_config = ConfigDict(extra="ignore") def to_document(self, *, exclude_none: bool = True) -> dict[str, Any]: return self.model_dump(mode="python", exclude_none=exclude_none) class IndividualUploadCreatePayload(ApplicationModel): usuario: str original_name: str selected_document_type: str provided_case_key: str = "" status: str = UPLOAD_STATUS_PRECHECK_EN_COLA ingestion_source: str = INGESTION_SOURCE_INDIVIDUAL source_file_hash: str stored_path: str file_size: int = 0 created_at: str updated_at: str detected_document_type: str = "" effective_document_type: str = "" extracted_text: str = "" extraction_metadata: dict[str, Any] = Field(default_factory=dict) classification_decision: dict[str, Any] = Field(default_factory=dict) historia_processing: dict[str, Any] = Field(default_factory=dict) analysis_quality: dict[str, Any] = Field(default_factory=dict) analysis_provider: str = "" analysis_model_name: str = "" failure: dict[str, Any] = Field(default_factory=dict) can_retry: bool = False retry_count: int = 0 last_retry_at: str = "" pending_actions: list[str] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) contradictions: list[str] = Field(default_factory=list) impact_summary: list[str] = Field(default_factory=list) provisional_case_number: str = "" provisional_patient_id: str = "" provisional_patient_name: str = "" redacted_identity_fields: list[Literal["patient_id", "patient_name"]] = Field(default_factory=list) confirmed_case_number: str = "" confirmed_patient_id: str = "" confirmed_patient_name: str = "" case_key: str = "" session_id: str = "" base_upload_id: str = "" depends_on_upload_id: str = "" session_case_key: str = "" session_status: str = "" session_origin: str = "" case_selection_locked: bool = False waiting_for_base_upload_id: str = "" blocked_by_base_failure: bool = False selected_type_override_confirmed: bool = False case_override_confirmed: bool = False precheck_completed_at: str = "" materialization_started_at: str = "" completed_at: str = "" cancelled_at: str = "" deleted_at: str = "" deletion_reason: str = "" deletion_document_id: str = "" deletion_case_key: str = "" deleted_by: str = "" failed_at: str = "" clinical_document_id: str = "" duplicate_of_upload_id: str = "" override_audit: dict[str, Any] = Field(default_factory=dict) error: str = "" audit_only: bool = False class IndividualUploadUpdatePayload(ApplicationModel): status: str | None = None updated_at: str | None = None detected_document_type: str | None = None effective_document_type: str | None = None extracted_text: str | None = None extraction_metadata: dict[str, Any] | None = None classification_decision: dict[str, Any] | None = None historia_processing: dict[str, Any] | None = None analysis_quality: dict[str, Any] | None = None analysis_provider: str | None = None analysis_model_name: str | None = None failure: dict[str, Any] | None = None can_retry: bool | None = None retry_count: int | None = None last_retry_at: str | None = None pending_actions: list[str] | None = None warnings: list[str] | None = None contradictions: list[str] | None = None impact_summary: list[str] | None = None provisional_case_number: str | None = None provisional_patient_id: str | None = None provisional_patient_name: str | None = None redacted_identity_fields: list[Literal["patient_id", "patient_name"]] | None = None confirmed_case_number: str | None = None confirmed_patient_id: str | None = None confirmed_patient_name: str | None = None case_key: str | None = None session_id: str | None = None base_upload_id: str | None = None depends_on_upload_id: str | None = None session_case_key: str | None = None session_status: str | None = None session_origin: str | None = None case_selection_locked: bool | None = None waiting_for_base_upload_id: str | None = None blocked_by_base_failure: bool | None = None selected_type_override_confirmed: bool | None = None case_override_confirmed: bool | None = None precheck_completed_at: str | None = None materialization_started_at: str | None = None completed_at: str | None = None cancelled_at: str | None = None deleted_at: str | None = None deletion_reason: str | None = None deletion_document_id: str | None = None deletion_case_key: str | None = None deleted_by: str | None = None failed_at: str | None = None clinical_document_id: str | None = None duplicate_of_upload_id: str | None = None stored_path: str | None = None override_audit: dict[str, Any] | None = None error: str | None = None audit_only: bool | None = None class IndividualUploadAcceptedView(ApplicationModel): upload_id: str status: str created_at: str poll_url: str duplicate: bool = False message: str = "" class IndividualUploadStatusView(ApplicationModel): upload_id: str original_name: str = "" status: str selected_document_type: str detected_document_type: str = "" effective_document_type: str = "" provided_case_key: str = "" case_key: str = "" session_id: str = "" base_upload_id: str = "" depends_on_upload_id: str = "" session_case_key: str = "" session_status: str = "" session_origin: str = "" case_selection_locked: bool = False waiting_for_base_upload_id: str = "" blocked_by_base_failure: bool = False created_at: str = "" updated_at: str = "" deleted_at: str = "" deletion_reason: str = "" deletion_document_id: str = "" deletion_case_key: str = "" deleted_by: str = "" pending_actions: list[str] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) contradictions: list[str] = Field(default_factory=list) impact_summary: list[str] = Field(default_factory=list) provisional_case_number: str = "" provisional_patient_id: str = "" provisional_patient_name: str = "" redacted_identity_fields: list[Literal["patient_id", "patient_name"]] = Field(default_factory=list) confirmed_case_number: str = "" confirmed_patient_id: str = "" confirmed_patient_name: str = "" selected_type_override_confirmed: bool = False case_override_confirmed: bool = False clinical_document_id: str = "" duplicate_of_upload_id: str = "" error: str = "" audit_only: bool = False extraction_metadata: dict[str, Any] = Field(default_factory=dict) classification_decision: dict[str, Any] = Field(default_factory=dict) historia_processing: dict[str, Any] = Field(default_factory=dict) analysis_quality: dict[str, Any] = Field(default_factory=dict) analysis_provider: str = "" analysis_model_name: str = "" failure: dict[str, Any] = Field(default_factory=dict) can_retry: bool = False retry_count: int = 0 last_retry_at: str = "" retry_url: str = "" @classmethod def from_record(cls, upload_id: str, item: dict[str, Any]) -> IndividualUploadStatusView: return cls( upload_id=upload_id, original_name=_string(item.get("original_name")), status=_string(item.get("status")) or UPLOAD_STATUS_PRECHECK_EN_COLA, selected_document_type=_string(item.get("selected_document_type")), detected_document_type=_string(item.get("detected_document_type")), effective_document_type=_string( item.get("effective_document_type") or item.get("detected_document_type") or item.get("selected_document_type") ), provided_case_key=_string(item.get("provided_case_key")), case_key=_string(item.get("case_key")), session_id=_string(item.get("session_id")), base_upload_id=_string(item.get("base_upload_id")), depends_on_upload_id=_string(item.get("depends_on_upload_id")), session_case_key=_string(item.get("session_case_key")), session_status=_string(item.get("session_status")), session_origin=_string(item.get("session_origin")), case_selection_locked=_bool_value(item.get("case_selection_locked")), waiting_for_base_upload_id=_string(item.get("waiting_for_base_upload_id")), blocked_by_base_failure=_bool_value(item.get("blocked_by_base_failure")), created_at=_string(item.get("created_at")), updated_at=_string(item.get("updated_at")), deleted_at=_string(item.get("deleted_at")), deletion_reason=_string(item.get("deletion_reason")), deletion_document_id=_string(item.get("deletion_document_id")), deletion_case_key=_string(item.get("deletion_case_key")), deleted_by=_string(item.get("deleted_by")), pending_actions=_list_value(item.get("pending_actions")), warnings=_list_value(item.get("warnings")), contradictions=_list_value(item.get("contradictions")), impact_summary=_list_value(item.get("impact_summary")), provisional_case_number=_string(item.get("provisional_case_number")), provisional_patient_id=_string(item.get("provisional_patient_id")), provisional_patient_name=_string(item.get("provisional_patient_name")), redacted_identity_fields=_list_value(item.get("redacted_identity_fields")), confirmed_case_number=_string(item.get("confirmed_case_number")), confirmed_patient_id=_string(item.get("confirmed_patient_id")), confirmed_patient_name=_string(item.get("confirmed_patient_name")), selected_type_override_confirmed=_bool_value(item.get("selected_type_override_confirmed")), case_override_confirmed=_bool_value(item.get("case_override_confirmed")), clinical_document_id=_string(item.get("clinical_document_id")), duplicate_of_upload_id=_string(item.get("duplicate_of_upload_id")), error=_string(item.get("error")), audit_only=_bool_value(item.get("audit_only")), extraction_metadata=dict(item.get("extraction_metadata") or {}), classification_decision=dict(item.get("classification_decision") or {}), historia_processing=dict(item.get("historia_processing") or {}), analysis_quality=dict(item.get("analysis_quality") or {}), analysis_provider=_string(item.get("analysis_provider")), analysis_model_name=_string(item.get("analysis_model_name")), failure=dict(item.get("failure") or {}), can_retry=_bool_value(item.get("can_retry")), retry_count=int(item.get("retry_count") or 0), last_retry_at=_string(item.get("last_retry_at")), retry_url=f"/api/cargue-individual/{upload_id}/reintentar" if _bool_value(item.get("can_retry")) else "", ) class IndividualUploadSessionView(ApplicationModel): session_id: str active_case_key: str = "" base_upload_id: str = "" session_status: str = "" base_ready: bool = False blocked_by_base_failure: bool = False updated_at: str = "" uploads: list[dict[str, Any]] = Field(default_factory=list)