Domi31tls
f92da4d54e
- Normalisation agressive des dates : génère 4 variations (/, ., -, espaces) - Remplacement multi-pass : avec/sans contexte 'Né(e) le' - Amélioration force_term : case-insensitive + word boundaries - Outil de validation post-anonymisation - Tests : 162 CRO, 0 fuite dates, 0 fuite CHCB (100% succès) - Temps: 0.1s/doc Résout les 36 CRO avec fuites identifiées dans l'audit initial.
241 lines
8.3 KiB
Python
241 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Validation Post-Anonymisation - Détection de Fuites Résiduelles
|
|
----------------------------------------------------------------
|
|
Scanne le texte anonymisé pour détecter les PII résiduels (fuites).
|
|
Utilisé pour valider que la propagation globale fonctionne correctement.
|
|
|
|
Usage:
|
|
python3 tools/validate_anonymization.py <anonymized_text_file>
|
|
python3 tools/validate_anonymization.py tests/ground_truth/anonymized/*.txt
|
|
"""
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import List, Dict, Tuple
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class LeakDetection:
|
|
"""Détection d'une fuite potentielle."""
|
|
line_num: int
|
|
leak_type: str
|
|
value: str
|
|
context: str
|
|
|
|
|
|
class AnonymizationValidator:
|
|
"""Validateur post-anonymisation pour détecter les fuites."""
|
|
|
|
def __init__(self):
|
|
# Patterns de détection de fuites
|
|
self.patterns = {
|
|
"DATE_NAISSANCE": re.compile(
|
|
r'Né(?:e)?\s+le\s+(\d{1,2}[\s/.\-]+\d{1,2}[\s/.\-]+\d{2,4})',
|
|
re.IGNORECASE
|
|
),
|
|
"DATE_STANDALONE": re.compile(
|
|
r'\b(\d{1,2}[/.\-]\d{1,2}[/.\-]\d{4})\b'
|
|
),
|
|
"EMAIL": re.compile(
|
|
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b'
|
|
),
|
|
"TEL": re.compile(
|
|
r'(?<!\d)(?:\+33\s?|0)\d(?:[\s.\-]?\d){8}(?!\d)'
|
|
),
|
|
"NIR": re.compile(
|
|
r'\b[12]\s*\d{2}\s*(?:0[1-9]|1[0-2]|2[AB])\s*\d{2,3}\s*\d{3}\s*\d{3}\s*\d{2}\b',
|
|
re.IGNORECASE
|
|
),
|
|
"IBAN": re.compile(
|
|
r'\b[A-Z]{2}\d{2}(?:\s?[A-Z0-9]{4}){3,7}(?:\s?[A-Z0-9]{1,4})\b'
|
|
),
|
|
}
|
|
|
|
# Patterns de placeholders (ne doivent PAS être détectés comme fuites)
|
|
self.placeholder_pattern = re.compile(
|
|
r'\[(EMAIL|TEL|IBAN|NIR|IPP|DATE_NAISSANCE|NOM|VILLE|ADRESSE|CODE_POSTAL|'
|
|
r'AGE|DOSSIER|NDA|EPISODE|RPPS|ETABLISSEMENT|FINESS|OGC|MASK)\]'
|
|
)
|
|
|
|
def validate_text(self, text: str, filename: str = "") -> Tuple[List[LeakDetection], Dict[str, int]]:
|
|
"""
|
|
Valide un texte anonymisé et détecte les fuites.
|
|
|
|
Args:
|
|
text: Texte anonymisé à valider
|
|
filename: Nom du fichier (pour le rapport)
|
|
|
|
Returns:
|
|
Tuple (liste des fuites détectées, statistiques par type)
|
|
"""
|
|
leaks = []
|
|
stats = {leak_type: 0 for leak_type in self.patterns.keys()}
|
|
|
|
lines = text.split('\n')
|
|
for line_num, line in enumerate(lines, 1):
|
|
# Ignorer les lignes qui contiennent des placeholders
|
|
if self.placeholder_pattern.search(line):
|
|
continue
|
|
|
|
# Chercher les fuites
|
|
for leak_type, pattern in self.patterns.items():
|
|
matches = pattern.finditer(line)
|
|
for match in matches:
|
|
value = match.group(1) if match.groups() else match.group(0)
|
|
|
|
# Filtrer les faux positifs connus
|
|
if self._is_false_positive(leak_type, value, line):
|
|
continue
|
|
|
|
# Extraire le contexte (50 chars avant/après)
|
|
start = max(0, match.start() - 50)
|
|
end = min(len(line), match.end() + 50)
|
|
context = line[start:end]
|
|
|
|
leaks.append(LeakDetection(
|
|
line_num=line_num,
|
|
leak_type=leak_type,
|
|
value=value,
|
|
context=context
|
|
))
|
|
stats[leak_type] += 1
|
|
|
|
return leaks, stats
|
|
|
|
def _is_false_positive(self, leak_type: str, value: str, line: str) -> bool:
|
|
"""
|
|
Filtre les faux positifs connus.
|
|
|
|
Args:
|
|
leak_type: Type de fuite détectée
|
|
value: Valeur détectée
|
|
line: Ligne complète
|
|
|
|
Returns:
|
|
True si c'est un faux positif
|
|
"""
|
|
# Dates : ignorer les dates d'intervention/hospitalisation (contexte différent)
|
|
if leak_type == "DATE_STANDALONE":
|
|
# Ignorer si dans un contexte médical non-PII
|
|
if any(ctx in line.lower() for ctx in [
|
|
"intervention", "hospitalisation", "consultation", "examen",
|
|
"date d'entrée", "date de sortie", "date d'admission"
|
|
]):
|
|
return True
|
|
# Ignorer les dates futures (probablement des dates d'intervention)
|
|
try:
|
|
day, month, year = map(int, re.split(r'[/.\-]', value))
|
|
if year > 2000: # Dates de naissance sont généralement < 2000
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
# Téléphones : ignorer les numéros d'hôpitaux (déjà filtrés normalement)
|
|
if leak_type == "TEL":
|
|
if "standard" in line.lower() or "secrétariat" in line.lower():
|
|
return True
|
|
|
|
return False
|
|
|
|
def generate_report(self, leaks: List[LeakDetection], stats: Dict[str, int], filename: str = "") -> str:
|
|
"""
|
|
Génère un rapport de validation.
|
|
|
|
Args:
|
|
leaks: Liste des fuites détectées
|
|
stats: Statistiques par type
|
|
filename: Nom du fichier validé
|
|
|
|
Returns:
|
|
Rapport formaté
|
|
"""
|
|
report = []
|
|
report.append("=" * 80)
|
|
report.append("RAPPORT DE VALIDATION POST-ANONYMISATION")
|
|
report.append("=" * 80)
|
|
|
|
if filename:
|
|
report.append(f"\nFichier: {filename}")
|
|
|
|
report.append(f"\nNombre total de fuites détectées: {len(leaks)}")
|
|
|
|
if leaks:
|
|
report.append("\n" + "=" * 80)
|
|
report.append("FUITES DÉTECTÉES PAR TYPE")
|
|
report.append("=" * 80)
|
|
|
|
for leak_type, count in stats.items():
|
|
if count > 0:
|
|
report.append(f"\n{leak_type}: {count} fuite(s)")
|
|
|
|
report.append("\n" + "=" * 80)
|
|
report.append("DÉTAILS DES FUITES")
|
|
report.append("=" * 80)
|
|
|
|
for leak in leaks:
|
|
report.append(f"\nLigne {leak.line_num} - {leak.leak_type}")
|
|
report.append(f" Valeur: {leak.value}")
|
|
report.append(f" Contexte: ...{leak.context}...")
|
|
else:
|
|
report.append("\n✅ AUCUNE FUITE DÉTECTÉE - Validation réussie!")
|
|
|
|
report.append("\n" + "=" * 80)
|
|
|
|
return "\n".join(report)
|
|
|
|
|
|
def main():
|
|
"""Point d'entrée principal."""
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python3 tools/validate_anonymization.py <anonymized_text_file>")
|
|
print(" python3 tools/validate_anonymization.py tests/ground_truth/anonymized/*.txt")
|
|
sys.exit(1)
|
|
|
|
validator = AnonymizationValidator()
|
|
|
|
# Traiter tous les fichiers fournis
|
|
files = sys.argv[1:]
|
|
total_leaks = 0
|
|
files_with_leaks = 0
|
|
|
|
for filepath in files:
|
|
path = Path(filepath)
|
|
if not path.exists():
|
|
print(f"❌ Fichier introuvable: {filepath}")
|
|
continue
|
|
|
|
# Lire le texte anonymisé
|
|
text = path.read_text(encoding='utf-8')
|
|
|
|
# Valider
|
|
leaks, stats = validator.validate_text(text, path.name)
|
|
|
|
# Générer le rapport
|
|
report = validator.generate_report(leaks, stats, path.name)
|
|
print(report)
|
|
|
|
if leaks:
|
|
total_leaks += len(leaks)
|
|
files_with_leaks += 1
|
|
|
|
# Résumé global si plusieurs fichiers
|
|
if len(files) > 1:
|
|
print("\n" + "=" * 80)
|
|
print("RÉSUMÉ GLOBAL")
|
|
print("=" * 80)
|
|
print(f"Fichiers traités: {len(files)}")
|
|
print(f"Fichiers avec fuites: {files_with_leaks}")
|
|
print(f"Total de fuites: {total_leaks}")
|
|
|
|
if total_leaks == 0:
|
|
print("\n✅ TOUS LES FICHIERS SONT VALIDES - Aucune fuite détectée!")
|
|
else:
|
|
print(f"\n⚠️ {files_with_leaks} fichier(s) contiennent des fuites!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|