from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from app.case_epicrisis.application.cost_ordering import COST_ORDER_VERSION, apply_factura_cost_ordering from app.case_epicrisis.application.curation import ( curation_context_payload, reconcile_case_objective_data, ) from app.case_epicrisis.application.medication_pertinence import ( apply_clinical_pertinence_reviews, build_medication_glosa_findings, ) from app.case_epicrisis.application.utils import ( build_ayudas_diagnosticas, build_medication_display_list, build_pdf_preparation_procedimientos, coerce_pdf_draft, coerce_pdf_preparation_user_metadata, extract_recomendaciones_medicas, extraer_imagenes_diagnosticas, extraer_procedimientos_factura, format_ayuda_diagnostica_presentacion, merge_soat_results, normalize_context_payload, normalize_factura_medicamentos, serialize_doc, strip_html_tags, ) from app.case_epicrisis.domain.models import ( CaseEpicrisisDocuments, ClinicalSnapshot, CodingSnapshot, EpicrisisEvidenceSnapshot, EpicrisisGenerationBlockedError, EpicrisisRuleEvaluation, EpicrisisRuleEvaluationMode, EpicrisisRuleStatus, EpicrisisSummarySourceUnit, PdfPreparationSoatGroup, PreservedEpicrisisState, ) from app.case_epicrisis.domain.ports import ( CaseDocumentsReader, CaseEpicrisisCacheRepository, EpicrisisCodingGateway, EpicrisisPreGenerationRule, EpicrisisSummaryComposer, MedicationPertinenceClinicalReviewer, ) from app.services.clinical_document_projection import ( get_structured_document_model, render_document_analysis_html, ) from app.services.clinical_processing import ( extraer_antecedentes_historia, extraer_antecedentes_historia_estructurados, extraer_medicamentos_historia, extraer_metadatos_historia, extraer_procedimientos_historia, extraer_secciones_quirurgicas, ) from app.services.historia_medicamentos import consolidate_medication_items, medication_from_legacy _SUMMARY_COMPOSITION_WARNING = ( "No fue posible integrar cronológicamente los resúmenes. Se muestran por separado sin pérdida de " "contenido. Puedes regenerar la epicrisis para volver a intentarlo." ) def _summary_source_identifier(document_type: str, document: dict[str, Any] | None) -> str: raw_identifier = "" if isinstance(document, dict): raw_identifier = str( document.get("_id") or document.get("document_id") or document.get("nombre_archivo") or document.get("case_key") or "" ).strip() return f"{document_type}:{raw_identifier or 'sin-identificador'}" def build_integrated_summary_context( *, username: str, historia: dict[str, Any] | None, quirurgico: dict[str, Any] | None, metadatos_hc: dict[str, Any], summary_composer: EpicrisisSummaryComposer | None, ) -> dict[str, Any]: historia_summary = str(metadatos_hc.get("resumen") or "").strip() historia_date = str( metadatos_hc.get("fecha_ingreso") or (historia or {}).get("fecha_analisis") or "" ).strip() sources_metadata: list[dict[str, Any]] = [] historia_source: EpicrisisSummarySourceUnit | None = None if historia_summary: historia_source = EpicrisisSummarySourceUnit( identifier=_summary_source_identifier("historia_clinica", historia), document_type="historia_clinica", summary=historia_summary, available_date=historia_date, ) sources_metadata.append( { "identificador": historia_source.identifier, "tipo_documento": historia_source.document_type, "fecha_disponible": historia_source.available_date, "resumen_disponible": True, } ) if not quirurgico: return { "estado": "solo_historia", "texto": historia_summary, "parrafos": [historia_summary] if historia_summary else [], "fuentes": sources_metadata, "advertencia": None, } structured_qx = get_structured_document_model(quirurgico) surgical_summary = str(getattr(structured_qx, "resumen_clinico", "") or "").strip() surgical_date = str( getattr(structured_qx, "fecha_procedimiento", "") or quirurgico.get("fecha_analisis") or "" ).strip() qx_identifier = _summary_source_identifier("quirurgico", quirurgico) sources_metadata.append( { "identificador": qx_identifier, "tipo_documento": "quirurgico", "fecha_disponible": surgical_date, "resumen_disponible": bool(surgical_summary), } ) if not historia_source or not surgical_summary: return { "estado": "solo_historia", "texto": historia_summary, "parrafos": [historia_summary] if historia_summary else [], "fuentes": sources_metadata, "advertencia": None, } qx_source = EpicrisisSummarySourceUnit( identifier=qx_identifier, document_type="quirurgico", summary=surgical_summary, available_date=surgical_date, ) sources = [historia_source, qx_source] try: if summary_composer is None: raise RuntimeError("Compositor no configurado") composition = summary_composer.compose(username=username, sources=sources) return { "estado": "integrado", "texto": composition.narrative, "parrafos": [composition.narrative], "fuentes": sources_metadata, "advertencia": None, } except Exception: paragraphs = [historia_summary, surgical_summary] return { "estado": "fuentes_separadas", "texto": "\n\n".join(paragraphs), "parrafos": paragraphs, "fuentes": sources_metadata, "advertencia": _SUMMARY_COMPOSITION_WARNING, } def _clean_unique_texts(values: list[str]) -> list[str]: seen: set[str] = set() unique: list[str] = [] for raw in values: value = " ".join(str(raw or "").split()).strip() key = value.casefold() if not value or key in seen: continue seen.add(key) unique.append(value) return unique def _generic_document_snapshot(documents: list[dict[str, Any]]) -> tuple[list[str], list[str]]: summaries: list[str] = [] references: list[str] = [] for doc in documents: structured = get_structured_document_model(doc) if structured is not None: summary = getattr(structured, "resumen", "") or getattr(structured, "observaciones", "") if summary: summaries.append(str(summary)) for item in getattr(structured, "referencias_diagnosticas", []) or []: nombre = str(getattr(item, "nombre", "") or "").strip() concepto = str(getattr(item, "concepto", "") or "").strip() references.append(" - ".join([part for part in (nombre, concepto) if part])) continue raw_summary = str(doc.get("descripcion") or "").strip() if raw_summary: summaries.append(raw_summary) html_summary = strip_html_tags( str(doc.get("analisis_html") or render_document_analysis_html(doc) or "") ) if html_summary: summaries.append(html_summary[:400]) return _clean_unique_texts(summaries), _clean_unique_texts(references) def _document_types_available(documents: CaseEpicrisisDocuments) -> list[str]: available: list[str] = [] if documents.historia: available.append("historia_clinica") if documents.quirurgico: available.append("quirurgico") if documents.factura: available.append("factura") if documents.radiologia: available.append("radiologia") if documents.laboratorio: available.append("laboratorio") if documents.generico: available.append("generico") if documents.prescripcion: available.append("prescripcion") return available def _extract_prescription_medications(documents: list[dict[str, Any]]) -> list[dict[str, Any]]: medications: list[dict[str, Any]] = [] for document in documents: structured = get_structured_document_model(document) for item in getattr(structured, "medicamentos", []) or []: medication = medication_from_legacy( { "codigo_referencia": getattr(item, "codigo", "") or "", "nombre": getattr(item, "nombre", "") or "", "dosis": getattr(item, "dosis", "") or "", "posologia": getattr(item, "posologia", "") or "", "via": getattr(item, "via", "") or "", "frecuencia": getattr(item, "frecuencia", "") or "", "duracion": getattr(item, "duracion", "") or "", "cantidad": getattr(item, "cantidad", "") or "", "tipo_uso": "formulado", "fuente": "prescripcion", "texto_original": getattr(item, "diagnostico_contexto", "") or "", }, fuente="prescripcion", ) if medication: medications.append(medication) return consolidate_medication_items(medications) def _build_default_rule_evaluation() -> EpicrisisRuleEvaluation: return EpicrisisRuleEvaluation( status=EpicrisisRuleStatus.PASSED, generation_allowed=True, ready_for_epicrisis=True, findings=[], required_documents=[], missing_documents=[], evaluated_at=datetime.now(UTC), evaluation_mode=EpicrisisRuleEvaluationMode.DETERMINISTIC_ONLY, ) @dataclass class EvaluateEpicrisisRulesUseCase: rules: list[EpicrisisPreGenerationRule] def execute(self, snapshot: EpicrisisEvidenceSnapshot) -> EpicrisisRuleEvaluation: if not self.rules: return _build_default_rule_evaluation() evaluations = [rule.evaluate(snapshot) for rule in self.rules] findings = [item for evaluation in evaluations for item in evaluation.findings] required_documents = _clean_unique_texts( [item for evaluation in evaluations for item in evaluation.required_documents] ) missing_documents = _clean_unique_texts( [item for evaluation in evaluations for item in evaluation.missing_documents] ) status = EpicrisisRuleStatus.PASSED generation_allowed = True ready_for_epicrisis = True if any(item.status == EpicrisisRuleStatus.BLOCKED for item in evaluations): status = EpicrisisRuleStatus.BLOCKED generation_allowed = False ready_for_epicrisis = False elif any(item.status == EpicrisisRuleStatus.WARNING for item in evaluations): status = EpicrisisRuleStatus.WARNING evaluation_mode = ( EpicrisisRuleEvaluationMode.DETERMINISTIC_PLUS_LLM if any( item.evaluation_mode == EpicrisisRuleEvaluationMode.DETERMINISTIC_PLUS_LLM for item in evaluations ) else EpicrisisRuleEvaluationMode.DETERMINISTIC_ONLY ) evaluated_at = max(item.evaluated_at for item in evaluations) return EpicrisisRuleEvaluation( status=status, generation_allowed=generation_allowed, ready_for_epicrisis=ready_for_epicrisis, findings=findings, required_documents=required_documents, missing_documents=missing_documents, evaluated_at=evaluated_at, evaluation_mode=evaluation_mode, ) @dataclass class BuildCaseEpicrisisContextUseCase: documents_reader: CaseDocumentsReader coding_gateway: EpicrisisCodingGateway rules_evaluator: EvaluateEpicrisisRulesUseCase | None = None medication_pertinence_reviewer: MedicationPertinenceClinicalReviewer | None = None summary_composer: EpicrisisSummaryComposer | None = None def _load_documents(self, username: str, case_key: str) -> CaseEpicrisisDocuments: documents = CaseEpicrisisDocuments.from_payload( self.documents_reader.get_case_documents(username, case_key) ) if not documents.has_clinical_content(): raise ValueError("No se encontraron documentos clinicos para el caso.") return documents def _build_evidence_snapshot( self, documents: CaseEpicrisisDocuments, case_key: str, ) -> EpicrisisEvidenceSnapshot: historia_doc = documents.historia historia_html = documents.historia_html historia_metadatos = extraer_metadatos_historia(historia_doc or historia_html) antecedentes_historia = extraer_antecedentes_historia(historia_doc or historia_html) procedimientos_historia = extraer_procedimientos_historia(historia_doc or historia_html) procedimientos_factura = [ str(item.get("descripcion") or item.get("procedimiento") or "").strip() for item in extraer_procedimientos_factura(documents.factura) if str(item.get("descripcion") or item.get("procedimiento") or "").strip() ] hallazgos_qx, descripcion_qx = extraer_secciones_quirurgicas( documents.quirurgico or documents.quirurgico_html ) ayudas_diagnosticas = [ format_ayuda_diagnostica_presentacion(item) for item in build_ayudas_diagnosticas( { "factura": documents.factura, "radiologia": documents.radiologia, "laboratorio": documents.laboratorio, "generico": documents.generico, "historia": documents.historia, "quirurgico": documents.quirurgico, "hallazgos_quirurgicos": hallazgos_qx, "descripcion_procedimiento": descripcion_qx, } ) ] generico_resumenes, generico_referencias = _generic_document_snapshot(documents.generico) weak_signals = _clean_unique_texts( [ *[ format_ayuda_diagnostica_presentacion(item) for item in build_ayudas_diagnosticas( { "factura": None, "radiologia": documents.radiologia, "laboratorio": documents.laboratorio, "generico": [], "historia": None, "quirurgico": None, "hallazgos_quirurgicos": "", "descripcion_procedimiento": "", } ) ], *generico_resumenes, ] ) return EpicrisisEvidenceSnapshot( case_key=case_key, patient_name=documents.nombre_paciente, case_number=documents.case_number, has_quirurgico_document=bool(documents.quirurgico), document_types_available=_document_types_available(documents), historia_metadatos=historia_metadatos, procedimientos_factura=procedimientos_factura, procedimientos_historia=_clean_unique_texts([str(item) for item in procedimientos_historia]), antecedentes_historia=_clean_unique_texts([str(item) for item in antecedentes_historia]), historia_resumen=str(historia_metadatos.get("resumen") or "").strip(), historia_motivo_consulta=str(historia_metadatos.get("motivo_consulta") or "").strip(), generico_resumenes=generico_resumenes, generico_referencias=generico_referencias, ayudas_diagnosticas=_clean_unique_texts(ayudas_diagnosticas), hallazgos_quirurgicos=str(hallazgos_qx or "").strip(), descripcion_procedimiento=str(descripcion_qx or "").strip(), support_signals=generico_referencias, weak_signals=weak_signals, ) def _build_preserved_state( self, preserved_context: dict[str, Any] | None, ) -> PreservedEpicrisisState: preserved = normalize_context_payload(preserved_context or {}) return PreservedEpicrisisState( manual_soat_results=preserved.get("manual_soat_results") or [], pdf_draft=coerce_pdf_draft(preserved.get("pdf_draft")), pdf_preparation_user_metadata=coerce_pdf_preparation_user_metadata( preserved.get("pdf_preparation_user_metadata") ), ) def _resolve_blocking_reason(self, evaluation: EpicrisisRuleEvaluation | None) -> str: if evaluation is None: return "" for item in evaluation.findings: if item.blocking: return item.message return evaluation.findings[0].message if evaluation.findings else "" def _build_base_context( self, *, documents: CaseEpicrisisDocuments, case_key: str, preserved_state: PreservedEpicrisisState, rule_evaluation: EpicrisisRuleEvaluation, evidence_snapshot: EpicrisisEvidenceSnapshot, ) -> dict[str, Any]: evaluation_payload = rule_evaluation.model_dump(mode="python", exclude_none=True) blocking_reason = self._resolve_blocking_reason(rule_evaluation) return { "nombre_paciente": documents.nombre_paciente, "case_key": case_key, "case_number": documents.case_number, "historia": serialize_doc(documents.historia), "quirurgico": serialize_doc(documents.quirurgico), "factura": normalize_factura_medicamentos(serialize_doc(documents.factura)), "radiologia": [serialize_doc(doc) for doc in documents.radiologia], "laboratorio": [serialize_doc(doc) for doc in documents.laboratorio], "generico": [serialize_doc(doc) for doc in documents.generico], "prescripcion": [serialize_doc(doc) for doc in documents.prescripcion], "metadatos_hc": evidence_snapshot.historia_metadatos, "antecedentes_hc": list(evidence_snapshot.antecedentes_historia), "procedimientos_hc": list(evidence_snapshot.procedimientos_historia), "medicamentos_hc": [], "medicamentos_hc_display": [], "recomendaciones_medicas": [], "ayudas_diagnosticas": [], "imagenes_diagnosticas": [], "hallazgos_quirurgicos": evidence_snapshot.hallazgos_quirurgicos, "descripcion_procedimiento": evidence_snapshot.descripcion_procedimiento, "soat_resultados": [], "glosa_analisis": "", "procedimientos_factura": [ {"codigo_soat": "", "descripcion": item} for item in evidence_snapshot.procedimientos_factura ], "auto_soat_results": [], "qx_agent_results": [], "manual_soat_results": preserved_state.manual_soat_results, "pdf_draft": preserved_state.pdf_draft, "pdf_preparation_user_metadata": preserved_state.pdf_preparation_user_metadata, "codigos_desde_soat": merge_soat_results([], preserved_state.manual_soat_results), "regen_url": f"/epicrisis?case_key={case_key}®en=1", "epicrisis_cached": False, "orden_costo_version": COST_ORDER_VERSION, "rule_evaluation": evaluation_payload, "epicrisis_generation_allowed": rule_evaluation.generation_allowed, "required_documents": list(rule_evaluation.required_documents), "missing_documents": list(rule_evaluation.missing_documents), "blocking_reason": blocking_reason, } def _build_clinical_snapshot(self, documents: CaseEpicrisisDocuments) -> ClinicalSnapshot: historia_doc = documents.historia quirurgico_doc = documents.quirurgico historia_html = documents.historia_html quirurgico_html = documents.quirurgico_html medicamentos_hc = extraer_medicamentos_historia(historia_doc or historia_html) medicamentos_prescripcion = _extract_prescription_medications(documents.prescripcion) hallazgos_qx, descripcion_qx = extraer_secciones_quirurgicas(quirurgico_doc or quirurgico_html) ayudas_diagnosticas = build_ayudas_diagnosticas( { "factura": documents.factura, "radiologia": documents.radiologia, "laboratorio": documents.laboratorio, "generico": documents.generico, "historia": documents.historia, "quirurgico": documents.quirurgico, "hallazgos_quirurgicos": hallazgos_qx, "descripcion_procedimiento": descripcion_qx, } ) return ClinicalSnapshot( metadatos_hc=extraer_metadatos_historia(historia_doc or historia_html), antecedentes_hc=extraer_antecedentes_historia(historia_doc or historia_html), antecedentes_hc_estructurados=extraer_antecedentes_historia_estructurados( historia_doc or historia_html ), procedimientos_hc=extraer_procedimientos_historia(historia_doc or historia_html), medicamentos_hc=medicamentos_hc, medicamentos_hc_display=build_medication_display_list(medicamentos_hc), recomendaciones_medicas=extract_recomendaciones_medicas(historia_doc), medicamentos_prescripcion=medicamentos_prescripcion, hallazgos_qx=hallazgos_qx, descripcion_qx=descripcion_qx, procedimientos_factura=extraer_procedimientos_factura(documents.factura), ayudas_diagnosticas=ayudas_diagnosticas, imagenes_diagnosticas=[ format_ayuda_diagnostica_presentacion(item) for item in ayudas_diagnosticas if item.get("tipo") == "imagen" ] or extraer_imagenes_diagnosticas(documents.factura, documents.radiologia), ) def _build_auto_soat_results( self, soat_resultados: list[dict[str, Any]], procedimientos_factura: list[dict[str, str]], ) -> list[dict[str, Any]]: auto_soat_results: list[dict[str, Any]] = [] descripciones_soat = [ { "codigo_soat": str(item.get("codigo_soat") or "").strip(), "descripcion": str(item.get("descripcion") or "").strip(), } for item in soat_resultados[:5] if str(item.get("descripcion") or "").strip() ] if descripciones_soat: auto_soat_results = merge_soat_results( auto_soat_results, self.coding_gateway.generate_codes_from_descriptions(descripciones_soat), ) if procedimientos_factura: auto_soat_results = merge_soat_results( auto_soat_results, self.coding_gateway.generate_codes_from_descriptions(procedimientos_factura), ) return auto_soat_results def _build_coding_snapshot( self, documents: CaseEpicrisisDocuments, clinical_snapshot: ClinicalSnapshot, ) -> CodingSnapshot: soat_resultados, glosa_html = self.coding_gateway.build_soat_and_glosa( historia_html=documents.historia_html, qx_html=documents.quirurgico_html, ) return CodingSnapshot( soat_resultados=soat_resultados, glosa_html=glosa_html, auto_soat_results=self._build_auto_soat_results( soat_resultados, clinical_snapshot.procedimientos_factura, ), qx_agent_results=self.coding_gateway.generate_qx_results( hallazgos=clinical_snapshot.hallazgos_qx, descripcion_procedimiento=clinical_snapshot.descripcion_qx, ), ) def execute( self, username: str, case_key: str, *, preserved_context: dict[str, Any] | None = None, ) -> dict[str, Any]: documents = self._load_documents(username, case_key) preserved_state = self._build_preserved_state(preserved_context) evidence_snapshot = self._build_evidence_snapshot(documents, case_key) rule_evaluation = ( self.rules_evaluator.execute(evidence_snapshot) if self.rules_evaluator is not None else _build_default_rule_evaluation() ) context = self._build_base_context( documents=documents, case_key=case_key, preserved_state=preserved_state, rule_evaluation=rule_evaluation, evidence_snapshot=evidence_snapshot, ) if not rule_evaluation.generation_allowed: raise EpicrisisGenerationBlockedError( self._resolve_blocking_reason(rule_evaluation) or "No se puede generar la epicrisis sin documento quirúrgico requerido.", evaluation=rule_evaluation, context=context, ) clinical_snapshot = self._build_clinical_snapshot(documents) coding_snapshot = self._build_coding_snapshot(documents, clinical_snapshot) integrated_summary = build_integrated_summary_context( username=username, historia=documents.historia, quirurgico=documents.quirurgico, metadatos_hc=clinical_snapshot.metadatos_hc, summary_composer=self.summary_composer, ) metadatos_hc = dict(clinical_snapshot.metadatos_hc) metadatos_hc["resumen"] = integrated_summary["texto"] context.update( { "metadatos_hc": metadatos_hc, "resumen_clinico_integrado": integrated_summary, "antecedentes_hc": clinical_snapshot.antecedentes_hc, "antecedentes_hc_estructurados": clinical_snapshot.antecedentes_hc_estructurados, "procedimientos_hc": clinical_snapshot.procedimientos_hc, "medicamentos_hc": clinical_snapshot.medicamentos_hc, "medicamentos_hc_display": clinical_snapshot.medicamentos_hc_display, "recomendaciones_medicas": clinical_snapshot.recomendaciones_medicas, "medicamentos_prescripcion": clinical_snapshot.medicamentos_prescripcion, "ayudas_diagnosticas": clinical_snapshot.ayudas_diagnosticas, "imagenes_diagnosticas": clinical_snapshot.imagenes_diagnosticas, "hallazgos_quirurgicos": clinical_snapshot.hallazgos_qx, "descripcion_procedimiento": clinical_snapshot.descripcion_qx, "soat_resultados": coding_snapshot.soat_resultados, "glosa_analisis": coding_snapshot.glosa_html, "procedimientos_factura": clinical_snapshot.procedimientos_factura, "auto_soat_results": coding_snapshot.auto_soat_results, "qx_agent_results": coding_snapshot.qx_agent_results, "codigos_desde_soat": merge_soat_results( merge_soat_results( coding_snapshot.auto_soat_results, coding_snapshot.qx_agent_results, ), preserved_state.manual_soat_results, ), } ) context = normalize_context_payload(context) context = self.coding_gateway.refresh_diagnostico_context(context) curation = reconcile_case_objective_data( [ documents.historia, documents.quirurgico, documents.factura, *documents.radiologia, *documents.laboratorio, *documents.generico, *documents.prescripcion, ], preserved_context=preserved_context, ) if curation is not None: context.update(curation_context_payload(curation)) apply_factura_cost_ordering(context) if self.medication_pertinence_reviewer is not None: reviews = self.medication_pertinence_reviewer.review( medications=context.get("medicamentos_caso") or [], diagnoses=context.get("diagnosticos_consolidados") or [], procedures=context.get("procedimientos_clinicos") or [], ) context["medicamentos_caso"] = apply_clinical_pertinence_reviews( context.get("medicamentos_caso") or [], reviews, ) context["medicamentos_glosa"] = build_medication_glosa_findings(context["medicamentos_caso"]) return context @dataclass class GetCaseEpicrisisContextUseCase: cache_repository: CaseEpicrisisCacheRepository build_case_context_use_case: BuildCaseEpicrisisContextUseCase def execute( self, username: str, case_key: str, *, regen: bool = False, ) -> dict[str, Any]: existing_doc = self.cache_repository.get(username, case_key) existing_context = existing_doc.get("contexto") if existing_doc else None if not regen and isinstance(existing_context, dict): return normalize_context_payload(existing_context) preserved_context = ( normalize_context_payload(existing_context) if isinstance(existing_context, dict) else None ) try: context = self.build_case_context_use_case.execute( username, case_key, preserved_context=preserved_context, ) except EpicrisisGenerationBlockedError as exc: self.cache_repository.upsert( username=username, case_key=case_key, context=exc.context, regen_requested=regen, ) raise self.cache_repository.upsert( username=username, case_key=case_key, context=context, regen_requested=regen, ) return context @dataclass class UpdateCaseEpicrisisDraftUseCase: cache_repository: CaseEpicrisisCacheRepository get_case_context_use_case: GetCaseEpicrisisContextUseCase def execute( self, username: str, case_key: str, *, payload: dict[str, Any], ) -> dict[str, Any]: context = normalize_context_payload( self.get_case_context_use_case.execute(username, case_key, regen=False) ) incoming_manual = payload.get("manual_soat_results") if incoming_manual is not None: context["manual_soat_results"] = merge_soat_results([], incoming_manual) current_draft = coerce_pdf_draft(context.get("pdf_draft")) draft_payload = payload.get("pdf_draft") if isinstance(payload.get("pdf_draft"), dict) else payload merged_draft = {**current_draft, **coerce_pdf_draft(draft_payload)} context["codigos_desde_soat"] = merge_soat_results( merge_soat_results(context.get("auto_soat_results"), context.get("qx_agent_results")), context.get("manual_soat_results"), ) context["pdf_draft"] = merged_draft refreshed_context = self.get_case_context_use_case.build_case_context_use_case.coding_gateway.refresh_diagnostico_context( context ) self.cache_repository.upsert( username=username, case_key=case_key, context=refreshed_context, regen_requested=False, ) return refreshed_context @dataclass class UpdatePdfPreparationMetadataUseCase: cache_repository: CaseEpicrisisCacheRepository get_case_context_use_case: GetCaseEpicrisisContextUseCase def execute( self, username: str, case_key: str, *, procedure_key: str, group: int | None, ) -> dict[str, Any]: context = normalize_context_payload( self.get_case_context_use_case.execute(username, case_key, regen=False) ) procedure = next( ( item for item in build_pdf_preparation_procedimientos(context) if str(item.get("key") or "") == procedure_key ), None, ) if procedure is None: raise LookupError("El procedimiento no es publicable en Preparar PDF.") if procedure.get("clasificacion") != "quirurgico": raise ValueError("El grupo manual solo aplica a procedimientos quirúrgicos publicables.") if group is not None and procedure.get("grupo_soat_origen") == "oficial": raise ValueError("El grupo SOAT oficial tiene precedencia y no puede reemplazarse.") metadata = coerce_pdf_preparation_user_metadata(context.get("pdf_preparation_user_metadata")) assignments = [ item for item in metadata["procedure_soat_groups"] if str(item.get("procedure_key") or "") != procedure_key ] if group is not None: assignments.append( PdfPreparationSoatGroup( procedure_key=procedure_key, group=group, actor=username, updated_at=datetime.now(UTC), ).model_dump(mode="json") ) context["pdf_preparation_user_metadata"] = {"procedure_soat_groups": assignments} refreshed_context = normalize_context_payload(context) self.cache_repository.upsert( username=username, case_key=case_key, context=refreshed_context, regen_requested=False, ) return refreshed_context