feat(server): assainissement PII couche regex+structurelle (tokens typés cohérents)

pii_sanitizer.anonymize_text() remplace la PII par des tokens typés et cohérents ([IPP_1], [AGE_1], [NOM_1]) : protège la donnée ET garde la structure (type de champ) utile à l'apprentissage des variables. Sans modèle, déployable partout. Filet regex (IPP/NIR/TEL/EMAIL/AGE, repris de anonymisation) + règles structurelles cliniques (NOM (NAISSANCE) Prénom ; [Nom Prénom] PACS) + blacklist logiciels anti-FP. 5 tests verts. Couche NER (noms libres) en complément ensuite. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-28 19:08:43 +02:00
parent 46ad5973d1
commit 8e4d09594c
2 changed files with 214 additions and 0 deletions
--- a/agent_v0/server_v1/pii_sanitizer.py
+++ b/agent_v0/server_v1/pii_sanitizer.py
@@ -0,0 +1,133 @@
+"""Assainissement PII des données capturées (titres de fenêtre, texte saisi, OCR).
+
+Côté serveur. Remplace la PII par des **tokens typés et cohérents**
+(`[IPP_1]`, `[AGE_1]`, `[NOM_1]`…) : on protège la donnée **et** on garde la
+structure (champ de type NOM/IPP) utile à l'apprentissage des variables.
+
+Couche 1 (ce module, sans modèle) : filet **regex** sur la PII structurée
+(IPP, NIR, téléphone, email, âge) + règles **structurelles** des titres
+cliniques (`NOM (NAISSANCE) Prénom`, `[Nom Prénom]` des fenêtres PACS). Regex
+réutilisées du projet `anonymisation`.
+Couche 2 (à venir) : NER CamemBERT-bio (ONNX) pour les noms libres que la
+couche 1 ne capte pas — branchée plus tard, ce module marche sans.
+
+Branche feat/push-log-dgx — assainissement PII clinique.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Optional, Tuple
+
+# --- Filet regex (réutilisé de anonymisation/anonymizer_core_refactored_onnx.py) ---
+RE_IPP = re.compile(r"\b(?:I\.?P\.?P\.?|IPP|N°\s*Ipp)\s*[:\-]?\s*([A-Za-z0-9]{6,})\b", re.IGNORECASE)
+RE_NIR = re.compile(r"(?<!\d)[12]\s?\d{2}\s?\d{2}\s?\d{2}\s?\d{3}\s?\d{3}\s?\d{2}(?!\d)")
+RE_EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
+RE_TEL = re.compile(r"(?<!\d)(?:\+33\s?|0)\d(?:[ .\-]?\d){8}(?!\d)")
+# Âge format « titre » (« 90 ans »), plus large que le regex prose de anonymisation.
+RE_AGE = re.compile(r"\b(\d{1,3})\s*ans\b", re.IGNORECASE)
+
+_MAJ = r"A-ZÉÈÀÂÊÎÔÛÄËÏÖÜÇ"
+_MIN = r"a-zàâäéèêëïîôöùûüç"
+# Format clinique « NOM (NOM_NAISSANCE) Prénom » (ex. « ROSSIGNOL (SOUBIE) Pierrette »).
+RE_NOM_NAISSANCE = re.compile(
+    rf"\b[{_MAJ}][{_MAJ}\-']+\s+\([{_MAJ}][{_MAJ}\-']+\)\s+[{_MAJ}][{_MIN}\-']+\b"
+)
+# Patient entre crochets des fenêtres PACS (ex. « [DATTIN Alix] »), ≥ 2 tokens capitalisés.
+RE_NOM_BRACKET = re.compile(
+    rf"\[((?:[{_MAJ}][\w{_MIN}'\-]*\s+){{1,3}}[{_MAJ}][\w{_MIN}'\-]*)\]"
+)
+
+# Ordre = priorité ; group = portion à remplacer (0 = match entier).
+_DETECTORS: List[Tuple[re.Pattern, str, int]] = [
+    (RE_NOM_NAISSANCE, "NOM", 0),
+    (RE_NOM_BRACKET, "NOM", 0),
+    (RE_EMAIL, "EMAIL", 0),
+    (RE_NIR, "NIR", 0),
+    (RE_IPP, "IPP", 1),
+    (RE_TEL, "TEL", 0),
+    (RE_AGE, "AGE", 0),
+]
+
+# Anti-faux-positifs : termes logiciels/UI à ne jamais prendre pour un nom.
+# (Sous-ensemble inline ; les gazetteers complets arrivent avec la couche NER.)
+_SOFTWARE_BLACKLIST = {
+    "FIREFOX", "MOZILLA", "CHROME", "EDGE", "EXPERT", "SANTE", "SANTÉ", "PACS",
+    "CIM", "ARES", "EASILY", "CONSULTATION", "URGENCES", "SAISIE", "COURRIER",
+    "DOSSIER", "PATIENT", "FENETRE", "FENÊTRE", "GXD", "WINDOWS", "CITRIX",
+}
+
+
+def _normalize(etype: str, value: str) -> str:
+    """Clé de cohérence : même entité -> même token."""
+    if etype in ("IPP", "NIR", "TEL"):
+        return re.sub(r"\s+", "", value)
+    if etype == "EMAIL":
+        return value.lower()
+    return re.sub(r"\s+", " ", value).strip().upper()
+
+
+def _is_blacklisted_name(value: str) -> bool:
+    toks = [t for t in re.split(r"[^\wÀ-ÿ]+", value) if t]
+    return bool(toks) and all(t.upper() in _SOFTWARE_BLACKLIST for t in toks)
+
+
+def _assign_token(mapping: Dict, etype: str, norm: str) -> str:
+    key = (etype, norm)
+    if key in mapping:
+        return mapping[key]
+    n = 1 + sum(1 for k in mapping if isinstance(k, tuple) and k[0] == etype)
+    token = f"[{etype}_{n}]"
+    mapping[key] = token
+    return token
+
+
+def anonymize_text(
+    text: str, *, mapping: Optional[Dict] = None
+) -> Tuple[str, List[Dict]]:
+    """Remplace la PII de `text` par des tokens typés cohérents.
+
+    `mapping` : table de cohérence partagée (ex. à l'échelle d'une session) —
+    la même valeur PII reçoit le même token d'un appel à l'autre. Mutée en place ;
+    si None, une table locale est utilisée.
+
+    Retourne `(texte_assaini, entités)` où chaque entité =
+    `{"type", "original", "token", "start", "end"}` (positions dans le texte source).
+    """
+    if not text:
+        return text, []
+    if mapping is None:
+        mapping = {}
+
+    # 1) collecte des candidats (start, end, type, valeur)
+    spans: List[Tuple[int, int, str, str]] = []
+    for pattern, etype, group in _DETECTORS:
+        for m in pattern.finditer(text):
+            start, end = m.span(group)
+            if start == end:
+                continue
+            value = m.group(group)
+            if etype == "NOM" and _is_blacklisted_name(value):
+                continue
+            spans.append((start, end, etype, value))
+
+    # 2) résolution des chevauchements (priorité = ordre des détecteurs, puis position)
+    spans.sort(key=lambda s: (s[0], s[1]))
+    accepted: List[Tuple[int, int, str, str]] = []
+    last_end = -1
+    for start, end, etype, value in spans:
+        if start >= last_end:
+            accepted.append((start, end, etype, value))
+            last_end = end
+
+    # 3) substitution (de droite à gauche pour préserver les indices)
+    entities: List[Dict] = []
+    out = text
+    for start, end, etype, value in sorted(accepted, key=lambda s: s[0], reverse=True):
+        token = _assign_token(mapping, etype, _normalize(etype, value))
+        out = out[:start] + token + out[end:]
+        entities.append(
+            {"type": etype, "original": value, "token": token, "start": start, "end": end}
+        )
+    entities.reverse()
+    return out, entities
--- a/tests/unit/test_pii_sanitizer.py
+++ b/tests/unit/test_pii_sanitizer.py
@@ -0,0 +1,81 @@
+"""Tests de l'assainissement PII des données capturées (titres, texte, OCR).
+
+Couche 1 (sans modèle) : filet regex sur la PII structurée (IPP, NIR, TEL,
+EMAIL, AGE) + règles structurelles cliniques (NOM (NAISSANCE) Prénom ;
+[Nom Prénom] des fenêtres PACS), avec tokens TYPÉS et COHÉRENTS ([IPP_1]…).
+
+Réutilise l'approche du projet `anonymisation` (placeholders + regex). La
+couche NER (noms libres) viendra en complément. Cas réels remontés en clinique
+le 28/06 (anonymisés ici par construction). Branche feat/push-log-dgx.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+
+def test_ipp_et_age_tokenises():
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    titre = "VIOLA (VIOLA) Liliane 90 ans - IPP: 168246 - Expert Sante - Mozilla Firefox"
+    out, ents = anonymize_text(titre)
+
+    assert "168246" not in out, out          # IPP retiré
+    assert "[IPP_1]" in out
+    assert "90 ans" not in out               # âge retiré
+    assert "[AGE_1]" in out
+    # le nom format clinique « NOM (NAISSANCE) Prénom » est tokenisé
+    assert "VIOLA" not in out and "Liliane" not in out, out
+    assert "[NOM_1]" in out
+    # le logiciel n'est pas pris pour de la PII
+    assert "Firefox" in out and "Expert Sante" in out
+    types = {e["type"] for e in ents}
+    assert {"IPP", "AGE", "NOM"} <= types
+
+
+def test_nom_entre_crochets_pacs():
+    """Le PACS met le patient entre crochets : `[DATTIN Alix]`."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    titre = "GXD5 Pacs 4.0.4.307 CIM ARES - [DATTIN Alix] - Mozilla Firefox"
+    out, _ = anonymize_text(titre)
+
+    assert "DATTIN" not in out and "Alix" not in out, out
+    assert "[NOM_1]" in out
+    assert "Pacs" in out and "Firefox" in out  # contexte logiciel préservé
+
+
+def test_coherence_meme_ipp_meme_token():
+    """Même valeur PII -> même token (sur un mapping partagé de session)."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    mapping: dict = {}
+    o1, _ = anonymize_text("IPP: 168246 ouvert", mapping=mapping)
+    o2, _ = anonymize_text("dossier IPP: 168246 fermé", mapping=mapping)
+    o3, _ = anonymize_text("IPP: 270020 autre", mapping=mapping)
+
+    assert "[IPP_1]" in o1 and "[IPP_1]" in o2      # même patient -> même token
+    assert "[IPP_2]" in o3                           # patient différent -> token différent
+    assert "270020" not in o3
+
+
+def test_email_et_telephone():
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    out, _ = anonymize_text("contact j.dupont@chu.fr / 06 12 34 56 78")
+    assert "@chu.fr" not in out and "[EMAIL_1]" in out
+    assert "06 12 34 56 78" not in out and "[TEL_1]" in out
+
+
+def test_texte_sans_pii_inchange():
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    t = "Expert Sante - Consultation - Mozilla Firefox"
+    out, ents = anonymize_text(t)
+    assert out == t
+    assert ents == []