from __future__ import annotations import re from typing import Any, cast from app.case_epicrisis.application.audit_traceability import project_auditoria_traceability from app.case_epicrisis.application.cost_ordering import ( attach_and_sort_costs, extract_invoice_cost_lines, normalize_money, sort_publishable_procedures, ) from app.case_epicrisis.application.publication import ( build_publication_projections, project_publication_items, ) from app.case_epicrisis.domain.chronology import ( project_chronological_item, project_effective_date, sort_chronological_items, ) from app.case_epicrisis.domain.models import ( IncapacidadClinica, ProcedimientoClinico, ProcedureClassification, ) from app.case_epicrisis.domain.publication import ( evaluate_publication_eligibility, is_diagnostic_aid_presentable, ) from app.services.clinical_processing import ( formatear_medicamento_canonico, normalizar_lista_medicamentos, ) from app.services.historia_medicamentos import reconcile_case_medications from .cie10 import ( _coerce_diagnostico_catalog, _is_real_cie10_code, _is_technical_description, ) from .common import ( _clean_unique_text_list, _normalize_whitespace, _procedure_key, ) from .diagnostic_aids import ( build_ayudas_diagnosticas, coerce_ayudas_diagnosticas, format_ayuda_diagnostica_presentacion, project_ayuda_diagnostica, ) from .factura import ( _servicios_factura, extraer_imagenes_diagnosticas, ) from .pdf_draft import ( _normalize_manual_procedure_entries, _normalize_manual_text_entries, coerce_pdf_draft, coerce_pdf_preparation_user_metadata, ) _PDF_PRIMARY_DIAGNOSTICO_SOURCES = {"historia_clinica", "quirurgico", "manual_pdf"} _PDF_PRIMARY_AYUDA_SOURCES = {"radiologia", "laboratorio", "factura", "manual_pdf"} _PDF_PREPARATION_SECTIONS = { "diagnosticos", "procedimientos", "medicamentos", "ayudas_diagnosticas", } def build_summary_clinical_projection(context: dict[str, Any]) -> dict[str, Any]: """Proyección única del resumen para pantalla, PDF y Excel.""" raw_integrated = context.get("resumen_clinico_integrado") integrated: dict[str, Any] = ( cast(dict[str, Any], raw_integrated) if isinstance(raw_integrated, dict) else {} ) raw_metadatos = context.get("metadatos_hc") metadatos: dict[str, Any] = cast(dict[str, Any], raw_metadatos) if isinstance(raw_metadatos, dict) else {} has_explicit_summary = bool(integrated.get("texto") or integrated.get("parrafos")) paragraphs = [str(item).strip() for item in integrated.get("parrafos") or [] if str(item or "").strip()] text = str(integrated.get("texto") or "").strip() or "\n\n".join(paragraphs) text = text or str(metadatos.get("resumen") or "").strip() sources = integrated.get("fuentes") if isinstance(integrated.get("fuentes"), list) else [] return { "estado": str(integrated.get("estado") or "solo_historia").strip(), "texto": text, "parrafos": paragraphs or ([text] if text else []), "causa_fallback": str(integrated.get("causa_fallback") or "").strip() or None, "razon_fallback": str(integrated.get("razon_fallback") or "").strip() or None, "advertencia": str(integrated.get("advertencia") or "").strip() or None, "detalle_fallback": str(integrated.get("detalle_fallback") or "").strip() or None, "regenerable": bool(integrated.get("regenerable")), "fuentes": sort_chronological_items(sources) if has_explicit_summary else [], } def _chronology_fields( raw: Any, *, source: str = "", clinical_date: Any = None, document_date: Any = None, source_position: int | None = None, ) -> dict[str, Any]: candidate = dict(raw) if isinstance(raw, dict) else {} if source: candidate.setdefault("document_type", source) if source_position is not None: candidate.setdefault("source_position", source_position) direct_projection = project_effective_date(candidate) projected = ( direct_projection if direct_projection.is_dated else project_effective_date( candidate, clinical_date=clinical_date, document_date=document_date, ) ) fields = { "effective_date": projected.effective_date, "date_source": projected.date_source.value, "date_precision": projected.date_precision.value, "date_warning": projected.warning, } if source: fields["document_type"] = source if source_position is not None: fields["source_position"] = source_position return fields def _pdf_preparation_override_map(context: dict[str, Any], section: str) -> dict[str, dict[str, Any]]: draft = coerce_pdf_draft(context.get("pdf_draft")) return { str(item.get("item_key")): item for item in draft.get("pdf_preparation_overrides") or [] if isinstance(item, dict) and item.get("section") == section and str(item.get("item_key") or "").strip() } def _apply_diagnostico_override(item: dict[str, Any]) -> None: code = str(item.get("codigo") or item.get("codigo_cie10") or "").strip().upper() description = str( item.get("descripcion_clinica_normalizada") or item.get("descripcion") or item.get("diagnostico") or "" ).strip() concept = str( item.get("concepto_cie10_oficial") or item.get("concepto_diagnostico_presentacion") or item.get("diagnostico") or description ).strip() suffix = "" if item.get("es_pop"): procedures = item.get("procedimientos_pop_asociados") or [item.get("procedimiento_pop_asociado")] suffix = " — POSTOPERATORIO" + (f" {procedures[0]}" if procedures and procedures[0] else "") item["codigo"] = code item["descripcion"] = description item["concepto_diagnostico_presentacion"] = concept item["descripcion_clinica_normalizada"] = description item["texto_presentacion"] = f"{code} - {concept}" if code and concept else code or concept item["texto_presentacion"] += suffix def _apply_medicamento_override(item: dict[str, Any], values: dict[str, Any]) -> None: if "nombre" in values: item["medicamento"] = str(values["nombre"] or "").strip() if "medicamento" in values: item["nombre"] = str(values["medicamento"] or "").strip() item["estados"] = ( [str(values.get("estado") or item.get("estados", [""])[0]).strip()] if ("estado" in values or item.get("estados")) else [] ) def _apply_ayuda_diagnostica_override(item: dict[str, Any], values: dict[str, Any]) -> None: for field in ("tipo", "nombre", "concepto"): if field in values: item[field] = str(values[field] or "").strip() if "estado" in values: item["estado_interpretacion"] = str(values["estado"] or "").strip() if "estado_interpretacion" in values: item["estado_interpretacion"] = str(values["estado_interpretacion"] or "").strip() if "estado_interpretacion" in values or "estado" in values: item["interpretado"] = item["estado_interpretacion"] == "interpretado" item["exportable"] = item["estado_interpretacion"] != "excluido" if item.get("estado_interpretacion") != "interpretado": item["interpretado"] = False item["concepto"] = "No interpretado" def _apply_section_override(item: dict[str, Any], values: dict[str, Any], section: str) -> None: if section == "diagnosticos": _apply_diagnostico_override(item) elif section == "medicamentos": _apply_medicamento_override(item, values) elif section == "ayudas_diagnosticas": _apply_ayuda_diagnostica_override(item, values) def apply_pdf_preparation_overrides( items: list[dict[str, Any]], context: dict[str, Any], section: str, ) -> list[dict[str, Any]]: if section not in _PDF_PREPARATION_SECTIONS: return items overrides = _pdf_preparation_override_map(context, section) if not overrides: return items result: list[dict[str, Any]] = [] for raw in items or []: if not isinstance(raw, dict): continue item = dict(raw) key = str(item.get("key") or "").strip() override = overrides.get(key) if not override: result.append(item) continue values = ( cast(dict[str, Any], override.get("values")) if isinstance(override.get("values"), dict) else {} ) item.update(values) item["_pdf_preparation_overridden"] = True _apply_section_override(item, values, section) result.append(item) return result def _diagnostico_display(item: Any) -> str: if not isinstance(item, dict): return str(item or "").strip() codigo = str(item.get("codigo") or item.get("codigo_asignado") or "").strip().upper() descripcion = str(item.get("descripcion") or "").strip() diagnostico = str(item.get("diagnostico") or "").strip() detalle = descripcion or diagnostico if codigo and detalle: return f"{codigo} - {detalle}" return codigo or detalle def _procedimiento_display(item: Any) -> str: if not isinstance(item, dict): return str(item or "").strip() codigo = str(item.get("codigo") or item.get("codigo_cups") or "").strip() descripcion = str(item.get("descripcion") or item.get("procedimiento") or "").strip() if codigo and descripcion: return f"{codigo} - {descripcion}" return codigo or descripcion def _add_text_value(values: set[str], text: Any) -> None: normalized = str(text or "").strip() if normalized: values.add(normalized) def _extend_with_display_values( values: set[str], items: Any, formatter, ) -> None: for item in items or []: _add_text_value(values, formatter(item)) def _iter_soat_diagnosticos(items: Any) -> list[Any]: diagnosticos: list[Any] = [] for item in items or []: if not isinstance(item, dict): continue principal = item.get("cie10_principal") if isinstance(principal, dict): diagnosticos.append(principal) diagnosticos.extend(item.get("cie10_secundarios") or []) return diagnosticos def _iter_soat_procedimientos(items: Any) -> list[Any]: procedimientos: list[Any] = [] for item in items or []: if not isinstance(item, dict): continue principal = item.get("cups_principal") if isinstance(principal, dict): procedimientos.append(principal) procedimientos.extend(item.get("cups_alternativos") or []) return procedimientos def _iter_factura_procedimientos(items: Any) -> list[dict[str, str]]: procedimientos: list[dict[str, str]] = [] for item in items or []: if not isinstance(item, dict): continue procedimientos.append( { "codigo": item.get("codigo_cups") or item.get("codigo_referencia"), "descripcion": item.get("descripcion"), } ) return procedimientos def _primary_diagnostico_sources(item: dict[str, Any]) -> set[str]: return {str(source).strip() for source in (item.get("fuentes") or []) if str(source).strip()} def _has_meaningful_diagnostico_text(item: dict[str, Any]) -> bool: diagnostico = _normalize_whitespace( item.get("descripcion_clinica_normalizada") or item.get("diagnostico") ) descripcion = _normalize_whitespace(item.get("descripcion") or item.get("diagnostico")) return any( clinical_text and not _is_technical_description(clinical_text) for clinical_text in (diagnostico, descripcion) ) def _is_pdf_primary_diagnostico(item: dict[str, Any], context: dict[str, Any] | None = None) -> bool: fuentes = _primary_diagnostico_sources(item) if fuentes: has_clinical_source = bool(fuentes & _PDF_PRIMARY_DIAGNOSTICO_SOURCES) else: tipo_contexto = str(item.get("tipo_contexto") or "").strip().lower() has_clinical_source = tipo_contexto not in {"no_clasificado", "procedimiento_soat"} if not has_clinical_source or not _has_meaningful_diagnostico_text(item): return False if context and context.get("curation_schema_version"): return evaluate_publication_eligibility(item, artifact_type="diagnostico").publishable estado = str(item.get("estado_codificacion") or "").strip() if estado == "codificado": return _is_real_cie10_code(item.get("codigo")) return estado in {"pendiente_validacion", "provisional_grupo", "inferido_revisable"} def build_pdf_primary_diagnosticos( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[dict[str, Any]]: items = [ dict(item) for item in _coerce_diagnostico_catalog(context.get("diagnosticos_consolidados")) if _is_pdf_primary_diagnostico(item, context) ] for item in items: item["auditoria"] = project_auditoria_traceability( item, section="diagnosticos", preparation_overrides=context.get("pdf_draft"), ) projected = apply_pdf_preparation_overrides(items, context, "diagnosticos") if apply_overrides else items return project_publication_items( projected, artifact_type="diagnostico", context=context, order_applied="revision", ) def _classify_cups(code: Any) -> ProcedureClassification: # La clasificación automática solo se acepta cuando ya fue resuelta por el catálogo CUPS. # Un código aislado no contiene el número de capítulo en sus primeros dígitos. return ProcedureClassification.SIN_CLASIFICAR def build_procedimientos_clinicos(context: dict[str, Any]) -> list[dict[str, Any]]: """Consolida fuentes clínicas/administrativas sin perder trazabilidad interna.""" if context.get("curation_schema_version"): resolved_states = {"realizado_y_facturado", "no_quirurgico_facturado"} curated = [] for source_position, raw in enumerate(context.get("procedimientos_curados") or []): if ( not isinstance(raw, dict) or not evaluate_publication_eligibility(raw, artifact_type="procedimiento").publishable ): continue classification = str(raw.get("classification") or "sin_clasificar") if classification not in {"quirurgico", "no_quirurgico"}: continue if raw.get("reconciliation_status") not in resolved_states and raw.get("status") not in { "aprobado", "corregido", }: continue item = { "key": raw.get("item_id") or _procedure_key(raw.get("description"), raw.get("codigo_cups")), "codigo_cups": str(raw.get("codigo_cups") or ""), "descripcion": str(raw.get("description") or ""), "clasificacion": classification, "clasificacion_automatica": classification, "clasificacion_manual": False, "soporte_clinico": "realizado" in (raw.get("states") or []), "fuentes": [ evidence.get("document_type") for evidence in raw.get("evidence") or [] if isinstance(evidence, dict) and evidence.get("document_type") ], "estado_documental": list(raw.get("states") or []), "evidencias": [ evidence for evidence in raw.get("evidence") or [] if isinstance(evidence, dict) ], "advertencias": list(raw.get("warnings") or []), "estado_conciliacion": raw.get("reconciliation_status"), "razon_conciliacion": str(raw.get("reconciliation_reason") or ""), "auditoria": project_auditoria_traceability( raw, section="procedimientos", preparation_overrides=context.get("pdf_draft"), ), } item.update(_chronology_fields(raw, source="curated", source_position=source_position)) curated.append(item) ordered = sort_publishable_procedures( curated, lines=extract_invoice_cost_lines(context.get("factura")), ) result = [ { key: value for key, value in item.items() if key not in {"codigo_soat", "grupo_quirurgico_soat", "costo_facturado"} } | ({"soat_group_aplica": True} if item.get("grupo_quirurgico_soat") is not None else {}) for item in ordered if isinstance(item, dict) ] return project_publication_items( result, artifact_type="procedimiento", context=context, order_applied="revision", ) merged: dict[str, ProcedimientoClinico] = {} chronology_by_key: dict[str, dict[str, Any]] = {} draft = coerce_pdf_draft(context.get("pdf_draft")) overrides = draft.get("procedure_classification_overrides") or {} def add( raw: Any, source: str, *, clinical: bool, surgical_evidence: bool = False, official_classification: str = "", clinical_date: Any = None, document_date: Any = None, source_position: int | None = None, ) -> None: if isinstance(raw, dict): description = str( raw.get("descripcion") or raw.get("procedimiento") or raw.get("concepto") or "" ).strip() cups = str(raw.get("codigo_cups") or raw.get("codigo") or "").strip() soat = str(raw.get("codigo_soat") or "").strip() else: description = str(raw or "").strip() match = re.match(r"^\s*(\d{6})\s*(?:[-:]|\s)\s*(.+)$", description) cups = match.group(1) if match else "" description = match.group(2).strip() if match else description soat = "" if not description: return base_key = _procedure_key(description, cups) key = base_key date_fields = _chronology_fields( raw, source=source, clinical_date=clinical_date, document_date=document_date, source_position=source_position, ) existing = merged.get(key) classification = ( ProcedureClassification(official_classification) if official_classification in {"quirurgico", "no_quirurgico"} else _classify_cups(cups) ) if surgical_evidence and classification == ProcedureClassification.SIN_CLASIFICAR: classification = ProcedureClassification.QUIRURGICO if existing is None: existing = ProcedimientoClinico( descripcion=description, codigo_cups=cups, codigo_soat=soat, clasificacion=classification, fuentes=[source], evidencia=description if clinical else "", soporte_clinico=clinical, ) else: if source == "factura": existing.descripcion = description existing.codigo_cups = existing.codigo_cups or cups existing.codigo_soat = existing.codigo_soat or soat existing.fuentes = list(dict.fromkeys([*existing.fuentes, source])) existing.soporte_clinico = existing.soporte_clinico or clinical if existing.clasificacion == ProcedureClassification.SIN_CLASIFICAR: existing.clasificacion = classification override = overrides.get(key) or overrides.get(base_key) if override: existing.clasificacion = ProcedureClassification(override) existing.clasificacion_manual = True merged[key] = existing if date_fields: current_date = chronology_by_key.get(key) chronology_by_key[key] = ( sort_chronological_items([current_date, date_fields])[0] if current_date else date_fields ) historia_date = project_effective_date(context.get("historia")).effective_date for source_position, item in enumerate(context.get("procedimientos_hc") or []): add( item, "historia_clinica", clinical=True, clinical_date=historia_date, source_position=source_position, ) quirurgico = context.get("quirurgico") or {} quirurgico_date = project_effective_date(quirurgico).effective_date for source_position, item in enumerate(quirurgico.get("codigos_cups") or []): add( item, "quirurgico", clinical=True, surgical_evidence=True, clinical_date=quirurgico_date, source_position=source_position, ) invoice_services = _servicios_factura(context.get("factura") or {}) factura_date = project_effective_date(context.get("factura")).effective_date if invoice_services: for source_position, item in enumerate(invoice_services.get("procedimientos_quirurgicos") or []): add( item, "factura", clinical=False, official_classification="quirurgico", clinical_date=item.get("fecha_servicio") if isinstance(item, dict) else None, document_date=factura_date, source_position=source_position, ) for source_position, item in enumerate(invoice_services.get("procedimientos_no_quirurgicos") or []): add( item, "factura", clinical=False, official_classification="no_quirurgico", clinical_date=item.get("fecha_servicio") if isinstance(item, dict) else None, document_date=factura_date, source_position=source_position, ) else: for source_position, item in enumerate(context.get("procedimientos_factura") or []): add( item, "factura", clinical=False, clinical_date=item.get("fecha_servicio") if isinstance(item, dict) else None, document_date=factura_date, source_position=source_position, ) for source_position, result in enumerate(context.get("codigos_desde_soat") or []): if not isinstance(result, dict): continue source = "qx_agent" if result in (context.get("qx_agent_results") or []) else "soat" surgical = source == "qx_agent" principal = result.get("cups_principal") if isinstance(principal, dict): add( {**principal, "codigo_soat": result.get("codigo_soat")}, source, clinical=surgical, surgical_evidence=surgical, clinical_date=quirurgico_date if surgical else None, source_position=source_position, ) for alternative in result.get("cups_alternativos") or []: add( alternative, source, clinical=surgical, surgical_evidence=surgical, clinical_date=quirurgico_date if surgical else None, source_position=source_position, ) procedures = [] for key, item in merged.items(): automatic_classification = item.clasificacion if automatic_classification == ProcedureClassification.SIN_CLASIFICAR and { "quirurgico", "qx_agent", } & set(item.fuentes): automatic_classification = ProcedureClassification.QUIRURGICO procedures.append( item.model_dump(mode="json") | { "key": key, "clasificacion_automatica": automatic_classification.value, } | chronology_by_key.get(key, {}) ) ordered = attach_and_sort_costs( procedures, category="procedimiento", lines=extract_invoice_cost_lines(context.get("factura")), ) result = [ { key: value for key, value in item.items() if key not in {"codigo_soat", "grupo_quirurgico_soat", "costo_facturado"} } | ({"soat_group_aplica": True} if item.get("grupo_quirurgico_soat") is not None else {}) if isinstance(item, dict) else item for item in ordered ] return project_publication_items( result, artifact_type="procedimiento", context=context, order_applied="revision", ) def build_pdf_primary_procedimientos(context: dict[str, Any], *, apply_overrides: bool = True) -> list[str]: """Contrato legado: la vista nueva consume ``procedimientos_clinicos``.""" procedimientos: list[str] = [] items = build_procedimientos_clinicos(context) if apply_overrides: items = apply_pdf_preparation_overrides(items, context, "procedimientos") for item in items: if not isinstance(item, dict): continue if not context.get("curation_schema_version") and not ( {"historia_clinica", "quirurgico"} & set(item.get("fuentes") or []) ): continue description = str(item.get("descripcion") or "").strip() code = str(item.get("codigo_cups") or "").strip() display = " - ".join(part for part in (code, description) if part) if display: procedimientos.append(display) return procedimientos def _parse_procedure_display(value: Any) -> tuple[str, str]: text = str(value or "").strip() match = re.match(r"^\s*(\d{6})\s*(?:[-:]|\s)\s*(.+)$", text) if match: return match.group(1), match.group(2).strip() return "", text def build_pdf_selectable_procedimientos( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[dict[str, Any]]: """Enriquece la selección legado sin cambiar sus fuentes ni su orden visible.""" clinical = build_procedimientos_clinicos(context) if apply_overrides: clinical = apply_pdf_preparation_overrides(clinical, context, "procedimientos") by_display = { " - ".join( part for part in ( str(item.get("codigo_cups") or "").strip(), str(item.get("descripcion") or "").strip(), ) if part ): item for item in clinical if isinstance(item, dict) } selectable: list[dict[str, Any]] = [] for display in build_pdf_primary_procedimientos(context, apply_overrides=apply_overrides): item = by_display.get(display) if item is not None: selectable.append(dict(item)) continue code, description = _parse_procedure_display(display) automatic = _classify_cups(code).value selectable.append( { "key": _procedure_key(description, code), "codigo_cups": code, "descripcion": description, "clasificacion": automatic, "clasificacion_automatica": automatic, "clasificacion_manual": False, "soporte_clinico": True, "fuentes": [], } ) draft = coerce_pdf_draft(context.get("pdf_draft")) overrides = draft.get("procedure_classification_overrides") or {} for entry in draft.get("manual_procedure_entries") or []: if context.get("curation_schema_version"): continue key = str(entry.get("key") or "").strip() text = str(entry.get("value") or "").strip() if not key or not text: continue code, description = _parse_procedure_display(text) classification = str(overrides.get(key) or ProcedureClassification.SIN_CLASIFICAR.value) selectable.append( { "key": key, "codigo_cups": code, "descripcion": description, "texto_manual": text, "clasificacion": classification, "clasificacion_automatica": ProcedureClassification.SIN_CLASIFICAR.value, "clasificacion_manual": classification in { ProcedureClassification.QUIRURGICO.value, ProcedureClassification.NO_QUIRURGICO.value, }, "soporte_clinico": True, "fuentes": ["manual_pdf"], "manual": True, } ) return selectable def build_pdf_preparation_procedimientos(context: dict[str, Any]) -> list[dict[str, Any]]: """Proyección exclusiva de UI; costos y fórmulas nunca cruzan este borde.""" selectable = [ item for item in build_pdf_selectable_procedimientos(context) if isinstance(item, dict) and item.get("clasificacion") in {"quirurgico", "no_quirurgico"} ] enriched = attach_and_sort_costs( selectable, category="procedimiento", lines=extract_invoice_cost_lines(context.get("factura")), ) metadata = coerce_pdf_preparation_user_metadata(context.get("pdf_preparation_user_metadata")) manual_groups = { str(item.get("procedure_key") or ""): int(item["group"]) for item in metadata["procedure_soat_groups"] } curated_groups = { str(item.get("item_id") or ""): item.get("soat_group") for item in context.get("procedimientos_curados") or [] if isinstance(item, dict) and item.get("soat_group") is not None } decorated: list[tuple[dict[str, Any], int | None, Any, int, str]] = [] for canonical_order, raw in enumerate(enriched): if not isinstance(raw, dict): continue item = project_chronological_item(raw) key = str(item.get("key") or "").strip() classification = str(item.get("clasificacion") or "") raw_official_group = item.get("grupo_quirurgico_soat") if raw_official_group is None: raw_official_group = curated_groups.get(key) official_group: int | None = None if raw_official_group is not None: try: parsed_official_group = int(raw_official_group) except (TypeError, ValueError): pass else: official_group = parsed_official_group if 2 <= parsed_official_group <= 23 else None manual_group = manual_groups.get(key) if classification == "quirurgico" else None effective_group = official_group if official_group is not None else manual_group origin = "oficial" if official_group is not None else "manual_visual" if manual_group else "" cost = item.get("costo_facturado") cost_value = normalize_money(cost.get("valor_total")) if isinstance(cost, dict) else None decorated.append((item, effective_group, cost_value, canonical_order, origin)) if any(entry[1] is not None for entry in decorated): decorated.sort( key=lambda entry: ( entry[1] is None, -(entry[1] or 0), entry[2] is None, -(entry[2] or 0), entry[3], ) ) elif any(entry[0].get("effective_date") for entry in decorated): chronology_rank = { str(item.get("key") or ""): position for position, item in enumerate(sort_chronological_items([entry[0] for entry in decorated])) } decorated.sort(key=lambda entry: chronology_rank.get(str(entry[0].get("key") or ""), entry[3])) else: decorated.sort(key=lambda entry: entry[3]) result = [] for visual_order, (item, group, _cost, canonical_order, origin) in enumerate(decorated): projection = { key: value for key, value in item.items() if key not in {"codigo_soat", "grupo_quirurgico_soat", "costo_facturado", "valoracion_soat"} } projection.update( { "grupo_soat_preparacion": group, "grupo_soat_origen": origin, "orden_visual": visual_order, "orden_canonico": canonical_order, "auditoria": project_auditoria_traceability( item, section="procedimientos", preparation_overrides=context.get("pdf_draft"), ), } ) if origin == "manual_visual": projection["grupo_soat_advertencia"] = ( "Asignación manual para organización visual; sin efecto tarifario" ) result.append(projection) return project_publication_items( result, artifact_type="procedimiento", context=context, order_applied="soat" if any(item.get("grupo_soat_preparacion") is not None for item in result) else "temporal", ) def build_selected_pdf_procedimientos(context: dict[str, Any]) -> list[dict[str, Any]]: """Aplica la selección persistida y descarta categorías no publicables en servidor.""" draft = coerce_pdf_draft(context.get("pdf_draft")) excluded = { str(value).strip() for value in draft.get("excluded_procedimientos") or [] if str(value).strip() } selected = [] for item in build_pdf_selectable_procedimientos(context): classification = str(item.get("clasificacion") or "") if classification not in {"quirurgico", "no_quirurgico"}: continue display = " - ".join( value for value in ( str(item.get("codigo_cups") or "").strip(), str(item.get("descripcion") or "").strip(), ) if value ) if display in excluded or str(item.get("descripcion") or "").strip() in excluded: continue chronology = project_chronological_item(item) projection = dict(item) projection.update( { "codigo_cups": str(item.get("codigo_cups") or ""), "descripcion": str(item.get("descripcion") or ""), "clasificacion": classification, "effective_date": chronology.get("effective_date"), "date_source": chronology.get("date_source", "unknown"), "date_precision": chronology.get("date_precision", "unknown"), "date_warning": chronology.get("date_warning"), } ) if item.get("soat_group_aplica"): projection.update( { key: item.get(key) for key in ( "soat_group_aplica", "document_type", "source_position", ) if item.get(key) is not None } ) selected.append(projection) return selected def _sort_clinical_procedure_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]: """SOAT domina cuando aplica; de lo contrario se usa la cronología clínica.""" source = [dict(item) for item in items if isinstance(item, dict)] if any(item.get("soat_group_aplica") is True for item in source): return source return sort_chronological_items(source) def build_pdf_primary_procedimientos_cronologicos( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[str]: """Salida clínica antigua→reciente; no reemplaza el orden SOAT de preparación.""" items = build_procedimientos_clinicos(context) if apply_overrides: items = apply_pdf_preparation_overrides(items, context, "procedimientos") ordered = _sort_clinical_procedure_items( [ item for item in items if isinstance(item, dict) and ( context.get("curation_schema_version") or {"historia_clinica", "quirurgico"} & set(item.get("fuentes") or []) ) ] ) result = [] for item in ordered: display = " - ".join( part for part in ( str(item.get("codigo_cups") or "").strip(), str(item.get("descripcion") or "").strip(), ) if part ) if display: result.append(display) return result def build_pdf_selectable_procedimientos_cronologicos( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[dict[str, Any]]: """Elementos seleccionables ordenados para publicación clínica.""" return project_publication_items( _sort_clinical_procedure_items( build_pdf_selectable_procedimientos(context, apply_overrides=apply_overrides) ), artifact_type="procedimiento", context=context, order_applied="temporal", ) def build_selected_pdf_procedimientos_cronologicos(context: dict[str, Any]) -> list[dict[str, Any]]: """Selección persistida en orden clínico, conservando el orden SOAT de preparación.""" return project_publication_items( _sort_clinical_procedure_items(build_selected_pdf_procedimientos(context)), artifact_type="procedimiento", context=context, order_applied="temporal", ) def extract_incapacidad_clinica(context: dict[str, Any]) -> dict[str, Any]: historia = context.get("historia") or {} structured = historia.get("analysis_structured") or historia.get("historia_json") or {} candidate = structured.get("incapacidad") if isinstance(structured, dict) else None if isinstance(candidate, dict): ordered = candidate.get("ordenada") is True or str(candidate.get("ordenada") or "").casefold() in { "si", "sí", "true", } return IncapacidadClinica( ordenada=ordered, fecha_inicio=str(candidate.get("fecha_inicio") or ""), fecha_fin=str(candidate.get("fecha_fin") or ""), duracion=str(candidate.get("duracion") or candidate.get("dias") or ""), unidad=str(candidate.get("unidad") or ("días" if candidate.get("dias") else "")), motivo=str(candidate.get("motivo") or ""), recomendacion=str(candidate.get("recomendacion") or ""), evidencia=str(candidate.get("evidencia") or ""), ).model_dump(mode="json") raw = _normalize_whitespace(historia.get("texto_extraido") or historia.get("texto") or "") match = re.search( r"(?i)\b(?:se\s+)?(?:ordena|prescribe|otorga)\s+incapacidad(?:\s+(?:m[eé]dica\s+)?)?" r"(?:por\s+)?(?P\d+)\s*(?Pd[ií]as?|horas?)\b[^.\n]{0,160}", raw, ) if not match: return IncapacidadClinica().model_dump(mode="json") return IncapacidadClinica( ordenada=True, duracion=match.group("duration"), unidad=match.group("unit"), evidencia=match.group(0), ).model_dump(mode="json") def _is_meaningful_pdf_medicamento(item: dict[str, Any]) -> bool: nombre = str(item.get("medicamento") or "").strip() dosis = str(item.get("dosis") or "").strip() cantidad = str(item.get("cantidad") or "").strip() texto_original = str(item.get("texto_original") or "").strip() if nombre.lower() == "no especificado" and not dosis and not cantidad: return False if nombre and nombre.lower() != "no especificado": return True return bool(dosis or cantidad or (texto_original and texto_original.lower() != "no especificado")) def build_pdf_primary_medicamentos( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[dict[str, Any]]: catalog = context.get("medicamentos_caso") if isinstance(catalog, list): items = [ dict(item) for item in catalog if isinstance(item, dict) and "administrado" in {str(status or "").strip().casefold() for status in (item.get("estados") or [])} and _is_meaningful_pdf_medicamento(item) and ( not context.get("curation_schema_version") or evaluate_publication_eligibility(item, artifact_type="medicamento").publishable ) ] projected = ( apply_pdf_preparation_overrides(items, context, "medicamentos") if apply_overrides else items ) return project_publication_items( projected, artifact_type="medicamento", context=context, order_applied="revision", ) factura = context.get("factura") or {} servicios_factura = _servicios_factura(factura) medications = reconcile_case_medications( normalizar_lista_medicamentos(context.get("medicamentos_hc"), fuente="historia_clinica"), servicios_factura.get("medicamentos") or [], ) items = [ item for item in medications if "administrado" in {str(status or "").strip().casefold() for status in (item.get("estados") or [])} and _is_meaningful_pdf_medicamento(item) and ( not context.get("curation_schema_version") or evaluate_publication_eligibility(item, artifact_type="medicamento").publishable ) ] projected = apply_pdf_preparation_overrides(items, context, "medicamentos") if apply_overrides else items return project_publication_items( projected, artifact_type="medicamento", context=context, order_applied="revision", ) def _is_primary_pdf_ayuda(ayuda: dict[str, Any]) -> bool: if ayuda.get("exportable") is False or ayuda.get("glosado"): return False # Detection is not approval: pending/ordered aids stay in Auditoría until # a human curation decision changes their publication status. if not evaluate_publication_eligibility(ayuda, artifact_type="ayuda_diagnostica").publishable: return False fuente = str(ayuda.get("fuente") or "").strip() return fuente in _PDF_PRIMARY_AYUDA_SOURCES or bool(ayuda.get("nombre")) def build_pdf_primary_ayudas_diagnosticas( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[dict[str, Any]]: ayudas = coerce_ayudas_diagnosticas(context.get("ayudas_diagnosticas")) or build_ayudas_diagnosticas( context ) items = [ project_ayuda_diagnostica(ayuda) for ayuda in ayudas if _is_primary_pdf_ayuda(ayuda) and is_diagnostic_aid_presentable(ayuda) ] projected = ( apply_pdf_preparation_overrides(items, context, "ayudas_diagnosticas") if apply_overrides else items ) return project_publication_items( projected, artifact_type="ayuda_diagnostica", context=context, order_applied="revision", ) def build_pdf_primary_ayudas_diagnosticas_cronologicos( context: dict[str, Any], *, apply_overrides: bool = True ) -> list[dict[str, Any]]: """Ayudas diagnósticas antiguas→recientes sin alterar el orden financiero.""" return project_publication_items( sort_chronological_items( build_pdf_primary_ayudas_diagnosticas(context, apply_overrides=apply_overrides) ), artifact_type="ayuda_diagnostica", context=context, order_applied="temporal", ) def build_clinical_chronology_snapshot(context: dict[str, Any]) -> dict[str, Any]: """Construye una única proyección ordenada para los adaptadores clínicos.""" integrated_summary = context.get("resumen_clinico_integrado") raw_summary_sources = ( integrated_summary.get("fuentes") if isinstance(integrated_summary, dict) and isinstance(integrated_summary.get("fuentes"), list) else [] ) summary_projection = build_summary_clinical_projection(context) procedures = context.get("procedimientos_clinicos") if not isinstance(procedures, list): procedures = build_procedimientos_clinicos(context) aids = context.get("ayudas_diagnosticas") if not isinstance(aids, list): aids = build_ayudas_diagnosticas(context) ordered_procedures = project_publication_items( _sort_clinical_procedure_items(procedures), artifact_type="procedimiento", context=context, order_applied="temporal", ) ordered_aids = project_publication_items( sort_chronological_items(aids), artifact_type="ayuda_diagnostica", context=context, order_applied="temporal", ) return { "summary_sources": sort_chronological_items(raw_summary_sources), "summary_projection": summary_projection, "procedures": ordered_procedures, "selectable_procedures": build_pdf_selectable_procedimientos_cronologicos(context), "selected_procedures": build_selected_pdf_procedimientos_cronologicos(context), "aids": ordered_aids, "primary_aids": build_pdf_primary_ayudas_diagnosticas_cronologicos(context), } def build_pdf_preparation_source_items(context: dict[str, Any], section: str) -> list[dict[str, Any]]: builders = { "diagnosticos": lambda: build_pdf_primary_diagnosticos(context, apply_overrides=False), "procedimientos": lambda: build_pdf_selectable_procedimientos(context, apply_overrides=False), "medicamentos": lambda: build_pdf_primary_medicamentos(context, apply_overrides=False), "ayudas_diagnosticas": lambda: build_pdf_primary_ayudas_diagnosticas(context, apply_overrides=False), } return builders[section]() if section in builders else [] def build_pdf_export_blocks(context: dict[str, Any]) -> dict[str, Any]: candidate_diagnoses = context.get("pdf_candidate_diagnosticos") if not isinstance(candidate_diagnoses, list): candidate_diagnoses = [] chronology = build_clinical_chronology_snapshot(context) procedimientos_clinicos = build_procedimientos_clinicos(context) return { "pdf_primary_diagnosticos": build_pdf_primary_diagnosticos(context), # Candidate data is retained for review/backward compatibility only; # no exporter consumes it as a publication source. "pdf_candidate_diagnosticos": [dict(item) for item in candidate_diagnoses if isinstance(item, dict)], "pdf_primary_procedimientos": build_pdf_primary_procedimientos(context), "pdf_selectable_procedimientos": build_pdf_selectable_procedimientos(context), "pdf_primary_medicamentos": build_pdf_primary_medicamentos(context), "pdf_primary_ayudas_diagnosticas": build_pdf_primary_ayudas_diagnosticas(context), "procedimientos_clinicos": procedimientos_clinicos, "procedimientos_clinicos_cronologicos": chronology["procedures"], "pdf_primary_procedimientos_cronologicos": build_pdf_primary_procedimientos_cronologicos(context), "pdf_selectable_procedimientos_cronologicos": chronology["selectable_procedures"], "ayudas_diagnosticas_cronologicas": chronology["aids"], "pdf_primary_ayudas_diagnosticas_cronologicos": chronology["primary_aids"], "procedimientos_seleccionados_cronologicos": chronology["selected_procedures"], "resumen_fuentes_cronologicas": chronology["summary_sources"], "resumen_clinico_proyeccion": chronology["summary_projection"], "incapacidad_clinica": extract_incapacidad_clinica(context), "publication_projections": build_publication_projections(context), } def build_pdf_chronological_export_blocks(context: dict[str, Any]) -> dict[str, Any]: """Recalcula solo las proyecciones cronológicas después de una curación o revisión.""" chronology = build_clinical_chronology_snapshot(context) return { "procedimientos_clinicos_cronologicos": chronology["procedures"], "pdf_primary_procedimientos_cronologicos": build_pdf_primary_procedimientos_cronologicos(context), "pdf_selectable_procedimientos_cronologicos": chronology["selectable_procedures"], "ayudas_diagnosticas_cronologicas": chronology["aids"], "pdf_primary_ayudas_diagnosticas_cronologicos": chronology["primary_aids"], "procedimientos_seleccionados_cronologicos": chronology["selected_procedures"], "resumen_fuentes_cronologicas": chronology["summary_sources"], "resumen_clinico_proyeccion": chronology["summary_projection"], } def _build_diagnosticos_base(context: dict[str, Any]) -> tuple[set[str], set[str]]: diagnosticos_catalog = _coerce_diagnostico_catalog(context.get("diagnosticos_consolidados")) if diagnosticos_catalog: return ( { item.get("texto_presentacion", "") for item in diagnosticos_catalog if item.get("texto_presentacion") }, {item.get("key", "") for item in diagnosticos_catalog if item.get("key")}, ) historia = context.get("historia") or {} quirurgico = context.get("quirurgico") or {} codigos_desde_soat = context.get("codigos_desde_soat") diagnosticos: set[str] = set() _extend_with_display_values( diagnosticos, (historia.get("codigos_cie10") or []) + (quirurgico.get("codigos_cie10") or []), _diagnostico_display, ) _extend_with_display_values( diagnosticos, _iter_soat_diagnosticos(codigos_desde_soat), _diagnostico_display, ) return diagnosticos, set() def _build_procedimientos_base( quirurgico: dict[str, Any], procedimientos_hc: Any, procedimientos_factura: Any, codigos_desde_soat: Any, ) -> set[str]: procedimientos: set[str] = set() _extend_with_display_values(procedimientos, procedimientos_hc, lambda item: str(item or "").strip()) _extend_with_display_values(procedimientos, quirurgico.get("codigos_cups"), _procedimiento_display) _extend_with_display_values( procedimientos, _iter_factura_procedimientos(procedimientos_factura), _procedimiento_display, ) _extend_with_display_values( procedimientos, _iter_soat_procedimientos(codigos_desde_soat), _procedimiento_display, ) return procedimientos def _build_medicamentos_base(context: dict[str, Any], factura: dict[str, Any]) -> tuple[set[str], set[str]]: medicamentos: set[str] = set() keys: set[str] = set() catalog = context.get("medicamentos_caso") if isinstance(catalog, list): for item in catalog: if not isinstance(item, dict): continue display = formatear_medicamento_canonico(item) if display: medicamentos.add(display) if item.get("key"): keys.add(str(item["key"])) return medicamentos, keys servicios_factura = _servicios_factura(factura) medicamentos_hc = normalizar_lista_medicamentos( context.get("medicamentos_hc"), fuente="historia_clinica", ) medicamentos_factura = normalizar_lista_medicamentos( servicios_factura.get("medicamentos"), fuente="factura", ) _extend_with_display_values(medicamentos, medicamentos_hc, formatear_medicamento_canonico) _extend_with_display_values(medicamentos, medicamentos_factura, formatear_medicamento_canonico) return medicamentos, keys def _build_imagenes_diagnosticas_base(context: dict[str, Any], factura: dict[str, Any]) -> set[str]: imagenes = set(_clean_unique_text_list(context.get("imagenes_diagnosticas"))) if imagenes: return imagenes return set(extraer_imagenes_diagnosticas(factura, context.get("radiologia"))) def _build_ayudas_diagnosticas_base(context: dict[str, Any]) -> tuple[set[str], set[str]]: ayudas = coerce_ayudas_diagnosticas(context.get("ayudas_diagnosticas")) or build_ayudas_diagnosticas( context ) return ( {format_ayuda_diagnostica_presentacion(item) for item in ayudas if item.get("nombre")}, {item.get("key", "") for item in ayudas if item.get("key")}, ) def build_selection_base(context: dict[str, Any]) -> dict[str, set[str]]: quirurgico = context.get("quirurgico") or {} factura = context.get("factura") or {} codigos_desde_soat = context.get("codigos_desde_soat") diagnosticos_textos, diagnosticos_keys = _build_diagnosticos_base(context) ayudas_textos, ayudas_keys = _build_ayudas_diagnosticas_base(context) medicamentos_textos, medicamento_keys = _build_medicamentos_base(context, factura) return { "diagnosticos": diagnosticos_textos, "diagnostico_keys": diagnosticos_keys, "procedimientos": _build_procedimientos_base( quirurgico, context.get("procedimientos_hc"), context.get("procedimientos_factura"), codigos_desde_soat, ), "medicamentos": medicamentos_textos, "medicamento_keys": medicamento_keys, "ayudas_diagnosticas": ayudas_textos, "ayudas_diagnosticas_keys": ayudas_keys, "imagenes_diagnosticas": _build_imagenes_diagnosticas_base(context, factura), } def reconcile_pdf_draft(pdf_draft: Any, context: dict[str, Any]) -> dict[str, Any]: draft = coerce_pdf_draft(pdf_draft) base = build_selection_base(context) diagnosticos_catalog = { item.get("key", ""): item.get("texto_presentacion", "") for item in _coerce_diagnostico_catalog(context.get("diagnosticos_consolidados")) } legacy_keys = [ key for key, text in diagnosticos_catalog.items() if text in draft["excluded_diagnosticos"] ] draft["excluded_diagnostico_keys"] = [ item for item in _clean_unique_text_list(draft["excluded_diagnostico_keys"] + legacy_keys) if item in base["diagnostico_keys"] ] draft["excluded_diagnosticos"] = [ diagnosticos_catalog[key] for key in draft["excluded_diagnostico_keys"] if key in diagnosticos_catalog ] draft["excluded_diagnosticos"] = [ item for item in draft["excluded_diagnosticos"] if item in base["diagnosticos"] ] selectable_procedure_texts = set(base["procedimientos"]) | set(draft["manual_procedimientos"]) draft["excluded_procedimientos"] = [ item for item in draft["excluded_procedimientos"] if item in selectable_procedure_texts ] draft["excluded_medicamentos"] = [ item for item in draft["excluded_medicamentos"] if item in base["medicamentos"] ] medication_catalog = { str(item.get("key") or ""): formatear_medicamento_canonico(item) for item in (context.get("medicamentos_caso") or []) if isinstance(item, dict) and item.get("key") } legacy_medication_keys = [ key for key, text in medication_catalog.items() if text in draft["excluded_medicamentos"] ] draft["excluded_medicamento_keys"] = [ item for item in _clean_unique_text_list(draft["excluded_medicamento_keys"] + legacy_medication_keys) if item in base["medicamento_keys"] ] draft["excluded_medicamentos"] = [ medication_catalog[key] for key in draft["excluded_medicamento_keys"] if key in medication_catalog ] legacy_ayuda_keys = [ item.get("key", "") for item in coerce_ayudas_diagnosticas(context.get("ayudas_diagnosticas")) if format_ayuda_diagnostica_presentacion(item) in draft["excluded_imagenes_diagnosticas"] ] draft["excluded_ayudas_diagnosticas_keys"] = [ item for item in _clean_unique_text_list(draft["excluded_ayudas_diagnosticas_keys"] + legacy_ayuda_keys) if item in base["ayudas_diagnosticas_keys"] ] draft["included_ayudas_diagnosticas_keys"] = [ item for item in _clean_unique_text_list(draft["included_ayudas_diagnosticas_keys"]) if item in base["ayudas_diagnosticas_keys"] and item not in draft["excluded_ayudas_diagnosticas_keys"] ] draft["excluded_imagenes_diagnosticas"] = [ item for item in draft["excluded_imagenes_diagnosticas"] if item in base["imagenes_diagnosticas"] ] draft["manual_diagnosticos"] = _clean_unique_text_list(draft["manual_diagnosticos"]) draft["manual_diagnostico_entries"] = _normalize_manual_text_entries( draft["manual_diagnostico_entries"], draft["manual_diagnosticos"], prefix="diagnostico" ) draft["manual_diagnosticos"] = [item["value"] for item in draft["manual_diagnostico_entries"]] draft["manual_procedimientos"] = _clean_unique_text_list(draft["manual_procedimientos"]) draft["manual_procedure_entries"] = _normalize_manual_procedure_entries( draft["manual_procedure_entries"], draft["manual_procedimientos"] ) draft["manual_procedimientos"] = [item["value"] for item in draft["manual_procedure_entries"]] draft["manual_medicamentos"] = _clean_unique_text_list(draft["manual_medicamentos"]) draft["manual_medicamento_entries"] = _normalize_manual_text_entries( draft["manual_medicamento_entries"], draft["manual_medicamentos"], prefix="medicamento" ) draft["manual_medicamentos"] = [item["value"] for item in draft["manual_medicamento_entries"]] draft["manual_imagenes_diagnosticas"] = _clean_unique_text_list(draft["manual_imagenes_diagnosticas"]) draft["manual_ayudas_diagnosticas"] = coerce_ayudas_diagnosticas(draft["manual_ayudas_diagnosticas"]) return draft