feat(q1): G - B-1 métadonnées sortie (audit.jsonl + XMP PDF)
Implémentation de la traçabilité B-1 sur les sorties d'anonymisation.
## .audit.jsonl — entrée metadata en 1ère ligne
Chaque .audit.jsonl commence maintenant par une entrée :
{"type": "metadata",
"app_version": "0.11.0-mvp",
"build_date": "...",
"build_commit": "...",
"build_branch": "...",
"processed_at": "<iso>",
"document_name": "...",
"ocr_used": bool,
"extracted_chars": int,
"quarantine_flags": []}
Permet de prouver a posteriori avec quelle config un document a été
anonymisé (audit DPO / CNIL).
## XMP PDF — _apply_pseudo_xmp_metadata()
Helper appelé avant doc.save() dans redact_pdf_vector et redact_pdf_raster :
1. doc.set_metadata({}) — efface TOUTES les métadonnées source
(CRITIQUE : les PDF source peuvent contenir le nom patient dans
/Author, /Title, /Keywords)
2. Pose nos métadonnées : creator/producer "Pseudonymisation v...",
title="Document anonymise", author vide, keywords avec commit+ts
3. Garde-fou : log + overwrite si une métadonnée source survit
(defense in depth)
## Constantes module-level
- APP_VERSION = "0.11.0-mvp" (à incrémenter avant chaque rebuild release)
- BUILD_DATE/BUILD_COMMIT/BUILD_BRANCH chargés depuis build_info.py
(regénéré à chaque rebuild EXE). Fallback "dev/unknown" en dev.
## Tests
74 passed, 10 xfailed — pas de régression.
Ref: docs/coordination/inbox/for-dom/2026-05-29_consolide_pseudocode-Q1-v2.md §7
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -20,6 +20,7 @@ import re
|
|||||||
import shutil
|
import shutil
|
||||||
import sys
|
import sys
|
||||||
from concurrent.futures import ProcessPoolExecutor
|
from concurrent.futures import ProcessPoolExecutor
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
@@ -103,6 +104,17 @@ except ImportError:
|
|||||||
SEUIL_TEXTE_MINI = 100
|
SEUIL_TEXTE_MINI = 100
|
||||||
SEUIL_RESCAN_RESIDUEL = 0
|
SEUIL_RESCAN_RESIDUEL = 0
|
||||||
|
|
||||||
|
# B-1 — métadonnées de sortie (audit.jsonl + XMP PDF)
|
||||||
|
# build_info.py est regénéré à chaque rebuild EXE (cf scripts/build_*).
|
||||||
|
# En mode dev, fallback sur valeurs neutres.
|
||||||
|
try:
|
||||||
|
from build_info import BUILD_DATE, BUILD_COMMIT, BUILD_BRANCH # type: ignore
|
||||||
|
except ImportError:
|
||||||
|
BUILD_DATE = "dev"
|
||||||
|
BUILD_COMMIT = "unknown"
|
||||||
|
BUILD_BRANCH = "dev"
|
||||||
|
APP_VERSION = "0.11.0-mvp" # incrémenter avant rebuild release
|
||||||
|
|
||||||
|
|
||||||
def _load_edsnlp_drug_names() -> set:
|
def _load_edsnlp_drug_names() -> set:
|
||||||
"""Charge les noms de médicaments mono-mot depuis edsnlp/resources/drugs.json.
|
"""Charge les noms de médicaments mono-mot depuis edsnlp/resources/drugs.json.
|
||||||
@@ -3868,6 +3880,50 @@ def _search_whole_word(page, token: str) -> list:
|
|||||||
rects.append(fitz.Rect(w[0], w[1], w[2], w[3]))
|
rects.append(fitz.Rect(w[0], w[1], w[2], w[3]))
|
||||||
return rects
|
return rects
|
||||||
|
|
||||||
|
def _apply_pseudo_xmp_metadata(doc) -> None:
|
||||||
|
"""B-1 — pose les métadonnées XMP de l'application sur un PDF de sortie.
|
||||||
|
|
||||||
|
CRITIQUE : efface d'abord les métadonnées source pour éviter les fuites
|
||||||
|
(les PDF d'origine peuvent contenir le nom du patient dans /Author, /Title,
|
||||||
|
/Keywords). Pose ensuite nos propres métadonnées et vérifie l'absence de
|
||||||
|
fuite résiduelle via assertion log.
|
||||||
|
"""
|
||||||
|
if doc is None:
|
||||||
|
return
|
||||||
|
# 1. Effacer toutes les métadonnées source
|
||||||
|
try:
|
||||||
|
doc.set_metadata({})
|
||||||
|
except Exception as e:
|
||||||
|
log.warning("XMP clear failed: %s", e)
|
||||||
|
# 2. Poser nos métadonnées
|
||||||
|
now_iso = datetime.now().astimezone().isoformat()
|
||||||
|
new_meta = {
|
||||||
|
"creator": f"Pseudonymisation v{APP_VERSION}",
|
||||||
|
"producer": f"Pseudonymisation v{APP_VERSION} commit {BUILD_COMMIT}",
|
||||||
|
"title": "Document anonymise",
|
||||||
|
"subject": "Pseudonymisation medicale",
|
||||||
|
"keywords": f"pseudonymisation; commit={BUILD_COMMIT}; ts={now_iso}",
|
||||||
|
"author": "",
|
||||||
|
"creationDate": "",
|
||||||
|
"modDate": "",
|
||||||
|
}
|
||||||
|
try:
|
||||||
|
doc.set_metadata(new_meta)
|
||||||
|
except Exception as e:
|
||||||
|
log.warning("XMP set_metadata failed: %s", e)
|
||||||
|
return
|
||||||
|
# 3. Garde-fou : vérifier qu'aucune métadonnée source n'a survécu
|
||||||
|
final_meta = doc.metadata or {}
|
||||||
|
for key in ("author", "title", "subject", "creator"):
|
||||||
|
val = (final_meta.get(key) or "")
|
||||||
|
if val and "Pseudonymisation" not in val and val not in ("Document anonymise", ""):
|
||||||
|
log.error("XMP leak suspect: %s=%r — overwriting", key, val)
|
||||||
|
try:
|
||||||
|
doc.set_metadata({**new_meta, key: ""})
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
def redact_pdf_vector(original_pdf: Path, audit: List[PiiHit], out_pdf: Path, ocr_word_map: OcrWordMap = None) -> None:
|
def redact_pdf_vector(original_pdf: Path, audit: List[PiiHit], out_pdf: Path, ocr_word_map: OcrWordMap = None) -> None:
|
||||||
if fitz is None:
|
if fitz is None:
|
||||||
raise RuntimeError("PyMuPDF non disponible – installez pymupdf.")
|
raise RuntimeError("PyMuPDF non disponible – installez pymupdf.")
|
||||||
@@ -3956,6 +4012,9 @@ def redact_pdf_vector(original_pdf: Path, audit: List[PiiHit], out_pdf: Path, oc
|
|||||||
# que process_pdf flag la quarantaine PDF.
|
# que process_pdf flag la quarantaine PDF.
|
||||||
log.warning("apply_redactions failed on page %d: %s", page.number, e)
|
log.warning("apply_redactions failed on page %d: %s", page.number, e)
|
||||||
raise
|
raise
|
||||||
|
# B-1 : nettoyer les métadonnées source (peuvent contenir nom patient
|
||||||
|
# dans /Author, /Title, etc.) puis poser nos propres métadonnées XMP.
|
||||||
|
_apply_pseudo_xmp_metadata(doc)
|
||||||
doc.save(str(out_pdf), deflate=True, garbage=4, clean=True, incremental=False)
|
doc.save(str(out_pdf), deflate=True, garbage=4, clean=True, incremental=False)
|
||||||
doc.close()
|
doc.close()
|
||||||
|
|
||||||
@@ -4178,6 +4237,8 @@ def redact_pdf_raster(original_pdf: Path, audit: List[PiiHit], out_pdf: Path, dp
|
|||||||
for pno, img_bytes, w, h in results:
|
for pno, img_bytes, w, h in results:
|
||||||
dst = out.new_page(width=w, height=h)
|
dst = out.new_page(width=w, height=h)
|
||||||
dst.insert_image(fitz.Rect(0, 0, w, h), stream=img_bytes)
|
dst.insert_image(fitz.Rect(0, 0, w, h), stream=img_bytes)
|
||||||
|
# B-1 : métadonnées XMP appliquées avant save (clear source + pose nôtre)
|
||||||
|
_apply_pseudo_xmp_metadata(out)
|
||||||
out.save(str(out_pdf), deflate=True, garbage=4, clean=True)
|
out.save(str(out_pdf), deflate=True, garbage=4, clean=True)
|
||||||
out.close()
|
out.close()
|
||||||
|
|
||||||
@@ -4743,8 +4804,24 @@ def process_pdf(
|
|||||||
# Filtrer les entrées de propagation globale (page=-1) avant d'écrire l'audit
|
# Filtrer les entrées de propagation globale (page=-1) avant d'écrire l'audit
|
||||||
# Ces entrées sont utilisées pour le remplacement dans le texte mais ne sont pas des détections réelles
|
# Ces entrées sont utilisées pour le remplacement dans le texte mais ne sont pas des détections réelles
|
||||||
audit_for_file = [hit for hit in anon.audit if hit.page != -1]
|
audit_for_file = [hit for hit in anon.audit if hit.page != -1]
|
||||||
|
|
||||||
|
# B-1 : entrée metadata en 1ère ligne du .audit.jsonl
|
||||||
|
# Permet de prouver a posteriori avec quelle config un document a été
|
||||||
|
# anonymisé (traçabilité DPO / CNIL).
|
||||||
|
metadata_record = {
|
||||||
|
"type": "metadata",
|
||||||
|
"app_version": APP_VERSION,
|
||||||
|
"build_date": BUILD_DATE,
|
||||||
|
"build_commit": BUILD_COMMIT,
|
||||||
|
"build_branch": BUILD_BRANCH,
|
||||||
|
"processed_at": datetime.now().astimezone().isoformat(),
|
||||||
|
"document_name": base,
|
||||||
|
"ocr_used": bool(ocr_used),
|
||||||
|
"extracted_chars": int(extracted_chars),
|
||||||
|
"quarantine_flags": [],
|
||||||
|
}
|
||||||
with audit_path.open("w", encoding="utf-8") as f:
|
with audit_path.open("w", encoding="utf-8") as f:
|
||||||
|
f.write(json.dumps(metadata_record, ensure_ascii=False) + "\n")
|
||||||
for hit in audit_for_file:
|
for hit in audit_for_file:
|
||||||
f.write(json.dumps(hit.__dict__, ensure_ascii=False) + "\n")
|
f.write(json.dumps(hit.__dict__, ensure_ascii=False) + "\n")
|
||||||
outputs = {"text": str(txt_path), "audit": str(audit_path)}
|
outputs = {"text": str(txt_path), "audit": str(audit_path)}
|
||||||
|
|||||||
Reference in New Issue
Block a user