rpa_vision_v3/agent_v0/server_v1/pii_sanitizer.py

"""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 copy
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


# Conteneurs de titre de fenêtre dans les events (window_focus_change, clic, saisie).
_TITLE_CONTAINERS = ("window", "to", "from")
_PLACEHOLDER_SAISIE = "[SAISIE]"


def sanitize_event(event: Dict, *, mapping: Optional[Dict] = None) -> Dict:
    """Assainit un event capturé avant persistance (copie, ne mute pas l'original).

    Principe « Léa apprend l'interface, pas la donnée » (décision Dom 28/06) :
    - `text_input` : le **contenu tapé** (`text`, `raw_keys`) = donnée de santé →
      remplacé par `[SAISIE]` (on garde le champ, pas la valeur — option b) ;
    - **titres de fenêtre** (`active_window_title`, et `title` dans `window`/`to`/
      `from`) : l'**identité patient** est tokenisée, l'app/écran est gardé
      (contexte d'apprentissage), via `anonymize_text` + `mapping` partagé (cohérence).
    """
    if mapping is None:
        mapping = {}
    ev = copy.deepcopy(event)

    # text_input : on ne garde pas le contenu
    if ev.get("type") == "text_input":
        for k in ("text", "raw_keys"):
            if ev.get(k) not in (None, ""):
                ev[k] = _PLACEHOLDER_SAISIE

    # titre direct (heartbeat)
    if isinstance(ev.get("active_window_title"), str):
        ev["active_window_title"] = anonymize_text(
            ev["active_window_title"], mapping=mapping
        )[0]

    # titres imbriqués (window / to / from)
    for key in _TITLE_CONTAINERS:
        sub = ev.get(key)
        if isinstance(sub, dict) and isinstance(sub.get("title"), str):
            sub["title"] = anonymize_text(sub["title"], mapping=mapping)[0]

    return ev