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}'\-]*)\]"
)
# « Prénom NOM » inversé, sans parenthèses ni crochets (ex. « Alix DATTIN »).
# 2e mot tout en MAJUSCULES → faible risque de FP (« Mozilla Firefox » ne matche pas).
RE_PRENOM_NOM = re.compile(rf"\b[{_MAJ}][{_MIN}]+\s+[{_MAJ}][{_MAJ}\-']+\b")

# GXD5 Diagnostics : numéro de dossier + nom patient tout-majuscules.
# Format réel : « GXD5 Diagnostics - 128008 - BENVENISTE MARIE-LAURENCE »
# Le numéro (128008) = ID dossier patient (PII). Le nom = PII.
# 2 groupes de capture : (1)=numéro, (2)=nom complet.
RE_GXD5_DIAG = re.compile(
    rf"GXD5\s+Diagnostics\s*-\s*(\d+)\s*-\s*([{_MAJ}][{_MAJ}\-' ]+)"
)

# 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_GXD5_DIAG, "DOSSIER", 1),   # numéro de dossier
    (RE_PRENOM_NOM, "NOM", 0),
    (RE_EMAIL, "EMAIL", 0),
    (RE_NIR, "NIR", 0),
    (RE_IPP, "IPP", 1),
    (RE_TEL, "TEL", 0),
    (RE_AGE, "AGE", 0),
]
# GXD5 nom (groupe 2) traité séparément — même regex, priorité juste après.
_DETECTORS.append((RE_GXD5_DIAG, "NOM", 2))

# 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é = rang détecteur, puis -longueur)
    # _DETECTORS est ordonné par priorité ; le rang dans cette liste détermine
    # qui gagne quand deux patterns chevauchent. Plus prioritaire + plus long
    # = accepté en premier, les plus courts/moins prioritaires sont éliminés.
    # Fix FN « Dossier VIOLA (VIOLA) Liliane » : RE_PRENOM_NOM captait
    # « Dossier VIOLA » (rang 2) et bloquait RE_NOM_NAISSANCE « VIOLA (VIOLA)
    # Liliane » (rang 0, plus prioritaire et plus long).
    det_rank = {p: i for i, (p, _, _) in enumerate(_DETECTORS)}
    spans.sort(key=lambda s: (det_rank.get(s[2], 999), -(s[1] - s[0]), s[0]))
    occupied: List[Tuple[int, int]] = []
    accepted: List[Tuple[int, int, str, str]] = []
    for start, end, etype, value in spans:
        if all(start >= oe or end <= os for os, oe in occupied):
            accepted.append((start, end, etype, value))
            occupied.append((start, 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


# Clés portant un titre de fenêtre, où qu'elles soient imbriquées dans l'event
# (top-level `active_window_title`, `window/to/from.title`, et surtout
# `vision_info.window_capture.window_title` — blind spot signalé par Qwen).
_TITLE_KEYS = ("title", "window_title", "active_window_title")
_PLACEHOLDER_SAISIE = "[SAISIE]"


def _walk_titles(obj, mapping: Dict) -> None:
    """Parcourt récursivement l'event et assainit toute valeur de titre de fenêtre."""
    if isinstance(obj, dict):
        for k, v in obj.items():
            if k in _TITLE_KEYS and isinstance(v, str):
                obj[k] = anonymize_text(v, mapping=mapping)[0]
            else:
                _walk_titles(v, mapping)
    elif isinstance(obj, list):
        for item in obj:
            _walk_titles(item, mapping)


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

    # tous les titres de fenêtre, où qu'ils soient imbriqués
    # (active_window_title, window/to/from.title, vision_info.window_capture.window_title…)
    _walk_titles(ev, mapping)

    return ev


# Clés d'un workflow core portant du texte potentiellement PII : cible OCR
# (`by_text`), noms d'écrans/labels dérivés des titres. Le contenu saisi est
# déjà neutralisé à la source (sanitize_event → [SAISIE]).
_WORKFLOW_TEXT_KEYS = ("by_text", "name", "label")


def _walk_workflow_text(obj, mapping: Dict) -> None:
    """Parcourt un workflow core et tokenise la PII des champs texte (cibles, noms)."""
    if isinstance(obj, dict):
        for k, v in obj.items():
            if k in _WORKFLOW_TEXT_KEYS and isinstance(v, str) and v:
                obj[k] = anonymize_text(v, mapping=mapping)[0]
            else:
                _walk_workflow_text(v, mapping)
    elif isinstance(obj, list):
        for item in obj:
            _walk_workflow_text(item, mapping)


def sanitize_workflow_dict(workflow_dict: Dict, *, mapping: Optional[Dict] = None) -> Dict:
    """Assainit un workflow core (JSON appris) avant import/persistance en DB VWB.

    Tokenise la PII des champs texte (cible OCR `by_text`, noms d'écrans, labels)
    via `anonymize_text`, en gardant l'interface intacte (« Léa apprend
    l'interface, pas la donnée »). Copie — l'original n'est pas muté.

    Limite (couche 1) : ne capte que la PII structurée (IPP, NOM clinique…) ;
    les noms libres relèvent de la couche 2 NER.
    """
    if mapping is None:
        mapping = {}
    wf = copy.deepcopy(workflow_dict)
    _walk_workflow_text(wf, mapping)
    return wf