Il y a 389 articles et 19 utilisateurs inscrits (ce bandeau est défini dans MediaWiki:Sitenotice)

Générateur METS local pour presse ad44

De AJLSM
Aller à la navigation Aller à la recherche

Générateur METS / ALTO / EAD pour Pleade

[modifier | modifier le wikicode]

Documentation fusionnée (version actuelle basée sur la v6.x du script)

Cette documentation fusionne :

  • la documentation originale v5.2
  • les ajouts récents concernant :
    • audit des fichiers ALTO
    • redimensionnement des coordonnées ALTO
    • ajout automatique des ID ALTO
    • modes d'exécution avancés

Description générale

[modifier | modifier le wikicode]

Le script make_mets_package.py prépare automatiquement des lots de numérisation pour publication dans **Pleade**.

Il réalise :

  • la génération des fichiers **METS**
  • la génération optionnelle d’un **EAD maître**
  • la vérification et correction des fichiers **ALTO**
  • l'analyse technique des fichiers ALTO (audit)
  • la structuration des répertoires **alto / images / mets**

Le script fonctionne **sans dépendance externe** (pas de Pillow).

Architecture des lots d'entrée

[modifier | modifier le wikicode]

Structure attendue : 

collection/
 ├── collection.csv
 └── DOCID/
      └── fascicule/
           ├── alto/
           └── jpg/

Chaque fascicule contient :

  • images JPEG
  • fichiers ALTO XML
  • métadonnées issues du CSV.

Architecture des données générées

[modifier | modifier le wikicode]

Sortie : 

output/
 ├── alto/
 │   └── DOCID/...
 ├── images/
 │   └── DOCID/...
 └── mets/
     └── DOCID/
         └── mets-OBJID.xml

Génération des METS

[modifier | modifier le wikicode]

Pour chaque fascicule :

  • création d’un fichier
mets-OBJID.xml

Les liens vers les ressources sont relatifs aux racines configurées dans Pleade.

Exemple : 

DOCID/FRAD044_PR_451_1_1/FRAD044_PR_451_1_1_001.jpg
DOCID/FRAD044_PR_451_1_1/FRAD044_PR_451_1_1_001.xml

Le namespace XLink utilisé est : 

xmlns:xlink="http://www.w3.org/1999/xlink"

Métadonnées injectées dans le METS

[modifier | modifier le wikicode]

Les informations proviennent du fichier CSV.

Champ CSV Destination
Cote PR dc:cote
Titre + Périodicité + Commentaires dc:title
Nbre de pages estimé dc:pages-count
Lieu de conservation du document original mets:agent/mets:name

Gestion des dates

[modifier | modifier le wikicode]

Certaines colonnes étant instables, les dates sont récupérées par position (index 1-based).

Colonne Usage
7 date de début
8 date de fin
9 date révolutionnaire

Les valeurs sont concaténées dans : 

<dc:date>

Génération de l'EAD maître

[modifier | modifier le wikicode]

Option : 

--ead

Le script produit un EAD contenant :

  • le fonds
  • les séries (titres de périodiques)
  • les items (fascicules)

Lien vers le METS : 

dao href="DOCID/mets-OBJID.xml?base=mets"

Avec paramètres : 

&np=premiere_page.jpg
&nd=derniere_page.jpg

Analyse des fichiers ALTO

[modifier | modifier le wikicode]

Le script peut analyser les fichiers ALTO afin de détecter les anomalies suivantes :

  • XML invalide
  • attributs XML dupliqués
  • incohérence dimensions image / ALTO
  • absence d'identifiants

Ces anomalies peuvent être reportées dans un rapport CSV.

Correction automatique des ALTO

[modifier | modifier le wikicode]

Lorsque nécessaire, le script peut :

  • ajouter les identifiants manquants sur :
    • TextBlock
    • TextLine
    • String
  • recalculer les coordonnées lorsque les dimensions de l'image ne correspondent pas à celles utilisées lors de l'OCR.

Cela permet d'assurer l'affichage correct des surlignages dans Pleade.

Redimensionnement des coordonnées ALTO

[modifier | modifier le wikicode]

Lorsque l'image utilisée dans Pleade est différente de celle utilisée lors de l'OCR :

  • les coordonnées sont recalculées
  • les proportions sont conservées

Exemple : 

ALTO : 1603x2064
IMAGE : 3206x4128
scale = 2

Toutes les coordonnées sont multipliées par ce facteur.

Nettoyage XML

[modifier | modifier le wikicode]

Le script corrige automatiquement certains problèmes XML fréquents :

  • attributs dupliqués
  • encodages incohérents

Modes d'exécution

[modifier | modifier le wikicode]

Traitement complet

[modifier | modifier le wikicode]
python3 make_mets_package.py INPUT OUTPUT \
 --pleade-strict \
 --ead \
 --eadid FRAD044_test_presse \
 --ead-title "La presse de Loire Atlantique" \
 --fonds-title "La presse de Loire Atlantique" \
 --repository AD44 \
 --genreform "Périodiques" \
 --ead-out ead/FRAD044_test_presse.xml

Audit des ALTO

[modifier | modifier le wikicode]
python3 make_mets_package.py INPUT OUTPUT \
 --audit-only \
 --report-csv audit_alto.csv

Produit un rapport : 

audit_alto.csv

Correction ALTO uniquement

[modifier | modifier le wikicode]
python3 make_mets_package.py INPUT OUTPUT \
 --alto-only

Dans ce mode :

  • les METS ne sont pas générés
  • seuls les ALTO sont corrigés.

Options disponibles

[modifier | modifier le wikicode]
Option Description
--pleade-strict évite de modifier les ALTO déjà corrects
--audit-only analyse uniquement les ALTO
--alto-only corrige uniquement les ALTO
--report-csv produit un rapport d’audit
--ead génère l’EAD maître

Encodage CSV

[modifier | modifier le wikicode]

Encodages testés automatiquement :

  • utf-8-sig
  • cp1252
  • latin1

Délimiteur CSV : 

;

Bonnes pratiques

[modifier | modifier le wikicode]
  • lancer d'abord un audit :
--audit-only
  • conserver les ALTO originaux
  • vérifier le rapport CSV en cas d'anomalie

Historique

[modifier | modifier le wikicode]

v5.2 :

  • génération METS
  • génération EAD
  • gestion CSV

v6.x :

  • audit ALTO
  • redimensionnement coordonnées ALTO
  • ajout automatique d'ID ALTO
  • nettoyage XML


Script python pour générer la conversion

[modifier | modifier le wikicode]
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
make_mets_package.py (v6.6)

Fix Pleade probable:
- NE PAS redéclarer xmlns:xlink sur chaque <mets:FLocat> (ça rebinde le préfixe xlink
  sur une autre URI et peut casser le parsing Pleade).
- Garder le namespace standard déclaré à la racine :
    xmlns:xlink="http://www.w3.org/1999/xlink"

METS:
- xlink:href des JPEG/ALTO relatifs aux racines Pleade images/alto :
    DOCID/<relpath>/<page>.jpg
    DOCID/<relpath>/<page>.xml

EAD:
- dao@href inclut base=mets + np/nd (premier/dernier jpg), avec & échappés en &amp;.
"""

from __future__ import annotations

import argparse
import csv
import datetime as _dt
import re
import hashlib
import struct
import xml.etree.ElementTree as ET
import shutil
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from xml.sax.saxutils import escape

CSV_ENCODING_CANDIDATES = ["utf-8-sig", "cp1252", "latin1"]

COL_COTE = "Cote PR"
COL_TITRE = "Titre"
COL_PERIO = "Périodicité"
COL_COMM = "Commentaires sur le fascicule"
COL_LIEU = "Lieu de conservation du document original"
COL_PAGES_EST = "Nbre de pages estimé"
COL_METS_PATH = "METS"

# EAD: colonnes par indices (1-based)
EAD_DAY_COL_1BASED = 3
EAD_MONTHTXT_COL_1BASED = 4
EAD_YEAR_COL_1BASED = 6

# Dates: 7/8/9 (1-based)
DATE_START_COL_1BASED = 7
DATE_END_COL_1BASED   = 8
DATE_REV_COL_1BASED   = 9

IGNORED_FILENAMES = {".DS_Store", "Thumbs.db"}

def utc_now_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")

def read_csv_rows(csv_path: Path) -> Tuple[List[str], List[List[str]]]:
    raw = csv_path.read_bytes()
    last_err: Optional[Exception] = None
    for enc in CSV_ENCODING_CANDIDATES:
        try:
            text = raw.decode(enc)
            reader = csv.reader(text.splitlines(), delimiter=';')
            all_rows = list(reader)
            if not all_rows:
                return [], []
            headers = [h.strip() for h in all_rows[0]]
            rows = [[(c or "").strip() for c in r] for r in all_rows[1:] if any((c or "").strip() for c in r)]
            return headers, rows
        except Exception as e:
            last_err = e
    raise RuntimeError(f"Impossible de lire le CSV {csv_path} (dernier essai: {last_err})")

def row_to_dict(headers: List[str], row: List[str]) -> Dict[str, str]:
    return {headers[i]: (row[i] if i < len(row) else "") for i in range(len(headers)) if headers[i]}

def get_col_1based(row: List[str], idx_1based: int) -> str:
    i = idx_1based - 1
    if i < 0 or i >= len(row):
        return ""
    return (row[i] or "").strip()

def objid_from_mets_cell(mets_cell: str) -> Optional[str]:
    if not mets_cell:
        return None
    s = mets_cell.replace("\\", "/")
    base = s.rsplit("/", 1)[-1]
    if base.lower().endswith(".xml"):
        base = base[:-4]
    return base.strip() or None

def choose_doc_root(collection_dir: Path) -> Optional[Path]:
    candidates = [p for p in collection_dir.iterdir()
                  if p.is_dir() and not p.name.startswith(".") and p.name != "__MACOSX"]
    if not candidates:
        return None
    return sorted(candidates, key=lambda p: p.name.lower())[0]

def find_fascicule_dirs(doc_root: Path) -> List[Path]:
    fascicules: List[Path] = []
    for p in doc_root.rglob("*"):
        if p.is_dir() and p.name.lower() == "alto":
            fascicules.append(p.parent)
    seen = set()
    uniq = []
    for f in fascicules:
        if f in seen:
            continue
        seen.add(f)
        uniq.append(f)
    return sorted(uniq, key=lambda x: str(x).lower())

def find_child_dir(parent: Path, names: Tuple[str, ...]) -> Optional[Path]:
    for n in names:
        p = parent / n
        if p.exists() and p.is_dir():
            return p
        for cand in parent.iterdir():
            if cand.is_dir() and cand.name.lower() == n.lower():
                return cand
    return None

def list_pages_from_files(files: List[Path], fascicule_id: str, ext: str) -> List[Tuple[str, Path]]:
    out = []
    pat = re.compile(rf"^{re.escape(fascicule_id)}_(\d{{3}})\.{re.escape(ext)}$", re.IGNORECASE)
    for fp in files:
        if fp.name in IGNORED_FILENAMES:
            continue
        m = pat.match(fp.name)
        if not m:
            continue
        out.append((f"{fascicule_id}_{m.group(1)}", fp))
    out.sort(key=lambda t: int(t[0].split("_")[-1]))
    return out

def ensure_dir(p: Path) -> None:
    p.mkdir(parents=True, exist_ok=True)

def copy_files(pairs: List[Tuple[str, Path]], dst_dir: Path) -> None:
    ensure_dir(dst_dir)
    for _, src in pairs:
        shutil.copy2(src, dst_dir / src.name)

def build_dc_title(titre: str, periodicite: str, commentaires: str) -> str:
    parts = [x for x in (titre, periodicite, commentaires) if x]
    return " - ".join(parts)

def build_dc_date(deb: str, fin: str, rev: str) -> str:
    parts = [x for x in (deb, fin, rev) if x]
    return " - ".join(parts)




# -------- ALTO audit / fix helpers --------

ALTO_ID_TARGETS = (
    "ComposedBlock",
    "TextBlock",
    "TextLine",
    "String",
    "SP",
    "Illustration",
    "GraphicalElement",
)
ALTO_COORD_ATTRS = ("HPOS", "VPOS", "WIDTH", "HEIGHT")
ALTO_POINTS_ATTRS = ("POINTS", "POINTS_STRING", "SHAPE_POINTS")


def get_png_size(image_path: Path) -> Tuple[int, int]:
    with image_path.open("rb") as f:
        sig = f.read(8)
        if sig != b"\x89PNG\r\n\x1a\n":
            raise ValueError("Not a PNG file")
        _length = struct.unpack(">I", f.read(4))[0]
        chunk_type = f.read(4)
        if chunk_type != b"IHDR":
            raise ValueError("PNG missing IHDR")
        width, height = struct.unpack(">II", f.read(8))
        return width, height

def get_jpeg_size(image_path: Path) -> Tuple[int, int]:
    with image_path.open("rb") as f:
        if f.read(2) != b"\xff\xd8":
            raise ValueError("Not a JPEG file")
        while True:
            byte = f.read(1)
            if not byte:
                raise ValueError("Invalid JPEG: EOF before size")
            if byte != b"\xff":
                continue
            marker = f.read(1)
            while marker == b"\xff":
                marker = f.read(1)
            if not marker:
                raise ValueError("Invalid JPEG marker")
            m = marker[0]
            if m in (0xD8, 0xD9, 0x01) or (0xD0 <= m <= 0xD7):
                continue
            seglen = struct.unpack(">H", f.read(2))[0]
            if m in (0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF):
                _precision = f.read(1)
                height, width = struct.unpack(">HH", f.read(4))
                return width, height
            f.seek(seglen - 2, 1)

def get_gif_size(image_path: Path) -> Tuple[int, int]:
    with image_path.open("rb") as f:
        header = f.read(6)
        if header not in (b"GIF87a", b"GIF89a"):
            raise ValueError("Not a GIF file")
        width, height = struct.unpack("<HH", f.read(4))
        return width, height

def get_image_size(image_path: Path) -> Tuple[int, int]:
    suffix = image_path.suffix.lower()
    if suffix in (".jpg", ".jpeg"):
        return get_jpeg_size(image_path)
    if suffix == ".png":
        return get_png_size(image_path)
    if suffix == ".gif":
        return get_gif_size(image_path)
    raise ValueError(f"Format image non supporté sans bibliothèque standard (sans Pillow): {image_path.suffix}")

def alto_qname(ns: Optional[str], local: str) -> str:
    return f"{{{ns}}}{local}" if ns else local

def alto_local_name(tag: str) -> str:
    if tag.startswith("{"):
        return tag.split("}", 1)[1]
    return tag

def detect_alto_ns_from_text(xml_text: str) -> Optional[str]:
    m = re.search(r'<(?:[A-Za-z0-9_]+:)?alto\b[^>]*\bxmlns(?::[A-Za-z0-9_]+)?="([^"]+)"', xml_text)
    return m.group(1) if m else None

def clean_duplicate_attributes(xml_text: str) -> str:
    """
    Remove duplicate attributes within a single tag, keeping the first occurrence.
    """
    tag_re = re.compile(r'<[^!?][^<>]*?>', re.DOTALL)
    attr_re = re.compile(r'([A-Za-z_:][A-Za-z0-9_.:-]*)\s*=\s*"([^"]*)"')

    def repl_tag(match):
        tag = match.group(0)
        if tag.startswith("</"):
            return tag
        seen = set()
        parts = []
        last = 0
        for am in attr_re.finditer(tag):
            parts.append(tag[last:am.start()])
            name = am.group(1)
            if name not in seen:
                seen.add(name)
                parts.append(am.group(0))
            last = am.end()
        parts.append(tag[last:])
        cleaned = "".join(parts)
        cleaned = re.sub(r"\s{2,}", " ", cleaned)
        cleaned = cleaned.replace(" />", "/>")
        return cleaned

    return tag_re.sub(repl_tag, xml_text)

def parse_alto_tree(alto_path: Path) -> Tuple[ET.ElementTree, ET.Element, Optional[str]]:
    raw = alto_path.read_text(encoding="utf-8", errors="replace")
    cleaned = clean_duplicate_attributes(raw)
    root = ET.fromstring(cleaned)
    tree = ET.ElementTree(root)
    ns = None
    if root.tag.startswith("{"):
        ns = root.tag.split("}", 1)[0][1:]
    else:
        ns = detect_alto_ns_from_text(cleaned)
    return tree, root, ns

def find_alto_page(root: ET.Element, ns: Optional[str]) -> Optional[ET.Element]:
    return root.find(".//" + alto_qname(ns, "Page"))

def count_missing_alto_ids(root: ET.Element) -> Dict[str, int]:
    counts: Dict[str, int] = {}
    for elem in root.iter():
        name = alto_local_name(elem.tag)
        if name not in ALTO_ID_TARGETS:
            continue
        if "ID" not in elem.attrib or not elem.attrib["ID"].strip():
            counts[name] = counts.get(name, 0) + 1
    return counts

def ensure_alto_ids(root: ET.Element, prefix: str) -> Dict[str, int]:
    created: Dict[str, int] = {}
    counters: Dict[str, int] = {}
    used_ids = set()

    for elem in root.iter():
        current = elem.attrib.get("ID", "").strip()
        if current:
            used_ids.add(current)

    for elem in root.iter():
        name = alto_local_name(elem.tag)
        if name not in ALTO_ID_TARGETS:
            continue
        current = elem.attrib.get("ID", "").strip()
        if current:
            continue

        counters[name] = counters.get(name, 0) + 1
        new_id = f"{prefix}_{name}_{counters[name]:06d}"
        while new_id in used_ids:
            counters[name] += 1
            new_id = f"{prefix}_{name}_{counters[name]:06d}"

        elem.attrib["ID"] = new_id
        used_ids.add(new_id)
        created[name] = created.get(name, 0) + 1

    return created

def scale_alto_numeric_attr(elem: ET.Element, attr: str, scale_x: float, scale_y: float) -> None:
    if attr not in elem.attrib:
        return
    try:
        val = float(elem.attrib[attr])
    except ValueError:
        return
    if attr in ("HPOS", "WIDTH"):
        elem.attrib[attr] = str(int(round(val * scale_x)))
    elif attr in ("VPOS", "HEIGHT"):
        elem.attrib[attr] = str(int(round(val * scale_y)))

def scale_alto_points_attr(elem: ET.Element, attr: str, scale_x: float, scale_y: float) -> None:
    if attr not in elem.attrib:
        return
    raw = elem.attrib[attr].strip()
    if not raw:
        return
    if "," in raw:
        pts = []
        for chunk in raw.split():
            if "," not in chunk:
                pts.append(chunk)
                continue
            xs, ys = chunk.split(",", 1)
            try:
                x = round(float(xs) * scale_x)
                y = round(float(ys) * scale_y)
                pts.append(f"{x},{y}")
            except ValueError:
                pts.append(chunk)
        elem.attrib[attr] = " ".join(pts)
    else:
        parts = raw.split()
        out = []
        for i, token in enumerate(parts):
            try:
                num = float(token)
                out.append(str(int(round(num * (scale_x if i % 2 == 0 else scale_y)))))
            except ValueError:
                out.append(token)
        elem.attrib[attr] = " ".join(out)



def detect_declared_encoding(xml_bytes: bytes) -> str:
    head = xml_bytes[:200].decode("ascii", errors="ignore")
    m = re.search(r'encoding=["\']([^"\']+)["\']', head, re.I)
    return m.group(1) if m else "UTF-8"

def maybe_fix_alto(alto_src: Path, matching_jpg: Optional[Path], alto_dst: Path, audit_messages: List[str], *, pleade_strict: bool=False) -> None:
    original_bytes = alto_src.read_bytes()
    declared_encoding = detect_declared_encoding(original_bytes)
    raw = original_bytes.decode(declared_encoding, errors="replace")

    cleaned = clean_duplicate_attributes(raw)
    cleaned_duplicates = (cleaned != raw)

    root = ET.fromstring(cleaned)
    tree = ET.ElementTree(root)

    ns = None
    if root.tag.startswith("{"):
        ns = root.tag.split("}",1)[0][1:]

    page = find_alto_page(root, ns)
    if page is None:
        raise ValueError(f"Aucune balise <Page> trouvée dans {alto_src}")

    changed = False
    prefix = alto_src.stem

    if cleaned_duplicates:
        changed = True
        audit_messages.append(f"[ALTO XML] {alto_src.name}: attribut(s) dupliqué(s) nettoyé(s)")

    missing_before = count_missing_alto_ids(root)
    if sum(missing_before.values()) > 0:
        created = ensure_alto_ids(root, prefix)
        changed = True
        audit_messages.append(
            f"[ALTO IDs] {alto_src.name}: IDs ajoutés " +
            ", ".join(f"{k}={v}" for k, v in sorted(created.items()))
        )

    old_w = int(page.attrib["WIDTH"])
    old_h = int(page.attrib["HEIGHT"])

    if matching_jpg is not None and matching_jpg.exists():
        img_w, img_h = get_image_size(matching_jpg)
        scale_x = img_w / old_w
        scale_y = img_h / old_h
        ratio_old = old_w / old_h if old_h else 0
        ratio_new = img_w / img_h if img_h else 0
        ratio_gap = abs(ratio_old - ratio_new)

        if old_w == img_w and old_h == img_h:
            audit_messages.append(f"[ALTO SIZE] {alto_src.name}: OK ({old_w}x{old_h})")
        elif ratio_gap < 0.001:
            for elem in root.iter():
                for attr in ALTO_COORD_ATTRS:
                    scale_alto_numeric_attr(elem, attr, scale_x, scale_y)
                for attr in ALTO_POINTS_ATTRS:
                    scale_alto_points_attr(elem, attr, scale_x, scale_y)
            page.attrib["WIDTH"] = str(img_w)
            page.attrib["HEIGHT"] = str(img_h)
            changed = True
            audit_messages.append(
                f"[ALTO SIZE] {alto_src.name}: coordonnées recalculées "
                f"ALTO {old_w}x{old_h} -> image {img_w}x{img_h} "
                f"(scale_x={scale_x:.6f}, scale_y={scale_y:.6f})"
            )
        else:
            audit_messages.append(
                f"[ALTO SIZE] {alto_src.name}: ATTENTION ratio incompatible "
                f"ALTO {old_w}x{old_h} vs image {img_w}x{img_h} "
                f"(aucune correction automatique)"
            )
    else:
        audit_messages.append(f"[ALTO SIZE] {alto_src.name}: image correspondante introuvable")

    ensure_dir(alto_dst.parent)

    if pleade_strict and not changed:
        shutil.copy2(alto_src, alto_dst)
        return

    if changed:
        if ns:
            ET.register_namespace("", ns)
        ET.register_namespace("xsi", "http://www.w3.org/2001/XMLSchema-instance")
        tree.write(alto_dst, encoding=declared_encoding, xml_declaration=True)
    else:
        shutil.copy2(alto_src, alto_dst)

# -------- end ALTO audit / fix helpers --------


# -------- audit/report helpers --------

def append_audit_row(audit_rows: List[Dict[str, str]], **kwargs) -> None:
    row = {
        "collection": "",
        "docid": "",
        "issue_type": "",
        "file": "",
        "status": "",
        "details": "",
    }
    row.update({k: str(v) for k, v in kwargs.items()})
    audit_rows.append(row)

def write_audit_csv(audit_rows: List[Dict[str, str]], out_csv: Path) -> None:
    ensure_dir(out_csv.parent)
    fieldnames = ["collection", "docid", "issue_type", "file", "status", "details"]
    with out_csv.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames, delimiter=";")
        w.writeheader()
        for row in audit_rows:
            w.writerow(row)

def analyze_alto(alto_src: Path, matching_jpg: Optional[Path]) -> Dict[str, object]:
    raw = alto_src.read_text(encoding="utf-8", errors="replace")
    cleaned = clean_duplicate_attributes(raw)
    cleaned_duplicates = (cleaned != raw)

    root = ET.fromstring(cleaned)
    ns = None
    if root.tag.startswith("{"):
        ns = root.tag.split("}", 1)[0][1:]
    else:
        ns = detect_alto_ns_from_text(cleaned)

    page = find_alto_page(root, ns)
    if page is None:
        raise ValueError(f"Aucune balise <Page> trouvée dans {alto_src}")

    missing_before = count_missing_alto_ids(root)
    old_w = int(page.attrib["WIDTH"])
    old_h = int(page.attrib["HEIGHT"])

    dim_status = "NO_IMAGE"
    dim_details = "image correspondante introuvable"
    scale_x = ""
    scale_y = ""

    if matching_jpg is not None and matching_jpg.exists():
        img_w, img_h = get_image_size(matching_jpg)
        sx = img_w / old_w
        sy = img_h / old_h
        scale_x = f"{sx:.6f}"
        scale_y = f"{sy:.6f}"
        ratio_old = old_w / old_h if old_h else 0
        ratio_new = img_w / img_h if img_h else 0
        ratio_gap = abs(ratio_old - ratio_new)
        if old_w == img_w and old_h == img_h:
            dim_status = "OK"
            dim_details = f"{old_w}x{old_h}"
        elif ratio_gap < 0.001:
            dim_status = "RESCALABLE"
            dim_details = f"ALTO {old_w}x{old_h} -> IMG {img_w}x{img_h}"
        else:
            dim_status = "INCOMPATIBLE"
            dim_details = f"ALTO {old_w}x{old_h} vs IMG {img_w}x{img_h}"

    return {
        "cleaned_duplicates": cleaned_duplicates,
        "missing_ids": missing_before,
        "dim_status": dim_status,
        "dim_details": dim_details,
        "scale_x": scale_x,
        "scale_y": scale_y,
    }

# -------- end audit/report helpers --------



def make_mets_xml(
    objid: str,
    docid: str,
    relpath_from_docroot: Path,
    jpg_pages: List[Tuple[str, Path]],
    alto_pages: List[Tuple[str, Path]],
    csv_dict: Optional[Dict[str, str]],
    csv_row: Optional[List[str]],
) -> str:
    created = utc_now_iso()
    d = csv_dict or {}

    lieu = d.get(COL_LIEU, "")
    cote = d.get(COL_COTE, "")
    titre = d.get(COL_TITRE, "")
    perio = d.get(COL_PERIO, "")
    comm = d.get(COL_COMM, "")

    pages_est = d.get(COL_PAGES_EST, "")
    if not pages_est:
        pages_est = str(len(jpg_pages) if jpg_pages else len(alto_pages))

    r = csv_row or []
    d_deb = get_col_1based(r, DATE_START_COL_1BASED)
    d_fin = get_col_1based(r, DATE_END_COL_1BASED)
    d_rev = get_col_1based(r, DATE_REV_COL_1BASED)

    dc_title = build_dc_title(titre, perio, comm)
    dc_date = build_dc_date(d_deb, d_fin, d_rev)

    base_rel = relpath_from_docroot

    page_ids = []
    seen = set()
    for pid, _ in jpg_pages + alto_pages:
        if pid in seen:
            continue
        seen.add(pid)
        page_ids.append(pid)

    def _num(pid: str) -> int:
        try:
            return int(pid.split("_")[-1])
        except Exception:
            return 10**9
    page_ids.sort(key=_num)

    lines = []
    lines.append('<?xml version="1.0" encoding="utf-8"?>')
    lines.append('<mets:mets xmlns:mets="http://www.loc.gov/METS/" '
                 'xmlns:xlink="http://www.w3.org/1999/xlink" '
                 'xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" '
                 'OBJID="{objid}" '
                 'xsi:schemaLocation="http://www.loc.gov/METS/ http://www.loc.gov/standards/mets/mets.xsd">'.format(
                     objid=escape(objid)
                 ))
    lines.append(f'  <mets:metsHdr CREATEDATE="{created}">')
    lines.append('    <mets:agent ROLE="EDITOR" TYPE="ORGANIZATION">')
    lines.append(f'      <mets:name>{escape(lieu)}</mets:name>')
    lines.append('    </mets:agent>')
    lines.append('  </mets:metsHdr>')

    lines.append(f'  <mets:dmdSec ID="{escape(objid)}">')
    lines.append('    <mets:mdWrap MIMETYPE="text/xml" MDTYPE="EAD" LABEL="Données fournies par ad44">')
    lines.append('      <mets:xmlData>')
    lines.append('        <dc:root xmlns:dc="http://www.pleade.com/ns/1.0">')
    lines.append(f'          <dc:cote type="raw">{escape(cote)}</dc:cote>')
    lines.append(f'          <dc:title>{escape(dc_title)}</dc:title>')
    lines.append('          <dc:author/>')
    lines.append(f'          <dc:pages-count>{escape(pages_est)}</dc:pages-count>')
    lines.append(f'          <dc:date>{escape(dc_date)}</dc:date>')
    lines.append('        </dc:root>')
    lines.append('      </mets:xmlData>')
    lines.append('    </mets:mdWrap>')
    lines.append('  </mets:dmdSec>')

    lines.append('  <mets:fileSec>')
    if jpg_pages:
        lines.append('    <mets:fileGrp USE="JPEG" ID="FG_JPEG">')
        for pid, _src in jpg_pages:
            file_id = f"jpeg-{objid}-{pid}"
            href = (base_rel / f"{pid}.jpg").as_posix()
            page_num = int(pid.split("_")[-1])
            dmdid = f"f-{objid}{page_num}"
            lines.append(f'      <mets:file ID="{escape(file_id)}" MIMETYPE="image/jpeg" AMDID="{escape(objid)}" DMDID="{escape(dmdid)}">')
            lines.append(f'        <mets:FLocat LOCTYPE="URL" xlink:href="{escape(href)}"/>')
            lines.append('      </mets:file>')
        lines.append('    </mets:fileGrp>')
    if alto_pages:
        lines.append('    <mets:fileGrp USE="ALTO" ID="FG_ALTO">')
        for pid, _src in alto_pages:
            file_id = f"alto-{objid}-{pid}"
            href = (base_rel / f"{pid}.xml").as_posix()
            page_num = int(pid.split("_")[-1])
            dmdid = f"f-{objid}{page_num}"
            lines.append(f'      <mets:file ID="{escape(file_id)}" MIMETYPE="text/xml" AMDID="{escape(objid)}" DMDID="{escape(dmdid)}">')
            lines.append(f'        <mets:FLocat LOCTYPE="URL" xlink:href="{escape(href)}"/>')
            lines.append('      </mets:file>')
        lines.append('    </mets:fileGrp>')
    lines.append('  </mets:fileSec>')

    lines.append('  <mets:structMap ID="strmp.1" TYPE="PHYSICAL">')
    lines.append(f'    <mets:div ID="{escape(objid)}" TYPE="document" LABEL="{escape(objid)}" DMDID="{escape(objid)}">')
    order = 1
    for pid in page_ids:
        page_div_id = f"{objid}:{order}:{pid}"
        lines.append(f'      <mets:div ID="{escape(page_div_id)}" TYPE="page" ORDER="{order}" LABEL="Page {order}">')
        if any(p == pid for p, _ in jpg_pages):
            lines.append(f'        <mets:fptr FILEID="{escape(f"jpeg-{objid}-{pid}")}"/>')
        if any(p == pid for p, _ in alto_pages):
            lines.append(f'        <mets:fptr FILEID="{escape(f"alto-{objid}-{pid}")}"/>')
        lines.append('      </mets:div>')
        order += 1
    lines.append('    </mets:div>')
    lines.append('  </mets:structMap>')
    lines.append('</mets:mets>')
    return "\n".join(lines) + "\n"

def make_ead_master(
    eadid: str,
    titleproper: str,
    fonds_title: str,
    repository: str,
    genreform: str,
    series_items: List[Dict[str, str]],
) -> str:
    groups: Dict[str, List[Dict[str,str]]] = {}
    order: List[str] = []
    for it in series_items:
        k = it["series_key"]
        if k not in groups:
            groups[k] = []
            order.append(k)
        groups[k].append(it)

    def esc(s: str) -> str:
        return escape(s or "")

    lines = []
    lines.append('<?xml version="1.0" encoding="UTF-8" standalone="no"?>')
    lines.append('<ead>')
    lines.append('  <eadheader countryencoding="iso3166-1" dateencoding="iso8601" langencoding="iso639-2b" repositoryencoding="iso15511" scriptencoding="iso15924">')
    lines.append(f'    <eadid>{esc(eadid)}</eadid>')
    lines.append('    <filedesc><titlestmt>')
    lines.append(f'      <titleproper>{esc(titleproper)}</titleproper>')
    lines.append('    </titlestmt></filedesc>')
    lines.append('    <profiledesc>')
    lines.append('      <creation>Instrument de recherche produit depuis les fichiers, leur contenu et les répertoires fournis pour ce test</creation>')
    lines.append('      <langusage>Instrument de recherche rédigé en <language langcode="fre">français</language></langusage>')
    lines.append('    </profiledesc>')
    lines.append('  </eadheader>')

    lines.append('  <archdesc level="fonds">')
    lines.append('    <did>')
    lines.append('      <unitid label="Cotes extrêmes">PR</unitid>')
    lines.append(f'      <unittitle label="Intitulé de l\'unité documentaire">{esc(fonds_title)}</unittitle>')
    lines.append(f'      <repository label="Organisme responsable de l\'accès intellectuel">{esc(repository)}</repository>')
    lines.append('    </did>')
    lines.append('    <controlaccess>')
    lines.append(f'      <genreform type="typir">{esc(genreform)}</genreform>')
    lines.append('    </controlaccess>')
    lines.append('    <dsc type="in-depth">')

    series_counter = 1
    for series_key in order:
        items = groups[series_key]
        series_id = f"PR{series_counter:04d}"
        series_counter += 1

        lines.append(f'      <c id="{series_id}" level="series">')
        lines.append('        <did>')
        lines.append(f'          <unittitle>{esc(items[0]["series_title"])}</unittitle>')
        lines.append(f'          <unitid>{esc(items[0]["series_unitid"])}</unitid>')
        lines.append('        </did>')

        for idx, it in enumerate(items, start=1):
            item_id = f"{series_id}_{idx:02d}"
            lines.append(f'        <c id="{item_id}" level="item">')
            lines.append('          <did>')
            lines.append(f'            <unittitle>{esc(it["item_unittitle"])}</unittitle>')
            lines.append(f'            <unitdate>{esc(it["item_unitdate"])}</unitdate>')
            lines.append('          </did>')
            lines.append(f'          <dao href="{esc(it["dao_href"])}" role="mets" title="Consulter le numéro"/>')
            lines.append('        </c>')

        lines.append('      </c>')

    lines.append('    </dsc>')
    lines.append('  </archdesc>')
    lines.append('</ead>')
    return "\n".join(lines) + "\n"

def process_collection(collection_dir: Path, out_root: Path, series_acc: List[Dict[str,str]], *, audit_rows: Optional[List[Dict[str, str]]] = None, audit_only: bool = False, alto_only: bool = False, pleade_strict: bool = False) -> None:
    csv_files = [p for p in collection_dir.iterdir() if p.is_file() and p.suffix.lower() == ".csv"]
    if not csv_files:
        print(f"[WARN] Aucun CSV trouvé dans {collection_dir}")
        return
    csv_path = sorted(csv_files, key=lambda p: p.name.lower())[0]
    headers, rows = read_csv_rows(csv_path)

    mets_idx = None
    for i, h in enumerate(headers):
        if h.strip().lower() == "mets":
            mets_idx = i
            break

    by_objid: Dict[str, Tuple[Dict[str, str], List[str]]] = {}
    for row in rows:
        d = row_to_dict(headers, row)
        mets_cell = row[mets_idx] if (mets_idx is not None and mets_idx < len(row)) else d.get(COL_METS_PATH, "")
        objid = objid_from_mets_cell(mets_cell)
        if objid:
            by_objid[objid.lower()] = (d, row)

    doc_root = choose_doc_root(collection_dir)
    if not doc_root:
        print(f"[WARN] Aucun doc_root trouvé dans {collection_dir}")
        return
    docid = doc_root.name

    out_alto = out_root / "alto" / docid
    out_img  = out_root / "images" / docid
    out_mets = out_root / "mets" / docid
    ensure_dir(out_alto); ensure_dir(out_img); ensure_dir(out_mets)
    if audit_rows is None:
        audit_rows = []

    for fasc_dir in find_fascicule_dirs(doc_root):
        objid = fasc_dir.name
        relpath = fasc_dir.relative_to(doc_root)

        alto_dir = find_child_dir(fasc_dir, ("alto", "Alto"))
        if not alto_dir:
            print(f"[WARN] Pas de répertoire alto dans {fasc_dir}")
            continue
        jpg_dir = find_child_dir(fasc_dir, ("jpg", "Jpg"))

        alto_pages = list_pages_from_files([p for p in alto_dir.iterdir() if p.is_file()], objid, "xml")
        jpg_pages = []
        if jpg_dir:
            jpg_pages = list_pages_from_files([p for p in jpg_dir.iterdir() if p.is_file()], objid, "jpg")

        audit_messages: List[str] = []
        jpg_map = {pid: src for pid, src in jpg_pages}

        if not audit_only:
            ensure_dir(out_alto / relpath)

        for pid, alto_src in alto_pages:
            matching_jpg = jpg_map.get(pid)
            alto_dst = (out_alto / relpath / alto_src.name)
            try:
                analysis = analyze_alto(alto_src, matching_jpg)
                if analysis["cleaned_duplicates"]:
                    append_audit_row(
                        audit_rows,
                        collection=collection_dir.name,
                        docid=docid,
                        issue_type="alto_xml",
                        file=str(alto_src),
                        status="CLEANABLE",
                        details="attribut(s) dupliqué(s) détecté(s)",
                    )
                missing_ids = analysis["missing_ids"]
                if sum(missing_ids.values()) > 0:
                    append_audit_row(
                        audit_rows,
                        collection=collection_dir.name,
                        docid=docid,
                        issue_type="alto_ids",
                        file=str(alto_src),
                        status="MISSING_IDS",
                        details=", ".join(f"{k}={v}" for k, v in sorted(missing_ids.items())),
                    )
                append_audit_row(
                    audit_rows,
                    collection=collection_dir.name,
                    docid=docid,
                    issue_type="alto_size",
                    file=str(alto_src),
                    status=str(analysis["dim_status"]),
                    details=str(analysis["dim_details"]),
                )

                if not audit_only:
                    maybe_fix_alto(alto_src, matching_jpg, alto_dst, audit_messages, pleade_strict=pleade_strict)
            except ET.ParseError as e:
                msg = f"[ALTO XML] {alto_src}: XML invalide ({e})"
                audit_messages.append(msg)
                append_audit_row(
                    audit_rows,
                    collection=collection_dir.name,
                    docid=docid,
                    issue_type="alto_xml",
                    file=str(alto_src),
                    status="INVALID_XML",
                    details=str(e),
                )
                continue
            except Exception as e:
                msg = f"[ALTO ERROR] {alto_src}: {e}"
                audit_messages.append(msg)
                append_audit_row(
                    audit_rows,
                    collection=collection_dir.name,
                    docid=docid,
                    issue_type="alto_error",
                    file=str(alto_src),
                    status="ERROR",
                    details=str(e),
                )
                continue

        if (not audit_only) and jpg_pages:
            copy_files(jpg_pages, out_img / relpath)

        for msg in audit_messages:
            print(msg)

        d, r = by_objid.get(objid.lower(), ({}, []))

        if not audit_only and not alto_only:
            (out_mets / f"mets-{objid}.xml").write_text(
                make_mets_xml(objid, docid, relpath, jpg_pages, alto_pages, d, r),
                encoding="utf-8"
            )

        # ---- EAD entry ----
        titre = d.get(COL_TITRE, "")
        perio = d.get(COL_PERIO, "")
        cote = d.get(COL_COTE, "")
        series_title = " - ".join([x for x in (titre, perio) if x]) or docid
        series_key = f"{docid.lower()}|{cote.strip().lower()}|{titre.strip().lower()}|{perio.strip().lower()}"

        day = get_col_1based(r, EAD_DAY_COL_1BASED)
        monthtxt = get_col_1based(r, EAD_MONTHTXT_COL_1BASED)
        year = get_col_1based(r, EAD_YEAR_COL_1BASED)
        item_unittitle = " ".join([x for x in (day, monthtxt, year) if x]) or objid

        d_deb = get_col_1based(r, DATE_START_COL_1BASED)
        d_fin = get_col_1based(r, DATE_END_COL_1BASED)
        d_rev = get_col_1based(r, DATE_REV_COL_1BASED)
        item_unitdate = build_dc_date(d_deb, d_fin, d_rev)

        dao_href = f"{docid}/mets-{objid}.xml?base=mets"
        if jpg_pages:
            first_jpg = f"{jpg_pages[0][0]}.jpg"
            last_jpg  = f"{jpg_pages[-1][0]}.jpg"
            dao_href += f"&np={first_jpg}&nd={last_jpg}"

        series_acc.append({
            "series_key": series_key,
            "series_title": series_title,
            "series_unitid": cote,
            "item_unittitle": item_unittitle,
            "item_unitdate": item_unitdate,
            "dao_href": dao_href,
        })

def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("input_root")
    ap.add_argument("output_root")
    ap.add_argument("--ead", action="store_true", help="Génère un EAD maître")
    ap.add_argument("--eadid", default="FRAD044_test_presse")
    ap.add_argument("--ead-title", default="La presse de Loire Attantique")
    ap.add_argument("--fonds-title", default="La presse de Loire Attantique")
    ap.add_argument("--repository", default="AD44")
    ap.add_argument("--genreform", default="Périodiques")
    ap.add_argument("--ead-out", default="ead/FRAD044_test_presse.xml")

    ap.add_argument("--pleade-strict", action="store_true",
                    help="ne réécrit pas les ALTO déjà corrects")
    ap.add_argument("--audit-only", action="store_true",
                    help="analyse seulement les ALTO sans générer de METS")
    ap.add_argument("--alto-only", action="store_true",
                    help="corrige uniquement les ALTO")
    ap.add_argument("--report-csv",
                    help="écrit un rapport CSV d'audit")
    args = ap.parse_args()

    audit_only = getattr(args, "audit_only", False)
    alto_only = getattr(args, "alto_only", False)
    pleade_strict = getattr(args, "pleade_strict", False)
    report_csv = getattr(args, "report_csv", None)

    input_root = Path(args.input_root).expanduser().resolve()
    out_root = Path(args.output_root).expanduser().resolve()
    if audit_only and alto_only:
        raise SystemExit("--audit-only et --alto-only sont incompatibles")
    out_root.mkdir(parents=True, exist_ok=True)

    collections = [p for p in input_root.iterdir()
                   if p.is_dir() and not p.name.startswith(".") and p.name != "__MACOSX"]
    if not collections:
        raise SystemExit(f"Aucun lot trouvé dans {input_root}")

    series_items: List[Dict[str,str]] = []
    audit_rows: List[Dict[str, str]] = []
    for c in sorted(collections, key=lambda p: p.name.lower()):
        process_collection(c, out_root, series_items, audit_rows=audit_rows, audit_only=audit_only, alto_only=alto_only, pleade_strict=pleade_strict)

    if report_csv:
        write_audit_csv(audit_rows, out_root / report_csv)
        print(f"[AUDIT CSV] rapport écrit : {out_root / report_csv}")

    if args.ead and (not audit_only) and (not alto_only):
        ead_xml = make_ead_master(
            eadid=args.eadid,
            titleproper=args.ead_title,
            fonds_title=args.fonds_title,
            repository=args.repository,
            genreform=args.genreform,
            series_items=series_items,
        )
        ead_path = out_root / args.ead_out
        ensure_dir(ead_path.parent)
        ead_path.write_text(ead_xml, encoding="utf-8")
        print(f"EAD maître généré: {ead_path}")

    print(f"OK. Sortie générée dans: {out_root}")

if __name__ == "__main__":
    main()
Récupérée de "https://mediawiki.ajlsm.com/index.php?title=Générateur_METS_local_pour_presse_ad44&oldid=6030"