"""Cycle de vie des imports : quarantaine, analyse isolée, propositions, validation humaine, annulation.""" from __future__ import annotations import hashlib import json import logging import os import resource import secrets import shutil import socket import struct import subprocess import sys import uuid from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path from typing import Any from sqlalchemy import func, select, update from sqlalchemy.orm import Session from ..core import audit, isoweek from ..core.config import get_config from ..models import ( STATUS_LABELS, Bilan, ImportApplication, ImportFile, ImportProposal, ImportSource, Indicator, Job, User, ) from ..services import bilans as bilan_service from ..services import settings as settings_service from . import detect log = logging.getLogger("edcf.imports") SANDBOX_TIMEOUT = 240 SANDBOX_MEMORY = 900 * 1024 * 1024 OCR_FORMATS = {"png", "jpeg", "pdf"} class ImportError_(Exception): def __init__(self, status: int, message: str, code: str = "import_error", extra: dict[str, Any] | None = None): super().__init__(message) self.status, self.message, self.code, self.extra = status, message, code, extra or {} def utcnow() -> datetime: return datetime.now(timezone.utc) def import_dir(import_id: uuid.UUID) -> Path: return get_config().imports_dir / str(import_id) # --- Antivirus (clamd, protocole INSTREAM) ------------------------------------------------------- def antivirus_scan(data: bytes) -> tuple[str, str | None]: cfg = get_config() if cfg.av_mode == "disabled": return "desactive", None if not cfg.clamav_socket: return "non_execute", "Antivirus ClamAV non configuré sur ce serveur." try: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: s.settimeout(60) s.connect(cfg.clamav_socket) s.sendall(b"zINSTREAM\0") for i in range(0, len(data), 65536): chunk = data[i:i + 65536] s.sendall(struct.pack("!L", len(chunk)) + chunk) s.sendall(struct.pack("!L", 0)) reply = s.recv(4096).decode(errors="replace").strip("\0\n ") except OSError as e: return "erreur", f"Antivirus injoignable ({e.__class__.__name__})." if reply.endswith("OK"): return "sain", None if "FOUND" in reply: return "infecte", reply.split(":", 1)[-1].replace("FOUND", "").strip() return "erreur", "Réponse inattendue de l'antivirus." # --- Réception ---------------------------------------------------------------------------------- def create_import(db: Session, user: User, filename: str | None, data: bytes, declared_mime: str | None, ip: str | None) -> ImportFile: max_mb = int(settings_service.get(db, "import_max_mb")) display_name = detect.safe_display_name(filename) if len(data) > max_mb * 1024 * 1024: raise ImportError_(413, f"Fichier trop volumineux ({len(data) / 1048576:.1f} Mo). Maximum : {max_mb} Mo.", "too_large") quota = int(settings_service.get(db, "import_hourly_quota")) recent = db.execute(select(func.count(ImportFile.id)).where( ImportFile.uploaded_by == user.id, ImportFile.uploaded_at > utcnow() - timedelta(hours=1))).scalar() or 0 if recent >= quota: raise ImportError_(429, "Trop d'imports en une heure : réessayez plus tard.", "quota") digest = hashlib.sha256(data).hexdigest() try: found = detect.sniff(data, display_name) except detect.FileRejected as e: audit.record(db, "import.rejected", user_id=user.id, username=user.username, target_type="import", ip=ip, details={"fichier": display_name, "taille": len(data), "sha256": digest, "motif": str(e), "mime_annonce": (declared_mime or "")[:100]}) db.commit() raise ImportError_(422, str(e), "rejected") from None av_status, av_detail = antivirus_scan(data) cfg = get_config() if av_status == "infecte": audit.record(db, "import.infected", user_id=user.id, username=user.username, target_type="import", ip=ip, details={"fichier": display_name, "sha256": digest, "signature": av_detail}) db.commit() raise ImportError_(422, "Fichier refusé : l'antivirus a détecté une menace.", "infected") if av_status in {"non_execute", "erreur"} and cfg.av_mode == "required": raise ImportError_(503, "Import impossible : l'analyse antivirus obligatoire n'est pas disponible.", "av_unavailable") warnings = list(found.warnings) if av_detail and av_status != "sain": warnings.append(av_detail) previous = db.execute(select(ImportFile).where(ImportFile.sha256 == digest, ImportFile.status != "rejete") .order_by(ImportFile.uploaded_at.desc()).limit(1)).scalar_one_or_none() if previous is not None: warnings.append(f"Doublon potentiel : ce fichier a déjà été importé le " f"{isoweek.fr_datetime(previous.uploaded_at)} (statut : {previous.status}).") imp = ImportFile(id=uuid.uuid4(), original_name=display_name, stored_name=secrets.token_hex(16), size=len(data), sha256=digest, detected_format=found.format, detected_mime=found.mime, declared_mime=(declared_mime or "")[:100] or None, status="en_attente", av_status=av_status, analysis={"warnings": warnings, "checks": found.info}, uploaded_by=user.id, duplicate_of=previous.id if previous else None) d = import_dir(imp.id) d.mkdir(parents=True, mode=0o700) target = d / imp.stored_name fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "wb") as f: f.write(data) db.add(imp) db.flush() db.add(Job(kind="analyse_import", payload={"import_id": str(imp.id)}, status="en_attente")) audit.record(db, "import.uploaded", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, ip=ip, details={"fichier": display_name, "format": found.format, "taille": len(data), "sha256": digest, "antivirus": av_status, "doublon_de": str(previous.id) if previous else None}) db.commit() return imp # --- Analyse (worker) --------------------------------------------------------------------------- def _limits() -> None: # exécuté dans le processus enfant, avant exec resource.setrlimit(resource.RLIMIT_AS, (SANDBOX_MEMORY, SANDBOX_MEMORY)) resource.setrlimit(resource.RLIMIT_CPU, (180, 200)) resource.setrlimit(resource.RLIMIT_FSIZE, (200 * 1024 * 1024, 200 * 1024 * 1024)) # Pas de RLIMIT_NPROC : il compte tous les processus de l'utilisateur (TasksMax systemd s'en charge). resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) os.setsid() def run_sandbox(imp: ImportFile, indicators: list[Indicator], options: dict[str, Any]) -> dict[str, Any]: d = import_dir(imp.id) work = d / "analyse" if work.exists(): shutil.rmtree(work) work.mkdir(mode=0o700) payload = { "indicators": [{"id": i.id, "code": i.code, "label": i.label, "aliases": list(i.aliases or []), "detail_kind": i.detail_kind} for i in indicators], "options": {**options, "original_name": imp.original_name, "tesseract": get_config().tesseract_bin}, } env = {"PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": str(work), "LANG": "C.UTF-8", "OMP_THREAD_LIMIT": "1"} root = str(Path(__file__).resolve().parents[2]) # -I : mode isolé (ni variables PYTHON*, ni site utilisateur) ; seul le code de l'application est chargé. boot = f"import sys; sys.path.insert(0, {root!r}); from edcf.imports.sandbox import main; main()" proc = subprocess.run( [sys.executable, "-I", "-B", "-c", boot, str(d / imp.stored_name), imp.detected_format, str(work)], input=json.dumps(payload).encode(), capture_output=True, timeout=SANDBOX_TIMEOUT, env=env, cwd=str(work), preexec_fn=_limits, check=False, ) if proc.stderr: log.warning("Analyse %s : %s", imp.id, proc.stderr.decode(errors="replace")[-500:]) try: out = json.loads(proc.stdout.decode("utf-8")) except (ValueError, UnicodeDecodeError): raise RuntimeError("L'analyse a été interrompue (ressources dépassées ou fichier malformé).") from None if not out.get("ok"): raise RuntimeError(out.get("error") or "Analyse impossible.") return out["result"] def process_import(db: Session, import_id: uuid.UUID, options: dict[str, Any] | None = None) -> None: imp = db.get(ImportFile, import_id, with_for_update=True) if imp is None or imp.status in {"annule", "rejete", "applique"}: return imp.status = "en_cours" db.commit() indicators = list(db.execute(select(Indicator).order_by(Indicator.position)).scalars()) by_code = {i.code: i for i in indicators} try: result = run_sandbox(imp, indicators, options or {}) except (RuntimeError, subprocess.TimeoutExpired) as e: msg = str(e) if isinstance(e, RuntimeError) else "Analyse trop longue : délai maximum dépassé." db.refresh(imp) imp.status, imp.error_message, imp.finished_at = "echec", msg, utcnow() audit.record(db, "import.failed", user_id=imp.uploaded_by, target_type="import", target_id=imp.id, details={"motif": msg}) db.commit() return db.refresh(imp) if imp.status == "annule": # annulé pendant l'analyse : résultat abandonné return db.query(ImportProposal).filter(ImportProposal.import_id == imp.id).delete() db.query(ImportSource).filter(ImportSource.import_id == imp.id).delete() db.flush() src_ids: dict[int, int] = {} for s in result["sources"]: image = s.get("image") if image: produced = import_dir(imp.id) / "analyse" / image if produced.is_file(): final = import_dir(imp.id) / f"view-{s['index']}.png" produced.replace(final) os.chmod(final, 0o600) src = ImportSource(import_id=imp.id, kind=s["kind"], index=s["index"], name=str(s["name"])[:160], hidden=bool(s.get("hidden")), width=s.get("width"), height=s.get("height"), image_name=f"view-{s['index']}.png" if image else None, grid=s.get("grid"), meta=s.get("meta") or {}) db.add(src) db.flush() src_ids[s["index"]] = src.id shutil.rmtree(import_dir(imp.id) / "analyse", ignore_errors=True) confs = [] for p in result["proposals"]: ind = by_code.get(p.get("indicator_code") or "") if ind is not None and not ind.is_active: p["warnings"].append("Cet indicateur est désactivé : il ne peut pas être importé.") p["decision"] = "ignore" db.add(ImportProposal( import_id=imp.id, source_id=src_ids.get(p["source_index"]), row_index=p["row_index"], raw_label=str(p.get("raw_label") or "")[:2000], indicator_id=ind.id if ind else None, match_score=p.get("match_score") or 0, raw_value=str(p.get("raw_value") or "")[:200], value=p.get("value"), value_error=p.get("value_error"), observation=bilan_service.clean_text(p.get("observation")), details=p.get("details") or [], label_ref=p.get("label_ref"), value_ref=p.get("value_ref"), observation_ref=p.get("observation_ref"), bbox=p.get("bbox"), value_bbox=p.get("value_bbox"), confidence=p.get("confidence") or 0, warnings=p.get("warnings") or [], decision=p.get("decision") or "a_verifier")) if ind is not None and p["source_index"] == result.get("selected_source"): confs.append(p.get("confidence") or 0) selected = [p for p in result["proposals"] if p["source_index"] == result.get("selected_source")] analysis = dict(imp.analysis or {}) analysis.update({ "selected_source": result.get("selected_source", 0), "week": result.get("week"), "lines": len(selected), "recognized": sum(1 for p in selected if p.get("indicator_code")), "values": sum(1 for p in selected if p.get("value") is not None), "errors": sum(1 for p in selected if p.get("value_error")), "duplicates": sum(1 for p in selected if p.get("duplicate")), "mean_confidence": round(sum(confs) / len(confs), 3) if confs else 0, "unused_lines": result.get("unused_lines", [])[:50], "warnings": list(dict.fromkeys((analysis.get("warnings") or []) + result.get("warnings", []))), "options": options or {}, }) week = result.get("week") or {} imp.analysis = analysis imp.target_iso_year, imp.target_iso_week = week.get("iso_year"), week.get("iso_week") imp.status, imp.error_message, imp.finished_at = "analyse", None, utcnow() audit.record(db, "import.analysed", user_id=imp.uploaded_by, target_type="import", target_id=imp.id, details={k: analysis[k] for k in ("lines", "recognized", "values", "errors", "mean_confidence")}) db.commit() def run_next_job(db: Session) -> bool: """Prend une tâche en attente (verrou SKIP LOCKED) ; renvoie False s'il n'y en a pas.""" job = db.execute(select(Job).where(Job.status == "en_attente").order_by(Job.id) .with_for_update(skip_locked=True).limit(1)).scalar_one_or_none() if job is None: return False job.status, job.started_at, job.attempts, job.progress = "en_cours", utcnow(), job.attempts + 1, 10 db.commit() try: if job.kind == "analyse_import": process_import(db, uuid.UUID(job.payload["import_id"]), job.payload.get("options")) job.status, job.progress = "termine", 100 except Exception as e: # la tâche échoue proprement, le worker continue log.exception("Tâche %s en échec", job.id) db.rollback() job = db.get(Job, job.id) assert job is not None job.status, job.message = "echec", type(e).__name__ imp = db.get(ImportFile, uuid.UUID(job.payload["import_id"])) if job.payload.get("import_id") else None if imp is not None and imp.status in {"en_attente", "en_cours"}: imp.status, imp.error_message = "echec", "Erreur interne pendant l'analyse." job.finished_at = utcnow() db.commit() return True def requeue_stale(db: Session) -> None: stale = utcnow() - timedelta(minutes=10) db.execute(update(Job).where(Job.status == "en_cours", Job.started_at < stale, Job.attempts < 3) .values(status="en_attente")) db.execute(update(Job).where(Job.status == "en_cours", Job.started_at < stale, Job.attempts >= 3) .values(status="echec", message="Abandon après 3 tentatives")) db.commit() def reanalyse(db: Session, imp: ImportFile, user: User, rotation: int | None, contrast: bool, ip: str | None) -> None: if imp.status not in {"analyse", "echec"}: raise ImportError_(409, "Cet import ne peut plus être réanalysé.", "bad_state") if imp.detected_format not in OCR_FORMATS: raise ImportError_(422, "La rotation et le contraste ne concernent que les images et les PDF.", "bad_format") imp.status = "en_attente" db.add(Job(kind="analyse_import", payload={"import_id": str(imp.id), "options": {"rotation": rotation, "contrast": contrast}})) audit.record(db, "import.reanalysed", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, ip=ip, details={"rotation": rotation, "contraste": contrast}) db.commit() # --- Lecture ------------------------------------------------------------------------------------ def proposal_dict(p: ImportProposal) -> dict[str, Any]: return {"id": p.id, "source_id": p.source_id, "row_index": p.row_index, "raw_label": p.raw_label, "indicator_id": p.indicator_id, "match_score": p.match_score, "raw_value": p.raw_value, "value": p.value, "value_error": p.value_error, "observation": p.observation, "details": p.details, "label_ref": p.label_ref, "value_ref": p.value_ref, "observation_ref": p.observation_ref, "bbox": p.bbox, "value_bbox": p.value_bbox, "confidence": p.confidence, "warnings": p.warnings, "decision": p.decision, "edited": p.edited} def import_summary(db: Session, imp: ImportFile, names: dict[int, str] | None = None) -> dict[str, Any]: if names is None: u = db.get(User, imp.uploaded_by) names = {imp.uploaded_by: u.display_name if u else "?"} a = imp.analysis or {} job = db.execute(select(Job).where(Job.payload["import_id"].astext == str(imp.id)) .order_by(Job.id.desc()).limit(1)).scalar_one_or_none() app = db.execute(select(ImportApplication).where(ImportApplication.import_id == imp.id) .order_by(ImportApplication.id.desc()).limit(1)).scalar_one_or_none() bilan = db.get(Bilan, app.bilan_id) if app else None return { "id": str(imp.id), "original_name": imp.original_name, "stored_name": imp.stored_name, "size": imp.size, "sha256": imp.sha256, "format": imp.detected_format, "mime": imp.detected_mime, "declared_mime": imp.declared_mime, "status": imp.status, "av_status": imp.av_status, "error": imp.error_message, "uploaded_at": imp.uploaded_at.isoformat(), "uploaded_by": names.get(imp.uploaded_by), "finished_at": imp.finished_at.isoformat() if imp.finished_at else None, "duplicate_of": str(imp.duplicate_of) if imp.duplicate_of else None, "target": {"iso_year": imp.target_iso_year, "iso_week": imp.target_iso_week} if imp.target_iso_week else None, "week_detection": a.get("week"), "lines": a.get("lines", 0), "recognized": a.get("recognized", 0), "values": a.get("values", 0), "errors": a.get("errors", 0), "duplicates": a.get("duplicates", 0), "mean_confidence": a.get("mean_confidence"), "warnings": a.get("warnings", []), "unused_lines": a.get("unused_lines", []), "selected_source": a.get("selected_source", 0), "options": a.get("options", {}), "purged": imp.purged_at is not None, "job": {"status": job.status, "progress": job.progress} if job else None, "applied": None if app is None else { "bilan": f"{bilan.iso_year}-S{bilan.iso_week:02d}" if bilan else None, "iso_year": bilan.iso_year if bilan else None, "iso_week": bilan.iso_week if bilan else None, "applied_at": app.applied_at.isoformat(), "reverted_at": app.reverted_at.isoformat() if app.reverted_at else None, "report": app.report}, } def import_detail(db: Session, imp: ImportFile) -> dict[str, Any]: sources = list(db.execute(select(ImportSource).where(ImportSource.import_id == imp.id) .order_by(ImportSource.index)).scalars()) proposals = list(db.execute(select(ImportProposal).where(ImportProposal.import_id == imp.id) .order_by(ImportProposal.source_id, ImportProposal.row_index)).scalars()) return {**import_summary(db, imp), "sources": [{"id": s.id, "kind": s.kind, "index": s.index, "name": s.name, "hidden": s.hidden, "width": s.width, "height": s.height, "has_image": bool(s.image_name) and imp.purged_at is None, "has_grid": s.grid is not None, "meta": s.meta} for s in sources], "proposals": [proposal_dict(p) for p in proposals]} # --- Corrections humaines ------------------------------------------------------------------------ def update_proposal(db: Session, imp: ImportFile, pid: int, changes: dict[str, Any], user: User, ip: str | None) -> ImportProposal: if imp.status != "analyse": raise ImportError_(409, "Cet import n'est plus modifiable.", "bad_state") p = db.get(ImportProposal, pid) if p is None or p.import_id != imp.id: raise ImportError_(404, "Ligne introuvable.", "not_found") before = proposal_dict(p) if "indicator_id" in changes: ind_id = changes["indicator_id"] if ind_id is not None: ind = db.get(Indicator, ind_id) if ind is None or not ind.is_active: raise ImportError_(422, "Indicateur inconnu ou désactivé.", "bad_indicator") p.indicator_id = ind_id if "value" in changes: value, err = bilan_service.parse_count(changes["value"]) if err: raise ImportError_(422, err, "bad_value") p.value, p.value_error = value, None if "observation" in changes: p.observation = bilan_service.clean_text(changes["observation"]) if "details" in changes: try: p.details = [format(bilan_service.parse_detail(d).normalize(), "f") for d in changes["details"] or [] if str(d).strip()] except ValueError as e: raise ImportError_(422, str(e), "bad_detail") from None if "decision" in changes: if changes["decision"] not in {"accepte", "ignore", "a_verifier"}: raise ImportError_(422, "Décision invalide.", "bad_decision") if changes["decision"] == "accepte": if p.indicator_id is None: raise ImportError_(422, "Choisissez l'indicateur avant d'accepter la ligne.", "no_indicator") if p.value_error: raise ImportError_(422, "Corrigez la valeur avant d'accepter la ligne.", "value_error") p.decision = changes["decision"] p.edited, p.updated_by, p.updated_at = True, user.id, utcnow() after = proposal_dict(p) diff = {k: {"avant": before[k], "apres": after[k]} for k in ("indicator_id", "value", "observation", "details", "decision") if before[k] != after[k]} if diff: audit.record(db, "import.proposal_corrected", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, ip=ip, details={"ligne": p.row_index + 1, **diff}) db.commit() return p def select_source(db: Session, imp: ImportFile, source_id: int, user: User) -> None: if imp.status != "analyse": raise ImportError_(409, "Cet import n'est plus modifiable.", "bad_state") src = db.get(ImportSource, source_id) if src is None or src.import_id != imp.id: raise ImportError_(404, "Feuille ou page introuvable.", "not_found") ocr = imp.detected_format in OCR_FORMATS for p in db.execute(select(ImportProposal).where(ImportProposal.import_id == imp.id)).scalars(): if p.source_id != src.id: p.decision = "ignore" elif any("total" in w.lower() for w in p.warnings or []) and p.indicator_id is None: p.decision = "ignore" else: sure = (p.indicator_id is not None and p.confidence >= 0.9 and not p.value_error and not p.warnings and not ocr) p.decision = "accepte" if sure else "a_verifier" a = dict(imp.analysis or {}) a["selected_source"] = src.index imp.analysis = a audit.record(db, "import.source_selected", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, details={"source": src.name}) db.commit() def bulk_decide(db: Session, imp: ImportFile, decision: str, min_confidence: float, user: User) -> int: if imp.status != "analyse": raise ImportError_(409, "Cet import n'est plus modifiable.", "bad_state") selected = (imp.analysis or {}).get("selected_source", 0) src = db.execute(select(ImportSource).where(ImportSource.import_id == imp.id, ImportSource.index == selected) ).scalar_one_or_none() n = 0 for p in db.execute(select(ImportProposal).where(ImportProposal.import_id == imp.id)).scalars(): if (src is not None and p.source_id != src.id) or p.decision != "a_verifier": continue if decision == "accepte" and (p.indicator_id is None or p.value_error or p.confidence < min_confidence): continue p.decision, p.updated_by, p.updated_at = decision, user.id, utcnow() n += 1 audit.record(db, "import.bulk_decision", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, details={"decision": decision, "confiance_min": min_confidence, "lignes": n}) db.commit() return n # --- Validation humaine et annulation ------------------------------------------------------------ def _values_snapshot(db: Session, bilan: Bilan) -> dict[str, Any]: payload = bilan_service.bilan_payload(db, bilan.iso_year, bilan.iso_week, bilan_service.find_bilan( db, bilan.iso_year, bilan.iso_week)) return {"status": payload["status"], "revision": payload["revision"], "rows": {str(r["indicator_id"]): {"label": r["label"], "value": r["value"], "observation": r["observation"], "details": [d["value"] for d in r["details"]]} for r in payload["rows"]}} def apply_import(db: Session, imp: ImportFile, iso_year: int, iso_week: int, user: User, ip: str | None ) -> dict[str, Any]: if imp.status != "analyse": raise ImportError_(409, "Cet import n'est pas prêt à être appliqué.", "bad_state") try: isoweek.validate_week(iso_year, iso_week) except ValueError as e: raise ImportError_(422, str(e), "bad_week") from None proposals = list(db.execute(select(ImportProposal).where(ImportProposal.import_id == imp.id)).scalars()) pending = [p for p in proposals if p.decision == "a_verifier"] if pending: raise ImportError_(422, f"{len(pending)} ligne(s) restent à vérifier : acceptez-les ou ignorez-les.", "pending", {"pending": [p.id for p in pending]}) accepted = [p for p in proposals if p.decision == "accepte"] if not accepted: raise ImportError_(422, "Aucune ligne acceptée : rien à importer.", "nothing") targets: dict[int, ImportProposal] = {} for p in accepted: if p.indicator_id is None or p.value_error: raise ImportError_(422, "Une ligne acceptée n'a pas d'indicateur ou contient une erreur.", "invalid") if p.indicator_id in targets: raise ImportError_(422, "Deux lignes acceptées visent le même indicateur : ignorez le doublon.", "duplicate_target") targets[p.indicator_id] = p sources = {s.id: s for s in db.execute(select(ImportSource).where(ImportSource.import_id == imp.id)).scalars()} bilan = bilan_service.get_or_create_for_update(db, iso_year, iso_week, user.id) if bilan.status not in bilan_service.EDITABLE: raise ImportError_(423, f"Le bilan de la semaine {iso_week} est {STATUS_LABELS[bilan.status].lower()} : " "rouvrez-le avant d'y importer des données.", "locked") before = _values_snapshot(db, bilan) indicators = {i.id: i for i in bilan_service.get_indicators(db)} rows, provenance = [], {} for ind_id, p in targets.items(): details = None if indicators[ind_id].detail_kind: details = [Decimal(str(d)) for d in p.details or []] rows.append(bilan_service.RowInput(ind_id, p.value, p.observation, details)) src = sources.get(p.source_id or 0) locator = None if src is not None: locator = f"Feuille « {src.name} »" if src.kind == "feuille" else src.name provenance[ind_id] = {"locator": locator, "ref": p.value_ref or p.label_ref, "bbox": {"source_id": p.source_id, **(p.value_bbox or p.bbox or {})} if (p.bbox or p.value_bbox) else None, "confidence": p.confidence} changed = bilan_service.apply_rows(db, bilan, rows, user.id, source="import", import_id=imp.id, provenance=provenance, reason=f"Import « {imp.original_name} »") if bilan.status == "brouillon": bilan.status = "a_verifier" bilan.revision += 1 bilan.updated_by, bilan.updated_at = user.id, utcnow() db.flush() after = _values_snapshot(db, bilan) report = { "fichier": imp.original_name, "sha256": imp.sha256, "semaine": f"{iso_year}-S{iso_week:02d}", "lignes_acceptees": len(accepted), "lignes_ignorees": sum(1 for p in proposals if p.decision == "ignore"), "lignes_modifiees": changed, "corrections_manuelles": sum(1 for p in accepted if p.edited), "semaine_detectee": (imp.analysis or {}).get("week"), "avant_apres": [{"indicateur": indicators[i].label, "avant": before["rows"].get(str(i), {}).get("value"), "apres": after["rows"].get(str(i), {}).get("value")} for i in targets], } db.add(ImportApplication(import_id=imp.id, bilan_id=bilan.id, before_snapshot=before, after_snapshot=after, report=report, applied_by=user.id)) imp.status = "applique" audit.record(db, "import.applied", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, ip=ip, details={k: report[k] for k in ("semaine", "lignes_acceptees", "lignes_ignorees", "lignes_modifiees", "corrections_manuelles")}) db.commit() return report def cancel_import(db: Session, imp: ImportFile, user: User, ip: str | None) -> dict[str, Any]: """Annulation complète : avant application = abandon ; après = retour à l'état « avant » du bilan.""" if imp.status in {"annule", "rejete"}: raise ImportError_(409, "Cet import est déjà annulé.", "bad_state") result: dict[str, Any] = {"reverted": False} if imp.status == "applique": app = db.execute(select(ImportApplication).where(ImportApplication.import_id == imp.id, ImportApplication.reverted_at.is_(None))).scalar_one() bilan = db.get(Bilan, app.bilan_id, with_for_update=True) if bilan is None: raise ImportError_(409, "Le bilan concerné n'existe plus.", "gone") if bilan.status not in bilan_service.EDITABLE: raise ImportError_(423, "Le bilan a été validé depuis l'import : rouvrez-le pour annuler l'import.", "locked") current = _values_snapshot(db, bilan) modified_since = [k for k, v in app.after_snapshot["rows"].items() if current["rows"].get(k) != v] rows = [] touched = [int(k) for k, v in app.after_snapshot["rows"].items() if app.before_snapshot["rows"].get(k) != v] for ind_id in touched: b = app.before_snapshot["rows"].get(str(ind_id), {"value": None, "observation": "", "details": []}) rows.append(bilan_service.RowInput(ind_id, b.get("value"), b.get("observation") or "", [Decimal(str(d)) for d in b.get("details") or []])) bilan_service.apply_rows(db, bilan, rows, user.id, source="restauration", reason=f"Annulation de l'import « {imp.original_name} »") bilan.revision += 1 bilan.updated_by, bilan.updated_at = user.id, utcnow() app.reverted_by, app.reverted_at = user.id, utcnow() result = {"reverted": True, "rows": len(touched), "modified_since": modified_since, "bilan": f"{bilan.iso_year}-S{bilan.iso_week:02d}"} imp.status = "annule" audit.record(db, "import.cancelled", user_id=user.id, username=user.username, target_type="import", target_id=imp.id, ip=ip, details=result) db.commit() return result def purge_expired(db: Session) -> int: days = int(settings_service.get(db, "import_retention_days")) limit = utcnow() - timedelta(days=days) n = 0 for imp in db.execute(select(ImportFile).where(ImportFile.uploaded_at < limit, ImportFile.purged_at.is_(None))).scalars(): shutil.rmtree(import_dir(imp.id), ignore_errors=True) imp.purged_at = utcnow() for s in db.execute(select(ImportSource).where(ImportSource.import_id == imp.id)).scalars(): s.image_name, s.grid = None, None n += 1 if n: audit.record(db, "import.purged", username="systeme", details={"fichiers": n, "retention_jours": days}) return n