from __future__ import annotations import re from typing import Any from app.case_epicrisis.domain.models import ( PdfPreparationSoatGroup, PdfPreparationUserMetadata, ProcedureClassification, ) from .common import ( _clean_unique_text_list, _normalize_ascii, _procedure_key, ) from .diagnostic_aids import ( coerce_ayudas_diagnosticas, merge_soat_results, ) def default_pdf_draft() -> dict[str, Any]: return { "excluded_diagnosticos": [], "excluded_diagnostico_keys": [], "manual_diagnosticos": [], "manual_diagnostico_entries": [], "excluded_procedimientos": [], "manual_procedimientos": [], "manual_procedure_entries": [], "procedure_classification_overrides": {}, "excluded_medicamentos": [], "excluded_medicamento_keys": [], "manual_medicamentos": [], "manual_medicamento_entries": [], "excluded_ayudas_diagnosticas_keys": [], "included_ayudas_diagnosticas_keys": [], "manual_ayudas_diagnosticas": [], "excluded_imagenes_diagnosticas": [], "manual_imagenes_diagnosticas": [], "hallazgos_extra": "", "descripcion_procedimiento_extra": "", "pdf_preparation_overrides": [], } def _normalize_pdf_preparation_overrides(values: Any) -> list[dict[str, Any]]: """Conserva correcciones publicables sin mezclar datos de los documentos fuente.""" overrides: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() for raw in values if isinstance(values, list) else []: if not isinstance(raw, dict): continue section = str(raw.get("section") or "").strip() item_key = str(raw.get("item_key") or "").strip() values_payload = raw.get("values") if not section or not item_key or not isinstance(values_payload, dict): continue identity = (section, item_key) if identity in seen: continue clean_values = { str(key).strip(): str(value or "").strip() for key, value in values_payload.items() if str(key).strip() } history = [] for entry in raw.get("history") if isinstance(raw.get("history"), list) else []: if not isinstance(entry, dict) or not str(entry.get("field") or "").strip(): continue history.append( { "field": str(entry.get("field") or "").strip(), "original_value": entry.get("original_value", ""), "new_value": entry.get("new_value", ""), "reason": str(entry.get("reason") or "").strip(), "actor": str(entry.get("actor") or "").strip(), "changed_at": str(entry.get("changed_at") or "").strip(), } ) seen.add(identity) overrides.append( { "section": section, "item_key": item_key, "values": clean_values, "history": history, } ) return overrides def _normalize_manual_soat_results(values: Any) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] for item in values or []: if not isinstance(item, dict): continue descripcion = str(item.get("descripcion") or "").strip() codigo_soat = str(item.get("codigo_soat") or "").strip() if not descripcion: continue results.append({**item, "descripcion": descripcion, "codigo_soat": codigo_soat}) return merge_soat_results([], results) def _normalize_manual_procedure_entries(values: Any, legacy_values: Any = None) -> list[dict[str, str]]: entries: list[dict[str, str]] = [] seen_keys: set[str] = set() for item in values if isinstance(values, list) else []: if not isinstance(item, dict): continue text = str(item.get("value") or item.get("texto") or item.get("descripcion") or "").strip() key = str(item.get("key") or "").strip() if not text or not key or key in seen_keys: continue seen_keys.add(key) entries.append({"key": key, "value": text}) for text in _clean_unique_text_list(legacy_values): if any(entry["value"].casefold() == text.casefold() for entry in entries): continue key = f"manual:{_procedure_key(text)}" if key in seen_keys: continue seen_keys.add(key) entries.append({"key": key, "value": text}) return entries def _normalize_manual_text_entries(values: Any, legacy_values: Any, *, prefix: str) -> list[dict[str, Any]]: entries: list[dict[str, Any]] = [] seen_keys: set[str] = set() for item in values if isinstance(values, list) else []: if not isinstance(item, dict): continue value = str(item.get("value") or item.get("texto") or "").strip() key = str(item.get("key") or "").strip() if not value or not key or key in seen_keys: continue seen_keys.add(key) entries.append({**item, "key": key, "value": value}) for value in _clean_unique_text_list(legacy_values): if any(str(entry["value"]).casefold() == value.casefold() for entry in entries): continue normalized = re.sub(r"[^a-z0-9]+", "-", _normalize_ascii(value).casefold()).strip("-") key = f"manual:{prefix}:{normalized or len(entries) + 1}" if key in seen_keys: continue seen_keys.add(key) entries.append({"key": key, "value": value}) return entries def coerce_pdf_preparation_user_metadata(values: Any) -> dict[str, Any]: if not isinstance(values, dict): return PdfPreparationUserMetadata().model_dump(mode="json") try: return PdfPreparationUserMetadata.model_validate(values).model_dump(mode="json") except ValueError: valid = [] for item in values.get("procedure_soat_groups") or []: try: valid.append(PdfPreparationSoatGroup.model_validate(item)) except (TypeError, ValueError): continue return PdfPreparationUserMetadata(procedure_soat_groups=valid).model_dump(mode="json") def _normalize_procedure_classification_overrides(values: Any) -> dict[str, str]: if not isinstance(values, dict): return {} valid = { ProcedureClassification.QUIRURGICO.value, ProcedureClassification.NO_QUIRURGICO.value, } return { str(key).strip(): str(value).strip() for key, value in values.items() if str(key).strip() and str(value).strip() in valid } def coerce_pdf_draft(values: Any) -> dict[str, Any]: draft = default_pdf_draft() if isinstance(values, dict): manual_procedimientos = _clean_unique_text_list(values.get("manual_procedimientos")) manual_procedure_entries = _normalize_manual_procedure_entries( values.get("manual_procedure_entries"), manual_procedimientos ) if manual_procedure_entries: manual_procedimientos = [item["value"] for item in manual_procedure_entries] manual_diagnostico_entries = _normalize_manual_text_entries( values.get("manual_diagnostico_entries"), values.get("manual_diagnosticos"), prefix="diagnostico" ) manual_medicamento_entries = _normalize_manual_text_entries( values.get("manual_medicamento_entries"), values.get("manual_medicamentos"), prefix="medicamento" ) draft.update( { "excluded_diagnosticos": _clean_unique_text_list(values.get("excluded_diagnosticos")), "excluded_diagnostico_keys": _clean_unique_text_list(values.get("excluded_diagnostico_keys")), "manual_diagnosticos": [item["value"] for item in manual_diagnostico_entries], "manual_diagnostico_entries": manual_diagnostico_entries, "excluded_procedimientos": _clean_unique_text_list(values.get("excluded_procedimientos")), "manual_procedimientos": manual_procedimientos, "manual_procedure_entries": manual_procedure_entries, "procedure_classification_overrides": ( _normalize_procedure_classification_overrides( values.get("procedure_classification_overrides") ) ), "excluded_medicamentos": _clean_unique_text_list(values.get("excluded_medicamentos")), "excluded_medicamento_keys": _clean_unique_text_list(values.get("excluded_medicamento_keys")), "manual_medicamentos": [item["value"] for item in manual_medicamento_entries], "manual_medicamento_entries": manual_medicamento_entries, "excluded_ayudas_diagnosticas_keys": _clean_unique_text_list( values.get("excluded_ayudas_diagnosticas_keys") ), "included_ayudas_diagnosticas_keys": _clean_unique_text_list( values.get("included_ayudas_diagnosticas_keys") ), "manual_ayudas_diagnosticas": coerce_ayudas_diagnosticas( values.get("manual_ayudas_diagnosticas") ), "excluded_imagenes_diagnosticas": _clean_unique_text_list( values.get("excluded_imagenes_diagnosticas") ), "manual_imagenes_diagnosticas": _clean_unique_text_list( values.get("manual_imagenes_diagnosticas") ), "hallazgos_extra": str(values.get("hallazgos_extra") or "").strip(), "descripcion_procedimiento_extra": str( values.get("descripcion_procedimiento_extra") or "" ).strip(), "pdf_preparation_overrides": _normalize_pdf_preparation_overrides( values.get("pdf_preparation_overrides") ), } ) if not draft["manual_ayudas_diagnosticas"] and draft["manual_imagenes_diagnosticas"]: draft["manual_ayudas_diagnosticas"] = coerce_ayudas_diagnosticas( draft["manual_imagenes_diagnosticas"] ) return draft