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
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 &.
"""
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()