"""Reglas únicas para decidir publicación clínica y enrutamiento a auditoría.""" from __future__ import annotations import math from collections.abc import Mapping, Sequence from datetime import UTC, datetime from enum import StrEnum from typing import Any from pydantic import BaseModel, ConfigDict, Field PUBLICATION_POLICY_VERSION = "v1" PUBLICABLE_STATUSES = frozenset({"validado_automaticamente", "aprobado", "corregido"}) RAG_MIN_SCORE = 82.0 RAG_MIN_MARGIN = 5.0 PENDING_STATUSES = frozenset({"pendiente_revision", "pendiente_servicio_externo", "conflicto", "por_revisar"}) EXCLUDED_STATUSES = frozenset({"rechazado", "excluido", "glosado"}) AUDITOR_ACTIONS = ("aprobar", "corregir", "rechazar") _DIAGNOSTIC_AID_REVIEW_ALERTS = frozenset( { "asociacion_canonica_no_confirmada", "asociacion_canonica_pendiente_revision", "contradiccion", "contradictorio", "evidencia_contradictoria", "factura_sin_soporte_clinico_asociado", "referencia_qx_pendiente_revision", "referencia_textual_pendiente_revision", "texto_no_extraible", } ) class PublicationAction(StrEnum): APPROVE = "aprobar" CORRECT = "corregir" REJECT = "rechazar" class PublicationProjectionState(StrEnum): """Estado de un artefacto dentro de una vista de publicación concreta.""" EVIDENCE = "evidencia" PENDING_REVIEW = "pending_review" SELECTABLE = "selectable" SELECTED = "selected" PUBLISHED = "published" EXCLUDED = "excluded" GLOSADO = "glosado" class PublicationProjection(BaseModel): """Contrato común de identidad, evidencia y decisión de publicación. The model is deliberately independent of MongoDB and HTTP. Consumers may add presentation fields around it, but the identity and publication facts remain the same across audit, preparation, chronology and exports. """ model_config = ConfigDict(extra="forbid", frozen=True) artifact_id: str occurrence_id: str = "" occurrence_ids: list[str] = Field(default_factory=list) artifact_type: str source: str = "" documents: list[str] = Field(default_factory=list) source_status: str = "" projection_status: PublicationProjectionState reason_publication: str = "" reason_review: str = "" evidences: list[dict[str, Any]] = Field(default_factory=list) order_applied: str = "revision" class PublicationEligibility(BaseModel): """Resultado auditable de la política de publicación para un artefacto.""" model_config = ConfigDict(extra="forbid") artifact_type: str status: str normalized_status: str = "" source: str = "" origin: str = "" publishable: bool = False requires_audit: bool = True excluded: bool = False reason: str actions: list[str] = Field(default_factory=lambda: list(AUDITOR_ACTIONS)) policy_version: str = PUBLICATION_POLICY_VERSION evaluated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) def _text(value: Any) -> str: return str(value or "").strip() def _artifact_type(value: Any) -> str: return _text(value).casefold().replace("-", "_").replace(" ", "_") or "desconocido" def _diagnostic_aid_human_approved(item: Mapping[str, Any]) -> bool: return any( isinstance(decision, Mapping) and _text(decision.get("decision")).casefold() in {"approve", "aprobado", "corregir", "correct", "corrected"} for decision in item.get("decisions") or [] ) def _status_from_item(item: Mapping[str, Any], *, artifact_type: str) -> tuple[str, str]: """Read current and inherited contracts without treating absence as approval.""" if artifact_type == "ayuda_diagnostica": value = _text(item.get("estado_interpretacion")) if value in {"no_interpretado", "pendiente_revision"}: audited_status = _text(item.get("status")) if audited_status in PUBLICABLE_STATUSES and _diagnostic_aid_human_approved(item): return audited_status, "status" return "pendiente_revision", "estado_interpretacion" if value == "excluido": return "excluido", "estado_interpretacion" if value == "interpretado": for key in ("estado_publicacion", "publication_status", "status"): publication_status = _text(item.get(key)) if publication_status: return publication_status, key return "aprobado", "estado_interpretacion" if item.get("interpretado") is True and _text(item.get("concepto") or item.get("interpretacion")): # Legacy cached projections predate ``estado_interpretacion``; # retain their explicit positive signal during migration. return "aprobado", "interpretado_heredado" for key in ("estado_publicacion", "publication_status", "status", "estado_codificacion", "estado"): value = _text(item.get(key)) if value: return value, key pertinence = item.get("pertinencia") if artifact_type == "medicamento" and isinstance(pertinence, Mapping): value = _text(pertinence.get("estado")) if value == "pertinente": return "aprobado", "pertinencia.estado" if value == "posible_glosa": return "glosado", "pertinencia.estado" if value: return "pendiente_revision", "pertinencia.estado" # ``exportable`` was the explicit legacy approval flag. It is not inferred # from a missing field, so old cached documents remain readable safely. if item.get("exportable") is True: return "aprobado", "exportable_heredado" return "", "" def _is_external_artifact(artifact_type: str) -> bool: return artifact_type in {"soat", "glosa", "glosas", "resultado_soat", "hallazgo_glosa"} def _has_sufficient_evidence(evidences: Sequence[Any]) -> bool: for evidence in evidences: if isinstance(evidence, Mapping): traceable = any( _text(evidence.get(key)) for key in ("document_uid", "document_id", "source_hash", "filename") ) anchored = any( _text(evidence.get(key)) for key in ("page", "section", "excerpt", "original_term", "occurrence_id") ) if traceable and anchored: return True elif _text(evidence): return True return False def _diagnostic_aid_has_review_issue(item: Mapping[str, Any]) -> bool: if item.get("trayectoria_requiere_revision") is True: return True if item.get("incidencias") or item.get("canonical_incidents"): return True alerts = {_text(value).casefold() for value in item.get("alertas") or [] if _text(value)} return bool(alerts.intersection(_DIAGNOSTIC_AID_REVIEW_ALERTS)) def _diagnostic_aid_review_reason(item: Mapping[str, Any]) -> str: alerts = {_text(value).casefold() for value in item.get("alertas") or [] if _text(value)} roles = { _text(evidence.get("evidence_role") or evidence.get("rol_evidencia")).casefold() for evidence in item.get("evidencias") or [] if isinstance(evidence, Mapping) } if "registro_administrativo" in roles or "evidencia_administrativa" in alerts: return "evidencia_administrativa_no_es_resultado_clinico" if "orden" in roles and not roles.intersection({"resultado_narrativo", "realizado"}): return "orden_sin_informe" if "referencia_textual_pendiente_revision" in alerts or "resultado_narrativo" in roles: return "referencia_narrativa_sin_informe_estructurado" if _text(item.get("estado_interpretacion")).casefold() == "interpretado": return "" return "La ayuda diagnóstica requiere aprobación humana antes de publicación." def is_diagnostic_aid_presentable(item: Mapping[str, Any] | None) -> bool: """Keep a detected aid available for traceability while it awaits review. This is intentionally different from ``evaluate_publication_eligibility``: an aid may remain in the curation and audit views without being treated as approved clinical prose. """ if not isinstance(item, Mapping): return False if item.get("exportable") is False or item.get("glosado") is True: return False status = _text(item.get("estado_interpretacion")).casefold() if status in {"excluido", "rechazado", "glosado"}: return False return bool(_text(item.get("nombre") or item.get("texto") or item.get("descripcion"))) class PublicationPolicy: """Central policy used by all clinical preparation and export projections.""" def evaluate( self, *, artifact_type: str, status: str | None = None, source: str = "", origin: str = "", evidences: Sequence[Any] = (), conflicts: Sequence[Any] = (), incidents: Sequence[Any] = (), automatic_validation: Mapping[str, Any] | None = None, ) -> PublicationEligibility: normalized_type = _artifact_type(artifact_type) normalized_status = _text(status) automatic = automatic_validation if isinstance(automatic_validation, Mapping) else {} has_blocking_issue = bool(list(conflicts) or list(incidents)) if _is_external_artifact(normalized_type): return PublicationEligibility( artifact_type=normalized_type, status=normalized_status, normalized_status=normalized_status, source=_text(source), origin=_text(origin), publishable=False, requires_audit=True, excluded=True, reason="SOAT y glosas se conservan para auditoría, fuera de exportables clínicos.", actions=list(AUDITOR_ACTIONS), ) if normalized_status in EXCLUDED_STATUSES: return PublicationEligibility( artifact_type=normalized_type, status=normalized_status, normalized_status=normalized_status, source=_text(source), origin=_text(origin), publishable=False, requires_audit=False, excluded=True, reason=f"Artefacto {normalized_status}; se conserva su trazabilidad y no se exporta.", actions=[], ) if normalized_status in PUBLICABLE_STATUSES: if normalized_type == "ayuda_diagnostica" and has_blocking_issue: return PublicationEligibility( artifact_type=normalized_type, status=normalized_status, normalized_status=normalized_status, source=_text(source), origin=_text(origin), publishable=False, requires_audit=True, reason="La ayuda diagnóstica tiene incidencias o evidencia pendiente de revisión.", actions=list(AUDITOR_ACTIONS), ) if normalized_status == "validado_automaticamente" and has_blocking_issue: return PublicationEligibility( artifact_type=normalized_type, status=normalized_status, normalized_status=normalized_status, source=_text(source), origin=_text(origin), publishable=False, requires_audit=True, reason="La validación automática tiene conflictos o incidencias pendientes.", actions=list(AUDITOR_ACTIONS), ) return PublicationEligibility( artifact_type=normalized_type, status=normalized_status, normalized_status=normalized_status, source=_text(source), origin=_text(origin), publishable=True, requires_audit=False, reason=( "Aprobado automáticamente por RAG con coincidencia confiable." if normalized_status == "validado_automaticamente" and bool(automatic.get("aprobado_automaticamente")) else "Estado permitido para publicación clínica." ), actions=[], ) reason = ( "Estado pendiente de revisión humana." if normalized_status in PENDING_STATUSES else "Estado ausente o desconocido; se requiere revisión humana." ) return PublicationEligibility( artifact_type=normalized_type, status=normalized_status, normalized_status=normalized_status, source=_text(source), origin=_text(origin), publishable=False, requires_audit=True, excluded=False, reason=reason, actions=list(AUDITOR_ACTIONS), ) def evaluate_publication_eligibility( item: Mapping[str, Any] | None = None, *, artifact_type: str = "desconocido", status: str | None = None, source: str = "", origin: str = "", evidences: Sequence[Any] | None = None, conflicts: Sequence[Any] | None = None, incidents: Sequence[Any] | None = None, automatic_validation: Mapping[str, Any] | None = None, ) -> PublicationEligibility: """Evaluate an item, accepting both the domain arguments and legacy maps.""" raw = item if isinstance(item, Mapping) else {} resolved_type = _artifact_type(raw.get("artifact_type") or artifact_type) resolved_status = status if resolved_status is None or ( resolved_type == "ayuda_diagnostica" and _text(raw.get("estado_interpretacion")) in {"no_interpretado", "pendiente_revision", "excluido"} ): resolved_status, _ = _status_from_item(raw, artifact_type=resolved_type) human_approved = resolved_type == "ayuda_diagnostica" and _diagnostic_aid_human_approved(raw) resolved_incidents = ( [] if human_approved else list(incidents or raw.get("incidencias") or raw.get("canonical_incidents") or ()) ) if resolved_type == "ayuda_diagnostica" and not human_approved and _diagnostic_aid_has_review_issue(raw): resolved_incidents.append({"codigo": "ayuda_diagnostica_requiere_revision"}) eligibility = PublicationPolicy().evaluate( artifact_type=resolved_type, status=resolved_status, source=source or _text(raw.get("source") or raw.get("fuente")), origin=origin or _text(raw.get("origin") or raw.get("origen")), evidences=evidences or raw.get("evidencias") or raw.get("evidence") or (), conflicts=conflicts or raw.get("conflicts") or (), incidents=resolved_incidents, automatic_validation=automatic_validation or raw.get("automatic") or raw.get("validacion_automatica"), ) if resolved_type == "ayuda_diagnostica" and not eligibility.publishable: reason = _diagnostic_aid_review_reason(raw) if reason: eligibility = eligibility.model_copy(update={"reason": reason}) return eligibility def is_publishable(item: Mapping[str, Any] | None, *, artifact_type: str) -> bool: return evaluate_publication_eligibility(item, artifact_type=artifact_type).publishable def resolve_rag_automatic( *, artifact_type: str, candidate: Any = None, candidates: Sequence[Any] = (), score: float | None = None, margin: float | None = None, evidences: Sequence[Any] = (), catalog_available: bool = True, candidate_is_leaf: bool = True, conflicts: Sequence[Any] = (), description_conflict: bool = False, external_blocked: bool = False, resolver_available: bool = True, catalog: str = "", catalog_version: str = "", resolver_version: str = "v1", ) -> dict[str, Any] | None: """Return a strict RAG approval trace, or ``None`` when review is required. This is deliberately a decision gate rather than a best-effort suggestion: every condition must be met before an artifact can enter the publication allowlist. """ candidate_values = [value for value in candidates if value not in (None, "")] if not candidate_values and candidate not in (None, ""): candidate_values = [candidate] candidate_ids = { _text(value.get("id") or value.get("canonical_id") or value.get("code") or value) if isinstance(value, Mapping) else _text(value) for value in candidate_values } if ( not resolver_available or not catalog_available or not candidate_is_leaf or len(candidate_values) != 1 or len(candidate_ids) != 1 or score is None or not math.isfinite(float(score)) or float(score) < RAG_MIN_SCORE or margin is None or not math.isfinite(float(margin)) or float(margin) < RAG_MIN_MARGIN or not _has_sufficient_evidence(evidences) or bool(list(conflicts)) or description_conflict or external_blocked ): return None selected = candidate_values[0] selected_value = dict(selected) if isinstance(selected, Mapping) else _text(selected) decided_at = datetime.now(UTC).isoformat() return { "status": "validado_automaticamente", "origin": "rag", "aprobado_automaticamente": True, "method": "rag_rapidfuzz", "score": round(float(score), 4), "margin": round(float(margin), 4), "candidate": selected_value, "catalog": _text(catalog), "catalog_version": _text(catalog_version), "resolver_version": _text(resolver_version), "decided_at": decided_at, "evidencias": [dict(value) if isinstance(value, Mapping) else _text(value) for value in evidences], "reason": "Aprobado automáticamente por RAG con coincidencia confiable", } __all__ = [ "AUDITOR_ACTIONS", "EXCLUDED_STATUSES", "PENDING_STATUSES", "PUBLICABLE_STATUSES", "PUBLICATION_POLICY_VERSION", "RAG_MIN_MARGIN", "RAG_MIN_SCORE", "PublicationAction", "PublicationEligibility", "PublicationPolicy", "PublicationProjection", "PublicationProjectionState", "evaluate_publication_eligibility", "is_publishable", "resolve_rag_automatic", ]