281 lines
9.5 KiB
Python
281 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import shutil
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from pypdf import PdfWriter
|
|
|
|
from copienator import (
|
|
CliError,
|
|
EvaluationWorkspace,
|
|
ExitCode,
|
|
atomic_write_json,
|
|
evaluation_parser,
|
|
execute,
|
|
read_json,
|
|
workspace_from_args,
|
|
)
|
|
|
|
OPERATORS = ("-x", "->", "x>", "ss", "sx", "xx", "xs")
|
|
OPERATOR_PATTERN = re.compile(r"\s+(-x|->|x>|ss|sx|xx|xs)\s+")
|
|
COPY_PATTERN = re.compile(r"Copie(\d+)\s+(.+)")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ManualInstruction:
|
|
copy_id: str
|
|
old_label: str
|
|
operator: str
|
|
new_label: str
|
|
pipe_first: bool
|
|
|
|
@property
|
|
def should_merge(self) -> bool:
|
|
return self.operator.endswith(">")
|
|
|
|
@property
|
|
def should_copy(self) -> bool:
|
|
return not self.should_merge and "s" not in self.operator
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
return evaluation_parser("Apply the instructions from manual_resolutions.txt.")
|
|
|
|
|
|
def parse_instructions(path: Path) -> list[ManualInstruction]:
|
|
instructions: list[ManualInstruction] = []
|
|
malformed: list[int] = []
|
|
for line_number, raw_line in enumerate(
|
|
path.read_text(encoding="utf-8").splitlines(), start=1
|
|
):
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("###"):
|
|
continue
|
|
operator_match = OPERATOR_PATTERN.search(line)
|
|
if operator_match is None:
|
|
malformed.append(line_number)
|
|
continue
|
|
left = line[: operator_match.start()].strip()
|
|
right = line[operator_match.end() :].strip()
|
|
copy_match = COPY_PATTERN.fullmatch(left)
|
|
new_label = right.strip("|").strip()
|
|
if copy_match is None or not new_label:
|
|
malformed.append(line_number)
|
|
continue
|
|
instructions.append(
|
|
ManualInstruction(
|
|
copy_id=copy_match.group(1),
|
|
old_label=copy_match.group(2).strip(),
|
|
operator=operator_match.group(1),
|
|
new_label=new_label,
|
|
pipe_first=right.startswith("|"),
|
|
)
|
|
)
|
|
if malformed:
|
|
lines = ", ".join(str(number) for number in malformed)
|
|
raise CliError(f"Malformed manual resolution instruction at line(s): {lines}")
|
|
return instructions
|
|
|
|
|
|
def set_suffix_and_clean_error(
|
|
results: dict[str, Any],
|
|
copy_id: str,
|
|
label: str,
|
|
suffix: str | None,
|
|
new_label_target: str | None = None,
|
|
) -> None:
|
|
for batch in results.get(label, []):
|
|
for item in batch:
|
|
if item["id"] != copy_id:
|
|
continue
|
|
if suffix:
|
|
item["result"]["suffix"] = suffix
|
|
error = item["result"].get("error", "")
|
|
if new_label_target:
|
|
if f"wrg-lbl:{new_label_target}?delayed" in error:
|
|
item["result"]["error"] = (
|
|
f"wrg-lbl-moved-to:{new_label_target}"
|
|
)
|
|
if f"(delayed){new_label_target}" in error:
|
|
item["result"]["error"] = error.replace(
|
|
f"(delayed){new_label_target}",
|
|
f"(->){new_label_target}",
|
|
)
|
|
|
|
|
|
def get_actual_pdf(copies_dir: Path, copy_id: str, label: str) -> Path:
|
|
base = copies_dir / f"Copie{copy_id}" / f"{label}.pdf"
|
|
for candidate in (
|
|
base,
|
|
base.with_name(f"{label}_new.pdf"),
|
|
base.with_name(f"{label}_old.pdf"),
|
|
):
|
|
if candidate.exists():
|
|
return candidate
|
|
return base
|
|
|
|
|
|
def safe_strip_suffix(stem: str) -> str:
|
|
if stem.endswith(("_new", "_old")):
|
|
return stem[:-4]
|
|
return stem
|
|
|
|
|
|
def _validate_pdf_inputs(
|
|
instructions: list[ManualInstruction],
|
|
initial_paths: dict[tuple[str, str], Path],
|
|
) -> None:
|
|
missing: set[Path] = set()
|
|
for instruction in instructions:
|
|
source = initial_paths[(instruction.copy_id, instruction.old_label)]
|
|
destination = initial_paths[(instruction.copy_id, instruction.new_label)]
|
|
if instruction.should_merge:
|
|
if not source.exists():
|
|
missing.add(source)
|
|
if not destination.exists():
|
|
missing.add(destination)
|
|
elif instruction.should_copy and not source.exists():
|
|
missing.add(source)
|
|
if missing:
|
|
rendered = ", ".join(str(path) for path in sorted(missing))
|
|
raise CliError(f"PDF input(s) required by manual resolutions not found: {rendered}")
|
|
|
|
|
|
def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
|
|
workspace.require_files("manual_resolutions.txt", "correction.json")
|
|
workspace.require_directories("Copies")
|
|
loaded = read_json(workspace.correction_file)
|
|
if not isinstance(loaded, dict):
|
|
raise CliError("correction.json must contain a JSON object")
|
|
results: dict[str, Any] = loaded
|
|
instructions = parse_instructions(workspace.manual_resolutions_file)
|
|
|
|
initial_paths: dict[tuple[str, str], Path] = {}
|
|
current_paths: dict[tuple[str, str], Path] = {}
|
|
for instruction in instructions:
|
|
for label in (instruction.old_label, instruction.new_label):
|
|
key = (instruction.copy_id, label)
|
|
if key not in initial_paths:
|
|
path = get_actual_pdf(workspace.copies_dir, *key)
|
|
initial_paths[key] = path
|
|
current_paths[key] = path
|
|
_validate_pdf_inputs(instructions, initial_paths)
|
|
|
|
files_to_old: set[Path] = set()
|
|
temp_files: list[Path] = []
|
|
try:
|
|
for instruction in instructions:
|
|
key_old = (instruction.copy_id, instruction.old_label)
|
|
key_new = (instruction.copy_id, instruction.new_label)
|
|
source = initial_paths[key_old]
|
|
destination = current_paths[key_new]
|
|
temp_output = (
|
|
workspace.copies_dir
|
|
/ f"Copie{instruction.copy_id}"
|
|
/ f"temp_{len(temp_files)}.pdf"
|
|
)
|
|
|
|
if instruction.operator.startswith("x"):
|
|
files_to_old.add(initial_paths[key_old])
|
|
if instruction.operator.endswith("x"):
|
|
files_to_old.add(initial_paths[key_new])
|
|
|
|
if instruction.should_merge:
|
|
writer = PdfWriter()
|
|
try:
|
|
if instruction.pipe_first:
|
|
writer.append(source)
|
|
writer.append(destination)
|
|
else:
|
|
writer.append(destination)
|
|
writer.append(source)
|
|
writer.write(temp_output)
|
|
except Exception:
|
|
temp_output.unlink(missing_ok=True)
|
|
raise
|
|
finally:
|
|
writer.close()
|
|
current_paths[key_new] = temp_output
|
|
temp_files.append(temp_output)
|
|
files_to_old.add(initial_paths[key_new])
|
|
elif instruction.should_copy:
|
|
shutil.copy(source, temp_output)
|
|
current_paths[key_new] = temp_output
|
|
temp_files.append(temp_output)
|
|
except Exception:
|
|
for temporary in temp_files:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
|
|
for pdf in files_to_old:
|
|
if not pdf.exists():
|
|
continue
|
|
copy_id = pdf.parent.name.removeprefix("Copie")
|
|
label = safe_strip_suffix(pdf.stem)
|
|
old_name = pdf.with_name(f"{label}_old.pdf")
|
|
if pdf != old_name:
|
|
old_name.unlink(missing_ok=True)
|
|
shutil.move(str(pdf), str(old_name))
|
|
set_suffix_and_clean_error(results, copy_id, label, "_old")
|
|
|
|
for instruction in instructions:
|
|
set_suffix_and_clean_error(
|
|
results,
|
|
instruction.copy_id,
|
|
instruction.old_label,
|
|
None,
|
|
instruction.new_label,
|
|
)
|
|
|
|
refaire_by_copy: dict[str, list[str]] = {}
|
|
for (copy_id, label), current_path in current_paths.items():
|
|
if not current_path.name.startswith("temp_"):
|
|
continue
|
|
final_name = workspace.copies_dir / f"Copie{copy_id}" / f"{label}_new.pdf"
|
|
final_name.unlink(missing_ok=True)
|
|
shutil.move(str(current_path), str(final_name))
|
|
set_suffix_and_clean_error(results, copy_id, label, "_new")
|
|
labels = refaire_by_copy.setdefault(f"Copie{copy_id}", [])
|
|
if label not in labels:
|
|
labels.append(label)
|
|
|
|
used_temps = set(current_paths.values())
|
|
for temporary in temp_files:
|
|
if temporary not in used_temps:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
atomic_write_json(workspace.correction_file, results)
|
|
refaire_tasks = [[copy_name, labels] for copy_name, labels in refaire_by_copy.items()]
|
|
if refaire_tasks:
|
|
atomic_write_json(workspace.refaire_file, refaire_tasks)
|
|
workspace.manual_resolutions_file.unlink()
|
|
|
|
print("Manual resolutions successfully applied.")
|
|
if refaire_tasks:
|
|
print(
|
|
f"File {workspace.refaire_file.name} generated. Run "
|
|
f'`python correction.py "{workspace.command_argument()}" --refaire` '
|
|
"to process updates."
|
|
)
|
|
else:
|
|
print("No new corrections required.")
|
|
return ExitCode.SUCCESS
|
|
|
|
|
|
def run(workspace: EvaluationWorkspace) -> ExitCode:
|
|
return resolve_manual(workspace)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
return execute(parser, argv, lambda args: run(workspace_from_args(args)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|