Dom/t2a_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e77c10da7dfab1dd33511a2a39ea8eded820f470
t2a_v2/src/config.py
dom 5d5f119057 feat: quality_tier CPAM (A/B/C) + requires_review + warnings catégorisés 
- ControleCPAM enrichi : quality_tier, requires_review, quality_warnings
- _assess_quality_tier() : classification basée sur score adversarial + warnings
  - Tier C (requires_review) : score <4, code hors périmètre, >2 preuves non traçables
  - Tier B : score 4-6, warnings mineurs
  - Tier A : score >=7, 0 critique
- _format_response() : bandeau "REVUE MANUELLE REQUISE" pour tier C,
  sections CRITIQUES/MINEURS séparées
- Badge qualité dans le viewer CPAM (vert A / orange B / rouge C)
- 17 tests : tier A/B/C, bandeau, séparation warnings, backward compat

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-20 11:01:21 +01:00

764 lines
26 KiB
Python
Raw Blame History

"""Configuration globale et modèles de données pour le pipeline T2A."""
from __future__ import annotations
import os
import contextvars
from functools import lru_cache
from pathlib import Path
from typing import Optional, Any, Dict
import yaml
from dotenv import load_dotenv
from pydantic import BaseModel, Field, field_validator
load_dotenv()
# --- Chemins ---
BASE_DIR = Path(__file__).resolve().parent.parent
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
ANONYMIZED_DIR = OUTPUT_DIR / "anonymized"
STRUCTURED_DIR = OUTPUT_DIR / "structured"
REPORTS_DIR = OUTPUT_DIR / "reports"
CONFIG_DIR = BASE_DIR / "config"
REFERENCE_RANGES_PATH = CONFIG_DIR / "reference_ranges.yaml"
BIO_RULES_PATH = CONFIG_DIR / "bio_rules.yaml"
LAB_SANITY_PATH = CONFIG_DIR / "lab_value_sanity.yaml"
RULES_DIR = CONFIG_DIR / "rules"
RULES_BASE_PATH = RULES_DIR / "base.yaml"
RULES_ENABLED_PATH = RULES_DIR / "enabled.yaml"
RULES_ROUTER_PATH = RULES_DIR / "router.yaml"
for d in (INPUT_DIR, ANONYMIZED_DIR, STRUCTURED_DIR, REPORTS_DIR, CONFIG_DIR, RULES_DIR):
d.mkdir(parents=True, exist_ok=True)
# --- Configuration anonymisation ---
KEEP_ESTABLISHMENT_NAME = os.environ.get("T2A_KEEP_ESTABLISHMENT", "True").lower() in ("true", "1", "yes")
NER_MODEL = os.environ.get("T2A_NER_MODEL", "Jean-Baptiste/camembert-ner")
NER_CONFIDENCE_THRESHOLD = float(os.environ.get("T2A_NER_THRESHOLD", "0.80"))
# --- Configuration Ollama ---
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434")
OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "gemma3:27b-cloud")
OLLAMA_TIMEOUT = int(os.environ.get("OLLAMA_TIMEOUT", "120"))
OLLAMA_CACHE_PATH = BASE_DIR / "data" / "ollama_cache.json"
OLLAMA_MAX_PARALLEL = int(os.environ.get("OLLAMA_MAX_PARALLEL", "2"))
# --- Modèles par rôle LLM ---
OLLAMA_MODELS: dict[str, str] = {
"coding": os.environ.get("T2A_MODEL_CODING", "gemma3:27b-cloud"),
"cpam": os.environ.get("T2A_MODEL_CPAM", "gemma3:27b-cloud"),
"validation": os.environ.get("T2A_MODEL_VALIDATION", "deepseek-v3.2:cloud"),
"qc": os.environ.get("T2A_MODEL_QC", "gemma3:12b"),
}
def get_model(role: str) -> str:
"""Retourne le modèle associé à un rôle LLM, ou le modèle global par défaut."""
return OLLAMA_MODELS.get(role, OLLAMA_MODEL)
# --- Configuration RUM / établissement ---
FINESS = os.environ.get("T2A_FINESS", "000000000")
NUM_UM = os.environ.get("T2A_NUM_UM", "0000")
# --- Configuration RAG ---
RAG_INDEX_DIR = BASE_DIR / "data" / "rag_index"
REFERENTIELS_DIR = BASE_DIR / "data" / "referentiels"
UPLOAD_MAX_SIZE_MB = 50
ALLOWED_EXTENSIONS = {".pdf", ".csv", ".xlsx", ".xls", ".txt"}
CIM10_DICT_PATH = BASE_DIR / "data" / "cim10_dict.json"
CIM10_SUPPLEMENTS_PATH = BASE_DIR / "data" / "cim10_supplements.json"
CMA_LEVELS_PATH = BASE_DIR / "data" / "cma_levels.json"
CCAM_DICT_PATH = BASE_DIR / "data" / "ccam_dict.json"
CIM10_PDF = Path(os.environ.get("T2A_CIM10_PDF", "/home/dom/ai/aivanov_CIM/cim-10-fr_2026_a_usage_pmsi_version_provisoire_111225.pdf"))
GUIDE_METHODO_PDF = Path(os.environ.get("T2A_GUIDE_METHODO_PDF", "/home/dom/ai/aivanov_CIM/guide_methodo_mco_2026_version_provisoire.pdf"))
CCAM_PDF = Path(os.environ.get("T2A_CCAM_PDF", "/home/dom/ai/aivanov_CIM/actualisation_ccam_descriptive_a_usage_pmsi_v4_2025.pdf"))
# --- Modèle d'embedding ---
EMBEDDING_MODEL = os.environ.get("T2A_EMBEDDING_MODEL", "dangvantuan/sentence-camembert-large")
# --- Modèle de re-ranking (cross-encoder, CPU uniquement) ---
RERANKER_MODEL = os.environ.get("T2A_RERANKER_MODEL", "cross-encoder/ms-marco-MiniLM-L-6-v2")
# --- Références biologiques (fallback) ---
@lru_cache(maxsize=1)
def load_reference_ranges() -> Dict[str, Any]:
"""Charge les intervalles de référence biologiques depuis config/reference_ranges.yaml.
Hiérarchie d'usage recommandée dans les règles :
1) Normes présentes dans le document (ex: [N: 135-145])
2) Table YAML (par bande d'âge)
3) "Safe zones" conservatrices si âge inconnu
Le YAML est volontairement éditable par des non-informaticiens (future UI).
"""
# Defaults minimalistes (adultes) si YAML absent
defaults: Dict[str, Any] = {
"version": 1,
"age_bands": {"adult_min_years": 18},
"fallback_ranges": {
"adult": {
"platelets": {"low": 150, "high": 450, "unit": "G/L"},
"sodium": {"low": 135, "high": 145, "unit": "mmol/L"},
"potassium": {"low": 3.5, "high": 5.0, "unit": "mmol/L"},
},
# Valeurs pédiatriques: à affiner (par bandes d'âge) si besoin.
# Pour les règles "ruled_out" on utilise plutôt les safe_zones_unknown_age
"child": {
"platelets": {"low": 150, "high": 450, "unit": "G/L"},
"sodium": {"low": 135, "high": 145, "unit": "mmol/L"},
"potassium": {"low": 3.5, "high": 5.0, "unit": "mmol/L"},
},
},
"safe_zones_unknown_age": {
"platelets_ruled_out_low": 170,
"sodium_ruled_out_low": 138,
"potassium_ruled_out_high": 4.9,
"potassium_ruled_out_low": 3.7,
},
}
path = REFERENCE_RANGES_PATH
if not path.exists():
return defaults
try:
import yaml # type: ignore
except Exception:
# PyYAML absent: on garde les valeurs par défaut
return defaults
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
return defaults
# Merge léger: defaults comme socle, YAML surcharge
merged = dict(defaults)
for k, v in data.items():
merged[k] = v
return merged
except Exception:
return defaults
# --- Règles biologiques (pilotées par YAML) ---
@lru_cache(maxsize=1)
def load_bio_rules() -> Dict[str, Any]:
"""Charge les règles biologiques depuis config/bio_rules.yaml.
Objectif: permettre d'activer/désactiver et de paramétrer les règles
de type "contradiction bio ⇒ ruled_out" sans modifier le code.
Le fichier est volontairement simple (future UI).
"""
defaults: Dict[str, Any] = {
"version": 1,
"rules": {
"hyponatremia": {"enabled": True, "codes": ["E87.1"], "analyte": "sodium"},
"hyperkalemia": {"enabled": True, "codes": ["E87.5"], "analyte": "potassium"},
"hypokalemia": {"enabled": True, "codes": ["E87.6"], "analyte": "potassium"},
},
}
path = BIO_RULES_PATH
if not path.exists():
return defaults
try:
import yaml # type: ignore
except Exception:
return defaults
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
return defaults
merged = dict(defaults)
for k, v in data.items():
merged[k] = v
return merged
except Exception:
return defaults
# --- Garde-fous de parsing des valeurs biologiques (anti-OCR) ---
@lru_cache(maxsize=1)
def load_lab_value_sanity() -> Dict[str, Any]:
"""Charge des garde-fous de parsing depuis config/lab_value_sanity.yaml.
But:
- éviter que des artefacts de lecture PDF/OCR (ex: "8" au lieu de "4.8")
déclenchent de faux diagnostics (hyperK, etc.)
- garder une trace *auditable* (valeurs suspectes / écartées)
Ce fichier est volontairement éditable (future UI).
"""
defaults: Dict[str, Any] = {
"version": 1,
"policy": {
# Si True: les valeurs hors bornes plausibles sont écartées du dossier.
# Sinon: elles sont gardées avec quality="discarded".
"drop_out_of_range": True,
# Si True: on conserve les valeurs suspectes (quality="suspect") pour audit,
# mais les règles qualité privilégient les valeurs "ok" quand elles existent.
"keep_suspect": True,
},
# Clés normalisées (minuscules, sans accents) : potassium, sodium, plaquettes...
"tests": {
"potassium": {
# Bornes très larges (mmol/L) : sert uniquement à écarter l'impossible.
"hard_min": 0.5,
"hard_max": 9.0,
# Heuristique anti-OCR : un chiffre seul >=6 est souvent une décimale perdue (4,8 -> 8)
"suspect": {"single_digit_over": 6.0},
},
"sodium": {"hard_min": 90.0, "hard_max": 200.0},
"plaquettes": {"hard_min": 5.0, "hard_max": 2000.0},
"hemoglobine": {"hard_min": 3.0, "hard_max": 25.0},
"creatinine": {"hard_min": 1.0, "hard_max": 5000.0},
"crp": {"hard_min": 0.0, "hard_max": 1000.0},
"alat": {"hard_min": 0.0, "hard_max": 5000.0},
"asat": {"hard_min": 0.0, "hard_max": 5000.0},
"ggt": {"hard_min": 0.0, "hard_max": 5000.0},
"pal": {"hard_min": 0.0, "hard_max": 5000.0},
"bilirubine totale": {"hard_min": 0.0, "hard_max": 2000.0},
},
}
path = LAB_SANITY_PATH
if not path.exists():
return defaults
try:
import yaml # type: ignore
except Exception:
return defaults
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
return defaults
merged = dict(defaults)
for k, v in data.items():
merged[k] = v
return merged
except Exception:
return defaults
# --- Catalogue de règles (vetos + décisions), piloté par YAML ---
def _flatten_rules_yaml(data: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
"""Transforme un YAML de règles en dict {rule_id: cfg}.
Formats supportés :
- {packs: {pack_name: {enabled: bool, rules: {RULE_ID: {...}}}}}
- {rules: {RULE_ID: {...}}} (overlay simple)
"""
out: Dict[str, Dict[str, Any]] = {}
# Overlay simple
rules_block = data.get("rules")
if isinstance(rules_block, dict):
for rid, cfg in rules_block.items():
if not isinstance(cfg, dict):
cfg = {}
out[str(rid)] = dict(cfg)
packs = data.get("packs")
if isinstance(packs, dict):
for pack_name, pack_cfg in packs.items():
if not isinstance(pack_cfg, dict):
continue
pack_enabled = bool(pack_cfg.get("enabled", True))
rules = pack_cfg.get("rules")
if not isinstance(rules, dict):
continue
for rid, cfg in rules.items():
if not isinstance(cfg, dict):
cfg = {}
merged = dict(cfg)
merged.setdefault("pack", str(pack_name))
# La désactivation du pack désactive ses règles
merged["enabled"] = bool(merged.get("enabled", True)) and pack_enabled
out[str(rid)] = merged
return out
def _merge_rule_catalog(base: Dict[str, Dict[str, Any]], overlay: Dict[str, Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
"""Merge overlay → base (par règle)."""
merged = {k: dict(v) for k, v in base.items()}
for rid, cfg in overlay.items():
if rid not in merged:
merged[rid] = dict(cfg)
else:
# override champ par champ
for k, v in cfg.items():
merged[rid][k] = v
return merged
@lru_cache(maxsize=1)
def load_rules_catalog() -> Dict[str, Dict[str, Any]]:
"""Charge le catalogue de règles depuis config/rules/*.yaml.
- base.yaml : socle partagé (vetos + décisions)
- enabled.yaml : sélection d'overlays (site/spécialité)
- specialties/<name>.yaml et sites/<name>.yaml : overrides ciblés
Politique : si une règle n'est pas listée, elle est considérée "enabled".
(=> ne casse pas le comportement historique)
"""
try:
import yaml # type: ignore
except Exception:
return {}
catalog: Dict[str, Dict[str, Any]] = {}
# 1) base
if RULES_BASE_PATH.exists():
try:
base_data = yaml.safe_load(RULES_BASE_PATH.read_text(encoding="utf-8")) or {}
if isinstance(base_data, dict):
catalog = _flatten_rules_yaml(base_data)
except Exception:
catalog = {}
# 2) enabled overlays
active_site = ""
active_specialty = ""
extra_files: list[str] = []
if RULES_ENABLED_PATH.exists():
try:
enabled_data = yaml.safe_load(RULES_ENABLED_PATH.read_text(encoding="utf-8")) or {}
if isinstance(enabled_data, dict):
active = enabled_data.get("active") or {}
if isinstance(active, dict):
active_site = str(active.get("site") or "").strip()
active_specialty = str(active.get("specialty") or "").strip()
extra = active.get("extra")
if isinstance(extra, list):
extra_files = [str(x) for x in extra if str(x).strip()]
except Exception:
pass
else:
# fallback env
active_site = os.environ.get("T2A_SITE", "").strip()
active_specialty = os.environ.get("T2A_SPECIALTY", "").strip()
# 3) specialty overlay
if active_specialty:
p = RULES_DIR / "specialties" / f"{active_specialty}.yaml"
if p.exists():
try:
data = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
if isinstance(data, dict):
catalog = _merge_rule_catalog(catalog, _flatten_rules_yaml(data))
except Exception:
pass
# 4) site overlay
if active_site:
p = RULES_DIR / "sites" / f"{active_site}.yaml"
if p.exists():
try:
data = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
if isinstance(data, dict):
catalog = _merge_rule_catalog(catalog, _flatten_rules_yaml(data))
except Exception:
pass
# 5) extra overlays
for rel in extra_files:
p = RULES_DIR / rel
if p.exists():
try:
data = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
if isinstance(data, dict):
catalog = _merge_rule_catalog(catalog, _flatten_rules_yaml(data))
except Exception:
pass
return catalog
# --- Routage dynamique des règles (packs) ---
# Contexte runtime, défini *par dossier* (contextvars => safe pour batch / multi-thread)
_RULES_RUNTIME_CTX: contextvars.ContextVar[dict | None] = contextvars.ContextVar("t2a_rules_runtime", default=None)
def set_rules_runtime(ctx: dict) -> contextvars.Token:
"""Active un contexte de règles pour le dossier courant."""
return _RULES_RUNTIME_CTX.set(ctx)
def reset_rules_runtime(token: contextvars.Token) -> None:
"""Restaure le contexte précédent."""
_RULES_RUNTIME_CTX.reset(token)
def get_rules_runtime() -> dict | None:
return _RULES_RUNTIME_CTX.get()
@lru_cache(maxsize=1)
def load_rules_router() -> Dict[str, Any]:
"""Charge la config de routage (config/rules/router.yaml).
- mode: 'strict' => une règle non listée dans base.yaml est considérée désactivée
quand le routage runtime est actif (objectif: éviter les surprises).
- defaults.enabled_packs: packs actifs par défaut sur tous les dossiers.
- triggers: conditions simples qui activent des packs additionnels.
"""
defaults: Dict[str, Any] = {
"version": 1,
"mode": "strict",
"defaults": {"enabled_packs": ["vetos_core", "decisions_core"]},
"triggers": [],
}
path = RULES_ROUTER_PATH
if not path.exists():
return defaults
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
# merge conservateur
if isinstance(data, dict):
defaults.update({k: v for k, v in data.items() if v is not None})
return defaults
except Exception:
return defaults
def rule_enabled(rule_id: str) -> bool:
"""Retourne True si la règle est activée.
Mode legacy (pas de routage runtime): une règle inconnue => True (comportement historique).
Mode routé (runtime actif):
- On *garde* l'info 'enabled' du catalogue (base.yaml / overlays)
- On **désactive** automatiquement les règles dont le pack n'est pas dans enabled_packs
- En mode 'strict', une règle inconnue => False (ça évite les surprises en prod)
"""
catalog = load_rules_catalog()
cfg = catalog.get(rule_id)
runtime = get_rules_runtime()
if runtime is None:
# legacy
if not cfg:
return True
return bool(cfg.get("enabled", True))
mode = str(runtime.get("mode") or "strict").lower()
enabled_packs = set(runtime.get("enabled_packs") or [])
always_on = set(runtime.get("always_on_rules") or [])
force_enable = set(runtime.get("force_enable_rules") or [])
force_disable = set(runtime.get("force_disable_rules") or [])
if rule_id in force_disable:
return False
if rule_id in force_enable:
return True
# Règles inconnues: strict => off, legacy => on
if cfg is None:
return False if mode == "strict" else True
# Respecte le flag d'activation du catalogue (l'admin peut couper une règle)
if not bool(cfg.get("enabled", True)):
return False
pack = cfg.get("pack")
if pack and (pack not in enabled_packs) and (rule_id not in always_on):
return False
return True
def rule_force_severity(rule_id: str) -> str | None:
"""Optionnel: force la sévérité d'un veto (HARD/MEDIUM/LOW)."""
cfg = load_rules_catalog().get(rule_id) or {}
sev = cfg.get("force_severity")
return str(sev) if sev else None
# --- Modèles de données CIM-10 ---
class RAGSource(BaseModel):
document: str
page: Optional[int] = None
code: Optional[str] = None
extrait: Optional[str] = None
class Sejour(BaseModel):
sexe: Optional[str] = None
age: Optional[int] = None
date_entree: Optional[str] = None
date_sortie: Optional[str] = None
duree_sejour: Optional[int] = None
mode_entree: Optional[str] = None
mode_sortie: Optional[str] = None
imc: Optional[float] = None
poids: Optional[float] = None
taille: Optional[float] = None
class PreuveClinique(BaseModel):
type: str # "biologie" | "imagerie" | "traitement" | "acte" | "clinique"
element: str # "CRP 180 mg/L"
interpretation: str # "syndrome inflammatoire majeur"
class CodeDecision(BaseModel):
"""Décision finale sur un code (audit-friendly).
- action=KEEP: on garde la suggestion
- action=DOWNGRADE: on remplace par un code moins spécifique (ex: D50→D64.9)
- action=REMOVE: on retire le code (ou on le laisse vide)
"""
action: str = "KEEP" # KEEP | DOWNGRADE | REMOVE
final_code: Optional[str] = None
downgraded_from: Optional[str] = None
reason: Optional[str] = None
needs_info: list[str] = Field(default_factory=list)
applied_rules: list[str] = Field(default_factory=list)
class Diagnostic(BaseModel):
texte: str
cim10_suggestion: Optional[str] = None
cim10_confidence: Optional[str] = None
# Statut clinique / qualité (pour affichage "barré" et exclusion métriques)
# - confirmed/probable/uncertain: actifs
# - ruled_out: visible mais barré (n'entre pas dans les métriques/GHM)
status: Optional[str] = None
ruled_out_reason: Optional[str] = None
# Sortie finale (post-traitement qualité)
cim10_final: Optional[str] = None
cim10_decision: Optional[CodeDecision] = None
justification: Optional[str] = None
raisonnement: Optional[str] = None
sources_rag: list[RAGSource] = Field(default_factory=list)
preuves_cliniques: list[PreuveClinique] = Field(default_factory=list)
est_cma: Optional[bool] = None
est_cms: Optional[bool] = None
niveau_severite: Optional[str] = None # "leger" | "modere" | "severe" | "non_evalue"
niveau_cma: Optional[int] = None # 1 (pas CMA) | 2 | 3 | 4 (niveau officiel ATIH)
source: Optional[str] = None # "trackare" | "edsnlp" | "regex" | "llm_das"
source_page: Optional[int] = None # numéro de page (1-indexed) dans le PDF source
source_excerpt: Optional[str] = None # extrait du texte source (~200 chars)
class DossierMetrics(BaseModel):
"""Métriques de qualité / reporting (audit-friendly).
Objectif : distinguer les éléments *actifs* (qui comptent pour le codage / GHM)
de ceux écartés par les règles qualité (vetos / décisions).
"""
das_total: int = 0
das_active: int = 0
das_excluded: int = 0 # total - active
das_removed: int = 0 # décision REMOVE (future: ruled_out)
das_ruled_out: int = 0 # visible mais barré (action RULED_OUT)
das_no_code: int = 0 # pas de code suggestion/final
actes_total: int = 0
actes_with_code: int = 0
dp_has_code: bool = False
class ActeCCAM(BaseModel):
texte: str
code_ccam_suggestion: Optional[str] = None
ccam_confidence: Optional[str] = None
justification: Optional[str] = None
raisonnement: Optional[str] = None
sources_rag: list[RAGSource] = Field(default_factory=list)
date: Optional[str] = None
validite: Optional[str] = None # "valide" | "obsolete" | "non_verifie"
alertes: list[str] = Field(default_factory=list)
source_page: Optional[int] = None
source_excerpt: Optional[str] = None
class Traitement(BaseModel):
medicament: str
posologie: Optional[str] = None
code_atc: Optional[str] = None
source_page: Optional[int] = None
source_excerpt: Optional[str] = None
class BiologieCle(BaseModel):
test: str
valeur: Optional[str] = None
# Valeur numérique parsée (si possible). Sert aux règles qualité.
valeur_num: Optional[float] = None
anomalie: Optional[bool] = None
# Qualité de parsing: ok | suspect | discarded
quality: Optional[str] = None
discard_reason: Optional[str] = None
source_page: Optional[int] = None
source_excerpt: Optional[str] = None
class Imagerie(BaseModel):
type: str
conclusion: Optional[str] = None
score: Optional[str] = None
source_page: Optional[int] = None
source_excerpt: Optional[str] = None
class Antecedent(BaseModel):
texte: str
source_page: Optional[int] = None
source_excerpt: Optional[str] = None
class Complication(BaseModel):
texte: str
source_page: Optional[int] = None
source_excerpt: Optional[str] = None
class DossierMedical(BaseModel):
source_file: str = ""
document_type: str = ""
sejour: Sejour = Field(default_factory=Sejour)
diagnostic_principal: Optional[Diagnostic] = None
diagnostics_associes: list[Diagnostic] = Field(default_factory=list)
actes_ccam: list[ActeCCAM] = Field(default_factory=list)
antecedents: list[Antecedent] = Field(default_factory=list)
traitements_sortie: list[Traitement] = Field(default_factory=list)
biologie_cle: list[BiologieCle] = Field(default_factory=list)
# Valeurs biologiques écartées (artefacts PDF/OCR) pour audit
biologie_discarded: list[dict] = Field(default_factory=list)
imagerie: list[Imagerie] = Field(default_factory=list)
complications: list[Complication] = Field(default_factory=list)
alertes_codage: list[str] = Field(default_factory=list)
source_files: list[str] = Field(default_factory=list)
ghm_estimation: Optional[GHMEstimation] = None
controles_cpam: list[ControleCPAM] = Field(default_factory=list)
veto_report: Optional["VetoReport"] = None
processing_time_s: float | None = None
metrics: Optional[DossierMetrics] = None
rules_runtime: Optional[dict] = None
@field_validator("antecedents", mode="before")
@classmethod
def _coerce_antecedents(cls, v):
"""Backward compat : convertit les anciennes list[str] en list[Antecedent]."""
if not isinstance(v, list):
return v
result = []
for item in v:
if isinstance(item, str):
result.append({"texte": item})
else:
result.append(item)
return result
@field_validator("complications", mode="before")
@classmethod
def _coerce_complications(cls, v):
"""Backward compat : convertit les anciennes list[str] en list[Complication]."""
if not isinstance(v, list):
return v
result = []
for item in v:
if isinstance(item, str):
result.append({"texte": item})
else:
result.append(item)
return result
# --- Rapport d'anonymisation ---
class GHMEstimation(BaseModel):
cmd: Optional[str] = None
cmd_libelle: Optional[str] = None
type_ghm: Optional[str] = None # "C" / "M" / "K"
severite: int = 1 # 1-4
ghm_approx: Optional[str] = None # ex: "07C??2"
cma_count: int = 0
cms_count: int = 0
alertes: list[str] = Field(default_factory=list)
class ControleCPAM(BaseModel):
numero_ogc: int
titre: str = ""
arg_ucr: str = ""
decision_ucr: str = ""
dp_ucr: Optional[str] = None
da_ucr: Optional[str] = None
dr_ucr: Optional[str] = None
actes_ucr: Optional[str] = None
contre_argumentation: Optional[str] = None
response_data: Optional[dict] = None
sources_reponse: list[RAGSource] = Field(default_factory=list)
quality_tier: Optional[str] = None # "A" | "B" | "C"
requires_review: bool = False
quality_warnings: list[str] = Field(default_factory=list)
# --- Qualité / Vetos (contestabilité) ---
class VetoIssue(BaseModel):
"""Un problème détecté lors du contrôle de contestabilité."""
veto: str
severity: str # HARD | MEDIUM | LOW
where: str
message: str
class VetoReport(BaseModel):
"""Rapport global de vetos pour un dossier."""
verdict: str # PASS | NEED_INFO | FAIL
score_contestabilite: int = 100 # 0-100
issues: list[VetoIssue] = Field(default_factory=list)
class AnonymizationReport(BaseModel):
source_file: str
total_replacements: int = 0
regex_replacements: int = 0
ner_replacements: int = 0
sweep_replacements: int = 0
entities_found: list[dict] = Field(default_factory=list)
Reference in New Issue View Git Blame Copy Permalink