sebastien
0a4f4a80d0
Bundle the trained heatmap/image orientation model and run it after provisional blank-page and split analysis. Confirm candidates with Gemini using low thinking; resolve mixed A3 sheet ordering with the complete statement and medium thinking. Keep A4 scans independent and export paired A3 scans in booklet reading order. Reanalyse uniformly inverted source sheets, validate ordering permutations, and preserve source provenance, corrected rotations and reading slots in manual review. Document the experiments, rejected approaches, calibration and runtime limits. Validation: focused splitting/orientation regression suites pass; bundled model checksum and wheel assets verified. Full Interro07 offline run: 44.26 seconds, eight candidates, mocked upright confirmations, source hashes and geometry intact.
190 lines
7.5 KiB
Python
190 lines
7.5 KiB
Python
"""Publish automatic splits into an evaluation, preserving sources and review data."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import shutil
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
from copienator import CliError, ExitCode
|
||
from copienator.auto_page_split import run as analyse_batch
|
||
from copienator.copy_errors import clear_copy_error, copy_errors, mark_copy_error
|
||
from copienator.filesystem import publish_file_sets
|
||
from copienator.orientation import report_path
|
||
|
||
|
||
def decision_path(workspace, stem):
|
||
return workspace.metadata_dir / "page_splitting" / f"{stem}.json"
|
||
|
||
|
||
def load_decisions(workspace, source):
|
||
"""Only reuse decisions for the exact preserved input and published output."""
|
||
try:
|
||
data = json.loads(decision_path(workspace, source.stem).read_text())
|
||
if data["source_sha256"] != hashlib.sha256(source.read_bytes()).hexdigest():
|
||
return None
|
||
output = workspace.copies_dir / source.name
|
||
digest = (
|
||
hashlib.sha256(output.read_bytes()).hexdigest() if output.exists() else None
|
||
)
|
||
if digest != data["output_sha256"]:
|
||
return None
|
||
pages = data["pages"]
|
||
if not isinstance(pages, list) or not all(
|
||
isinstance(row, dict)
|
||
and type(row.get("source_page")) is int
|
||
and isinstance(row.get("outputs"), list)
|
||
and all(
|
||
isinstance(o, dict)
|
||
and o.get("side") in {"left", "right", "whole"}
|
||
and type(o.get("keep")) is bool
|
||
and isinstance(o.get("crop_points"), list)
|
||
and len(o["crop_points"]) == 4
|
||
for o in row["outputs"]
|
||
)
|
||
for row in pages
|
||
):
|
||
return None
|
||
return pages
|
||
except (OSError, ValueError, KeyError, TypeError, AttributeError):
|
||
return None
|
||
|
||
|
||
def run(workspace, inputs: list[Path], *, workers=4):
|
||
if workers < 1:
|
||
raise CliError(
|
||
"Le nombre de traitements parallèles doit être positif.",
|
||
ExitCode.INVALID_ARGUMENTS,
|
||
)
|
||
if not inputs:
|
||
print("Aucune copie PDF à traiter.", flush=True)
|
||
return ExitCode.SUCCESS
|
||
if len({p.name for p in inputs}) != len(inputs):
|
||
raise CliError(
|
||
"Plusieurs copies portent le même nom.", ExitCode.INVALID_ARGUMENTS
|
||
)
|
||
copy_errors(workspace) # Validate existing review metadata before publishing.
|
||
workspace.runs_dir.mkdir(parents=True, exist_ok=True)
|
||
run_dir = Path(tempfile.mkdtemp(prefix="page-split-", dir=workspace.runs_dir))
|
||
snapshots = run_dir / "inputs"
|
||
snapshots.mkdir()
|
||
for source in inputs:
|
||
shutil.copy2(source, snapshots / source.name)
|
||
result = run_dir / "result"
|
||
print(f"Découpage automatique de {len(inputs)} copie(s).", flush=True)
|
||
summary = analyse_batch(snapshots, result, workers=workers,
|
||
orientation=True, statement=workspace.statement_pdf_file)
|
||
records = json.loads((result / "report.json").read_text())["pages"]
|
||
stages = {
|
||
name: run_dir / name for name in ("copies", "originals", "origins", "decisions")
|
||
}
|
||
for stage in stages.values():
|
||
stage.mkdir()
|
||
for source in inputs:
|
||
rows = [r for r in records if r["file"] == source.name]
|
||
expected = rows[0]["source_sha256"]
|
||
if hashlib.sha256(source.read_bytes()).hexdigest() != expected:
|
||
raise CliError(
|
||
f"{source.name} a changé pendant l’analyse. Aucun PDF remplacé."
|
||
)
|
||
shutil.copy2(snapshots / source.name, stages["originals"] / source.name)
|
||
for directory, label in (
|
||
(workspace.copies_dir, "previous_copies"),
|
||
(workspace.original_copies_dir, "previous_originals"),
|
||
):
|
||
prior = directory / source.name
|
||
if prior.exists():
|
||
backup = run_dir / label
|
||
backup.mkdir(exist_ok=True)
|
||
shutil.copy2(prior, backup / source.name)
|
||
output = result / "pdfs" / source.name
|
||
digest = (
|
||
hashlib.sha256(output.read_bytes()).hexdigest() if output.exists() else None
|
||
)
|
||
if digest:
|
||
shutil.copy2(output, stages["copies"] / source.name)
|
||
ordered = sorted(
|
||
[(r,o) for r in rows for o in r['outputs'] if o['keep']],
|
||
key=lambda item: item[1]['output_page'],
|
||
)
|
||
origins = [
|
||
{"origin": r["source_page"], "crop_points": o["crop_points"],
|
||
"source_rotation_correction": r.get('applied_rotation', 0),
|
||
"sheet_source_pages": r.get('sheet_source_pages', [r['source_page']]),
|
||
"logical_rotation": o.get('rotation', 0)}
|
||
for r,o in ordered
|
||
]
|
||
if digest:
|
||
(stages["origins"] / (source.stem + ".json")).write_text(
|
||
json.dumps({"version": 1, "sha256": digest, "pages": origins}),
|
||
encoding="utf-8",
|
||
)
|
||
(stages["decisions"] / (source.stem + ".json")).write_text(
|
||
json.dumps(
|
||
{
|
||
"version": 1,
|
||
"source_sha256": expected,
|
||
"output_sha256": digest,
|
||
"pages": rows,
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
names = [p.name for p in inputs]
|
||
for source in inputs:
|
||
if source.read_bytes() != (snapshots / source.name).read_bytes():
|
||
raise CliError(
|
||
f"{source.name} a changé pendant l’analyse. Aucun PDF remplacé."
|
||
)
|
||
publish_file_sets(
|
||
{
|
||
workspace.copies_dir: (stages["copies"], names),
|
||
workspace.original_copies_dir: (stages["originals"], []),
|
||
workspace.metadata_dir / "page_origins": (
|
||
stages["origins"],
|
||
[p.stem + ".json" for p in inputs],
|
||
),
|
||
workspace.metadata_dir / "page_splitting": (stages["decisions"], []),
|
||
}
|
||
)
|
||
for source in inputs:
|
||
if (
|
||
source.parent.resolve()
|
||
not in {
|
||
workspace.copies_dir.resolve(),
|
||
workspace.original_copies_dir.resolve(),
|
||
}
|
||
and source.read_bytes() == (snapshots / source.name).read_bytes()
|
||
):
|
||
# Root scans become preserved originals, as in the manual workflow.
|
||
source.unlink()
|
||
rows = [r for r in records if r["file"] == source.name]
|
||
output = workspace.copies_dir / source.name
|
||
clear_copy_error(workspace, output, source="orientation")
|
||
report_path(workspace, source.stem).unlink(missing_ok=True)
|
||
reasons = {reason for r in rows for reason in r["review_reasons"]}
|
||
if reasons or not output.exists():
|
||
mark_copy_error(
|
||
workspace,
|
||
output,
|
||
"Découpage automatique à vérifier manuellement."
|
||
if reasons
|
||
else "Toutes les pages semblent vides : vérifier l’original.",
|
||
source="page_splitter",
|
||
)
|
||
else:
|
||
clear_copy_error(workspace, output, source="page_splitter")
|
||
print(
|
||
f"Terminé : {summary['output_pages']} pages conservées, {summary['blank_removed']} pages vides retirées. "
|
||
"Orientation vérifiée ; feuilles A3 mises dans l’ordre de lecture, A4 conservées à leur place.",
|
||
flush=True,
|
||
)
|
||
print(
|
||
f"Rapport et sauvegardes : {run_dir}\n"
|
||
"Choisissez Manuellement pour vérifier ou modifier les propositions.",
|
||
flush=True,
|
||
)
|
||
return ExitCode.SUCCESS
|