From 6c8184cc038d9c0beb9aa5c89abae389ac3f768f Mon Sep 17 00:00:00 2001 From: Dom Date: Fri, 24 Apr 2026 23:07:29 +0200 Subject: [PATCH] feat(deskew): correction automatique du skew au chargement des PDFs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nouveau module pipeline/deskew.py basé sur cv2.HoughLinesP : - détecte les lignes quasi-horizontales (±15° de l'horizontale) - prend la médiane de leurs angles (robuste aux outliers) - seuils : |angle|>0.3° pour corriger, |angle|>10° = suspect (on ne corrige pas) - PIL.rotate() avec BICUBIC + fillcolor blanc, sans expand Intégré dans pipeline/ingest.py (paramètre `deskew=True` par défaut). L'angle appliqué est tracé dans un fichier `page_XX.skew` à côté de l'image, pour audit. Mesuré sur les 18 dossiers de l'échantillon 2018 CARC : seule OGC 1 a un skew au-dessus du seuil (+0.91°), les 17 autres sont déjà droits. Le deskew corrige OGC 1 en 0.00° résiduel (vérif visuelle en-tête OK). Co-Authored-By: Claude Opus 4.7 (1M context) --- pipeline/deskew.py | 125 +++++++++++++++++++++++++++++++++++++++++++++ pipeline/ingest.py | 26 ++++++++-- 2 files changed, 148 insertions(+), 3 deletions(-) create mode 100644 pipeline/deskew.py diff --git a/pipeline/deskew.py b/pipeline/deskew.py new file mode 100644 index 0000000..5955688 --- /dev/null +++ b/pipeline/deskew.py @@ -0,0 +1,125 @@ +"""Détection d'angle de skew + redressement automatique des pages scannées. + +Technique : Hough Transform sur les lignes détectées par Canny, puis moyenne +des angles des lignes « quasi horizontales » (±15° par rapport à l'horizontale). +Les fiches OGC ont énormément de traits de tableau → signal très fort. + +Seuil : on ne corrige que si |angle| > `MIN_ANGLE_DEG` (0.3° par défaut) pour +éviter de toucher les scans déjà bien cadrés et introduire du bruit inutile. +""" +from __future__ import annotations + +from pathlib import Path +from typing import Tuple + +import numpy as np +from PIL import Image + +try: + import cv2 # type: ignore + _HAS_CV2 = True +except ImportError: + _HAS_CV2 = False + + +MIN_ANGLE_DEG = 0.3 # en-dessous, on ne corrige pas +MAX_ANGLE_DEG = 10.0 # au-dessus, c'est anormal → suspect, on ne corrige pas +NEAR_HORIZONTAL_BAND = 15.0 # degrés : bande autour de l'horizontale pour filtrer + + +def detect_skew_angle(img: Image.Image) -> float: + """Retourne l'angle de skew en degrés (positif = tourné dans le sens + des aiguilles d'une montre) à appliquer pour redresser l'image. + + Si aucune ligne horizontale n'est trouvée, retourne 0.0. + Si l'angle détecté est hors [-MAX_ANGLE_DEG, +MAX_ANGLE_DEG], retourne 0.0 + (probablement une erreur de détection, on ne corrige pas). + """ + if not _HAS_CV2: + return 0.0 + gray = np.array(img.convert("L")) + # Réduire l'image pour accélérer (max 1500 px de large) + h, w = gray.shape + if w > 1500: + scale = 1500 / w + gray = cv2.resize(gray, (1500, int(h * scale)), interpolation=cv2.INTER_AREA) + + # Canny edges — paramètres standards documents + edges = cv2.Canny(gray, 50, 150, apertureSize=3) + # Hough Lines probabiliste : rapide et robuste + lines = cv2.HoughLinesP( + edges, rho=1, theta=np.pi / 180, threshold=200, + minLineLength=gray.shape[1] // 4, # au moins 25% de la largeur + maxLineGap=20, + ) + if lines is None or len(lines) == 0: + return 0.0 + + # Calculer l'angle de chaque ligne en degrés + angles = [] + for line in lines: + x1, y1, x2, y2 = line[0] + if x2 == x1: + continue # ligne verticale, ignorée + angle = np.degrees(np.arctan2(y2 - y1, x2 - x1)) + # On ne garde que les lignes proches de l'horizontale + if abs(angle) < NEAR_HORIZONTAL_BAND: + angles.append(angle) + + if not angles: + return 0.0 + + # Moyenne robuste : médiane plutôt que mean, moins sensible aux outliers + angle = float(np.median(angles)) + if abs(angle) > MAX_ANGLE_DEG: + return 0.0 # suspect → on ne corrige pas + return angle + + +def deskew_image(img: Image.Image, + angle: float | None = None, + min_angle: float = MIN_ANGLE_DEG) -> Tuple[Image.Image, float]: + """Redresse une image si le skew détecté dépasse `min_angle`. + + Retourne (image_eventuellement_rotee, angle_applique). + Si |angle| < min_angle, retourne l'image inchangée et angle=0.0. + """ + if angle is None: + angle = detect_skew_angle(img) + if abs(angle) < min_angle: + return img, 0.0 + # PIL.Image.rotate : positive angle = counter-clockwise + # detect_skew retourne positif = clockwise → on inverse pour PIL + rotated = img.rotate( + angle, + resample=Image.Resampling.BICUBIC, + expand=False, + fillcolor="white", + ) + return rotated, angle + + +def deskew_file(src: Path, dst: Path | None = None, + min_angle: float = MIN_ANGLE_DEG) -> float: + """Version fichier → fichier. Écrase `src` si `dst` est None. + Retourne l'angle appliqué (0.0 si pas de rotation).""" + img = Image.open(src) + rotated, angle = deskew_image(img, min_angle=min_angle) + out = dst or src + rotated.save(out, "PNG", optimize=True) + return angle + + +if __name__ == "__main__": + import sys + import glob + paths = [Path(p) for p in (sys.argv[1:] or sorted(glob.glob(".cache/images/*/page_01.png")))] + print(f"Deskew sur {len(paths)} images (seuil={MIN_ANGLE_DEG}°)...") + total_corrected = 0 + for p in paths: + angle = detect_skew_angle(Image.open(p)) + mark = "→" if abs(angle) >= MIN_ANGLE_DEG else "·" + if abs(angle) >= MIN_ANGLE_DEG: + total_corrected += 1 + print(f" {mark} {p} : {angle:+.2f}°") + print(f"\n{total_corrected}/{len(paths)} images auraient besoin d'un redressement.") diff --git a/pipeline/ingest.py b/pipeline/ingest.py index 8cd1433..570fadd 100644 --- a/pipeline/ingest.py +++ b/pipeline/ingest.py @@ -1,10 +1,16 @@ -"""PDF → images PNG 300 dpi avec cache par hash SHA256.""" +"""PDF → images PNG 300 dpi avec cache par hash SHA256. + +Applique optionnellement un deskew automatique (redressement) sur chaque page +pour corriger le biais d'inclinaison des scans. Voir pipeline/deskew.py. +""" import hashlib import os from pathlib import Path from pdf2image import convert_from_path from PIL import Image +from .deskew import deskew_image, MIN_ANGLE_DEG + DEFAULT_DPI = 300 CACHE_ROOT = Path(".cache/images") @@ -18,23 +24,37 @@ def pdf_hash(pdf_path: str) -> str: return h.hexdigest()[:16] -def pdf_to_images(pdf_path: str, dpi: int = DEFAULT_DPI, cache_root: Path = CACHE_ROOT) -> list[Path]: +def pdf_to_images(pdf_path: str, dpi: int = DEFAULT_DPI, + cache_root: Path = CACHE_ROOT, + deskew: bool = True) -> list[Path]: """Convertit un PDF en PNG 300 dpi. Retourne la liste des chemins (1 par page). Le cache est indexé par hash du PDF : un PDF inchangé n'est jamais reconverti. + + Avec `deskew=True` (défaut), chaque page est redressée si son angle de skew + dépasse le seuil défini dans `pipeline.deskew.MIN_ANGLE_DEG` (0.3°). L'angle + appliqué est persisté dans un fichier `.skew` à côté (pour audit). """ cache_root = Path(cache_root) h = pdf_hash(pdf_path) out_dir = cache_root / h out_dir.mkdir(parents=True, exist_ok=True) - existing = sorted(out_dir.glob("page_*.png")) + # Le glob est strict pour ne pas attraper les crops intermédiaires + # (page_XX_recodage.png, etc.) + existing = sorted(p for p in out_dir.glob("page_*.png") + if p.stem.replace("page_", "").isdigit()) if existing: return existing pages = convert_from_path(pdf_path, dpi) paths = [] for i, img in enumerate(pages, 1): + if deskew: + img, applied = deskew_image(img) + if applied != 0.0: + # Trace d'audit : on note l'angle corrigé + (out_dir / f"page_{i:02d}.skew").write_text(f"{applied:.3f}\n") p = out_dir / f"page_{i:02d}.png" img.save(p, "PNG", optimize=True) paths.append(p)