Domi31tls
890edb360e
Axe CLI (intégration dans d'autres programmes) : contrat stable, codes retour fiables. `--engines` liste les moteurs réellement disponibles dans CET exécutable CLI (`[OUI]/[NON] Label (requis/optionnel) — raison`) et sort 0, sans traiter. `input` devient optionnel pour ce mode (sinon code 2). Le fail-closed CamemBERT (code 3) et le best-effort EDS/GLiNER (jamais déclarés actifs si le chargement échoue) restent inchangés. Ne présume pas du build GUI. 2 tests (--engines → code 0 + moteurs listés ; absence d'input → code 2). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
291 lines
12 KiB
Python
291 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""CLI de production — anonymise UN fichier (ou un dossier) sans GUI.
|
|
|
|
Contrat principal (demande Dom 2026-06-10) :
|
|
|
|
Anonymisation-CLI.exe <fichier> <dossier_sortie>
|
|
|
|
- argument 1 : fichier unique existant (ou dossier parcouru récursivement) ;
|
|
- argument 2 : dossier de sortie (créé si absent) ; `--out` reste accepté en
|
|
compatibilité ;
|
|
- chemins avec espaces et accents supportés ;
|
|
- pas de GUI ;
|
|
- log lisible à côté de l'EXE (frozen) ou dans le cwd (venv).
|
|
|
|
Codes retour :
|
|
- 0 : anonymisation terminée, sortie produite ;
|
|
- 2 : entrée manquante (fichier/dossier introuvable ou aucun document) ;
|
|
- 3 : modèle OBLIGATOIRE indisponible (échec fail-closed, pas de mode dégradé
|
|
silencieux) ;
|
|
- 4 : sortie non produite (mise en quarantaine résiduelle ou fichier absent) ;
|
|
- 1 : erreur de traitement (exception).
|
|
|
|
Modèles (alignés sur la dernière version du moteur, comme la GUI v5) :
|
|
- OBLIGATOIRE : CamemBERT-bio ONNX (`models/camembert-bio-deid/onnx/model.onnx`).
|
|
Si absent/illisible et NER demandé -> le CLI ÉCHOUE (code 3), il n'affiche
|
|
jamais « OK » en mode dégradé.
|
|
- OPTIONNELS : EDS-Pseudo, GLiNER. Chargés best effort, tracés dans le log ;
|
|
leur absence est explicitement signalée, jamais masquée.
|
|
- `--no-ner` : mode regex seul ASSUMÉ par l'opérateur -> aucun modèle obligatoire.
|
|
|
|
Compatible frozen (PyInstaller) ET venv : en frozen, se positionne dans
|
|
`sys._MEIPASS` (comme launcher.py) pour que le core retrouve config/ data/
|
|
models/ ; écrit le log à côté de l'EXE ; résout les chemins entrée/sortie
|
|
contre le cwd de lancement.
|
|
|
|
Sortie de production (identique à la GUI v5) : burn raster
|
|
(`make_vector_redaction=False`, `also_make_raster_burn=True`), plus le texte
|
|
pseudonymisé (`.pseudonymise.txt`) et l'audit (`.audit.jsonl`).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# Console Windows (cp1252) : forcer UTF-8 pour éviter UnicodeEncodeError sur les
|
|
# accents (le FileHandler est déjà en utf-8).
|
|
for _stream in (sys.stdout, sys.stderr):
|
|
try:
|
|
_stream.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
|
|
except Exception:
|
|
pass
|
|
|
|
# cwd de lancement, capturé AVANT tout chdir (sert à résoudre entrée/sortie).
|
|
_LAUNCH_CWD = Path.cwd()
|
|
|
|
# Résolution des chemins en frozen, alignée sur launcher.py : les datas
|
|
# (config/, data/, models/) sont extraites dans _MEIPASS ; on s'y positionne
|
|
# pour que le core les trouve. Le log va à côté de l'EXE.
|
|
if getattr(sys, "frozen", False):
|
|
_APP_DIR = Path(sys._MEIPASS) # type: ignore[attr-defined]
|
|
_EXE_DIR = Path(sys.executable).parent
|
|
sys.path.insert(0, str(_APP_DIR))
|
|
os.chdir(str(_APP_DIR))
|
|
_LOG_PATH = _EXE_DIR / "anonymisation_cli.log"
|
|
else:
|
|
_APP_DIR = Path(__file__).resolve().parent.parent
|
|
if str(_APP_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_APP_DIR))
|
|
_LOG_PATH = Path("anonymisation_cli.log")
|
|
|
|
# Extensions acceptées (le core convertit les formats non-PDF en amont via
|
|
# format_converter ; ici on collecte large et on laisse le moteur trancher).
|
|
_SUPPORTED_EXT = {
|
|
".pdf", ".docx", ".odt", ".rtf", ".txt", ".html", ".htm",
|
|
".jpg", ".jpeg", ".png", ".tif", ".tiff", ".bmp",
|
|
}
|
|
|
|
# Le CLI production n'utilise pas le manager ONNX legacy/Optimum
|
|
# (NerModelManager). Le désactiver évite un second chargement natif ONNX dans le
|
|
# même process Windows/PyInstaller avant CamemBERT-bio, qui est le modèle
|
|
# obligatoire du CLI.
|
|
os.environ.setdefault("ANON_SKIP_LEGACY_ONNX_MANAGER", "1")
|
|
|
|
_n_cpu_threads = str(os.cpu_count() or 4)
|
|
for _env in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS",
|
|
"NUMEXPR_NUM_THREADS", "VECLIB_MAXIMUM_THREADS"):
|
|
os.environ.setdefault(_env, _n_cpu_threads)
|
|
|
|
|
|
def _resolve(p: str) -> Path:
|
|
"""Résout un chemin relatif contre le cwd de lancement (pas _MEIPASS)."""
|
|
path = Path(p)
|
|
return path if path.is_absolute() else (_LAUNCH_CWD / path)
|
|
|
|
|
|
def _mandatory_model_path() -> Path:
|
|
"""Chemin attendu du modèle ONNX obligatoire (relatif à l'app/_MEIPASS)."""
|
|
return _APP_DIR / "models" / "camembert-bio-deid" / "onnx" / "model.onnx"
|
|
|
|
|
|
def _print_engines() -> int:
|
|
"""Affiche les moteurs réellement disponibles dans cet exécutable.
|
|
|
|
Diagnostic « honnête » : ne déclare jamais disponible un moteur dont les
|
|
dépendances (ou le modèle, pour CamemBERT) ne chargent pas. Sortie 0.
|
|
"""
|
|
from engine_capabilities import capabilities_map
|
|
|
|
caps = capabilities_map()
|
|
print("Moteurs d'anonymisation — disponibilité dans cet exécutable :")
|
|
for cap in caps.values():
|
|
mark = "OUI" if cap.available else "NON"
|
|
flag = "requis" if cap.required else "optionnel"
|
|
print(f" [{mark}] {cap.label} ({flag}) — {cap.reason}")
|
|
return 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(
|
|
prog="Anonymisation-CLI",
|
|
description="Anonymise un fichier (ou dossier) sans GUI.",
|
|
)
|
|
ap.add_argument("input", nargs="?", default=None, help="Fichier unique existant (ou dossier parcouru récursivement)")
|
|
ap.add_argument(
|
|
"output", nargs="?", default=None,
|
|
help="Dossier de sortie (forme positionnelle). Créé si absent.",
|
|
)
|
|
ap.add_argument("--out", default=None, help="Dossier de sortie (compatibilité). Équivaut à l'argument positionnel.")
|
|
ap.add_argument("--limit", type=int, default=0, help="Nombre max de documents (0 = tous ; utile pour un dossier)")
|
|
ap.add_argument("--no-ner", action="store_true", help="Mode regex seul : désactive EDS-Pseudo + CamemBERT (aucun modèle obligatoire)")
|
|
ap.add_argument("--gliner", action="store_true", help="Active aussi GLiNER (optionnel, vote croisé)")
|
|
ap.add_argument("--engines", action="store_true", help="Liste les moteurs réellement disponibles dans cet exécutable et quitte (diagnostic honnête, code 0)")
|
|
ap.add_argument("--config", default=None, help="Chemin config dictionnaires.yml (défaut: runtime)")
|
|
args = ap.parse_args(argv)
|
|
|
|
# --- Diagnostic moteurs : honnêteté sur ce que le build embarque réellement ---
|
|
if args.engines:
|
|
return _print_engines()
|
|
|
|
if args.input is None:
|
|
ap.error("argument 'input' requis (sauf avec --engines)")
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
handlers=[
|
|
logging.StreamHandler(sys.stdout),
|
|
logging.FileHandler(str(_LOG_PATH), encoding="utf-8"),
|
|
],
|
|
)
|
|
log = logging.getLogger("anon_cli")
|
|
log.info("CLI: démarrage frozen=%s python=%s log=%s",
|
|
bool(getattr(sys, "frozen", False)), sys.version.split()[0], _LOG_PATH)
|
|
|
|
# Dossier de sortie : positionnel prioritaire, puis --out, sinon défaut explicite.
|
|
out_arg = args.output or args.out or "anonymise_out"
|
|
out_root = _resolve(out_arg)
|
|
|
|
# --- Entrée : fichier unique (contrat principal) ou dossier ---
|
|
inp = _resolve(args.input)
|
|
if not inp.exists():
|
|
log.error("CLI: entrée introuvable: %s", inp)
|
|
return 2
|
|
if inp.is_file():
|
|
docs = [inp]
|
|
else:
|
|
docs = sorted(p for p in inp.rglob("*") if p.is_file() and p.suffix.lower() in _SUPPORTED_EXT)
|
|
if args.limit > 0:
|
|
docs = docs[: args.limit]
|
|
if not docs:
|
|
log.error("CLI: aucun document supporté trouvé sous %s", inp)
|
|
return 2
|
|
|
|
# --- Modèles ---
|
|
# OBLIGATOIRE (sauf --no-ner) : CamemBERT-bio ONNX. Fail-closed.
|
|
eds_mgr = camembert_mgr = gliner_mgr = None
|
|
if not args.no_ner:
|
|
model_path = _mandatory_model_path()
|
|
if not model_path.exists():
|
|
log.error("CLI: MODÈLE OBLIGATOIRE absent: %s. "
|
|
"Build incomplet ou ressource manquante. Abandon (pas de mode dégradé). "
|
|
"Utilisez --no-ner pour forcer le mode regex seul.", model_path)
|
|
return 3
|
|
try:
|
|
from camembert_ner_manager import CamembertNerManager
|
|
camembert_mgr = CamembertNerManager()
|
|
camembert_mgr.load()
|
|
log.info("CLI: CamemBERT-bio ONNX chargé (obligatoire) ✓")
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("CLI: échec chargement du modèle OBLIGATOIRE CamemBERT-bio ONNX: %s. "
|
|
"Abandon (pas de mode dégradé silencieux). "
|
|
"Utilisez --no-ner pour forcer le mode regex seul.", e)
|
|
return 3
|
|
|
|
# OPTIONNEL : EDS-Pseudo (best effort, tracé).
|
|
try:
|
|
from eds_pseudo_manager import EdsPseudoManager
|
|
eds_mgr = EdsPseudoManager()
|
|
eds_mgr.load()
|
|
log.info("CLI: EDS-Pseudo chargé (optionnel) ✓")
|
|
except Exception as e: # noqa: BLE001
|
|
eds_mgr = None
|
|
log.warning("CLI: EDS-Pseudo (optionnel) INDISPONIBLE: %s — traitement poursuivi sans.", e)
|
|
|
|
# OPTIONNEL : GLiNER (sur demande).
|
|
if args.gliner:
|
|
try:
|
|
from gliner_manager import GlinerManager
|
|
gliner_mgr = GlinerManager()
|
|
gliner_mgr.load()
|
|
log.info("CLI: GLiNER chargé (optionnel) ✓")
|
|
except Exception as e: # noqa: BLE001
|
|
gliner_mgr = None
|
|
log.warning("CLI: GLiNER (optionnel) INDISPONIBLE: %s — traitement poursuivi sans.", e)
|
|
else:
|
|
log.warning("CLI: --no-ner -> MODE REGEX SEUL assumé (aucun modèle NER). "
|
|
"Qualité réduite : à n'utiliser qu'en connaissance de cause.")
|
|
|
|
import anonymizer_core_refactored_onnx as core
|
|
|
|
# H1 : aligne les threads torch (idempotent).
|
|
if hasattr(core, "_configure_torch_threads"):
|
|
core._configure_torch_threads()
|
|
|
|
use_ner = bool(eds_mgr or gliner_mgr or camembert_mgr)
|
|
log.info("CLI: %d document(s), ner=%s (camembert=%s eds=%s gliner=%s) -> sortie=%s",
|
|
len(docs), use_ner, bool(camembert_mgr), bool(eds_mgr), bool(gliner_mgr), out_root)
|
|
|
|
config_path = _resolve(args.config) if args.config else None
|
|
process_fn = getattr(core, "process_document", None) or core.process_pdf
|
|
path_key = "doc_path" if process_fn.__name__ == "process_document" else "pdf_path"
|
|
out_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
t0 = time.perf_counter()
|
|
ok = ko = quarantined = no_output = 0
|
|
for i, doc in enumerate(docs, 1):
|
|
td = time.perf_counter()
|
|
try:
|
|
res = process_fn(
|
|
**{path_key: doc},
|
|
out_dir=out_root,
|
|
make_vector_redaction=False,
|
|
also_make_raster_burn=True,
|
|
config_path=config_path,
|
|
use_hf=use_ner,
|
|
ner_manager=eds_mgr,
|
|
gliner_manager=gliner_mgr,
|
|
camembert_manager=camembert_mgr,
|
|
ogc_label=None,
|
|
)
|
|
dt = time.perf_counter() - td
|
|
if isinstance(res, dict) and res.get("status") == "quarantined":
|
|
quarantined += 1
|
|
log.warning("CLI: [%d/%d] %s QUARANTAINE reason=%s %.2fs — sortie NON produite.",
|
|
i, len(docs), doc.name, res.get("reason", "?"), dt)
|
|
continue
|
|
# Vérifie qu'une sortie anonymisée a bien été produite.
|
|
produced = []
|
|
if isinstance(res, dict):
|
|
produced = [v for k, v in res.items() if k.startswith("pdf") and v and Path(v).exists()]
|
|
if not produced:
|
|
no_output += 1
|
|
log.error("CLI: [%d/%d] %s AUCUNE sortie PDF produite (%.2fs).", i, len(docs), doc.name, dt)
|
|
continue
|
|
ok += 1
|
|
log.info("CLI: [%d/%d] %s OK %.2fs -> %s", i, len(docs), doc.name, dt,
|
|
", ".join(Path(p).name for p in produced))
|
|
except Exception as e: # noqa: BLE001
|
|
ko += 1
|
|
log.error("CLI: [%d/%d] %s ERREUR: %s", i, len(docs), doc.name, e)
|
|
|
|
total = time.perf_counter() - t0
|
|
log.info("CLI: DONE total=%.1fs ok=%d quarantined=%d no_output=%d ko=%d",
|
|
total, ok, quarantined, no_output, ko)
|
|
|
|
# Codes retour : priorité aux échecs durs.
|
|
if ko:
|
|
return 1
|
|
if quarantined or no_output:
|
|
return 4
|
|
return 0 if ok else 4
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|