358 lines
14 KiB
Python
358 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from pypdf import PdfWriter
|
|
from copienator.pdf_cut import split_pdf
|
|
from copienator.annotation_layout import (copy_answer_pdf, move_answer_pdf,
|
|
page_parts, save_parts, metadata_path)
|
|
from copienator.answer_origins import origins_for, save_origins
|
|
|
|
from copienator import (
|
|
CliError,
|
|
EvaluationWorkspace,
|
|
ExitCode,
|
|
atomic_write_json,
|
|
evaluation_parser,
|
|
execute,
|
|
read_json,
|
|
workspace_from_args,
|
|
)
|
|
|
|
OPERATORS = ("-x", "->", "x>", "ss", "sx", "xx", "xs")
|
|
CUT_PATTERN = re.compile(r"c\{(\d+(?:\.\d+)?)\}([12])([x>])")
|
|
OPERATOR_PATTERN = re.compile(r"\s+(-x|->|x>|ss|sx|xx|xs|c\{\d+(?:\.\d+)?\}[12][x>])\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 cut(self) -> tuple[float, int] | None:
|
|
match = CUT_PATTERN.fullmatch(self.operator)
|
|
return (float(match[1]), int(match[2])) if match else None
|
|
|
|
@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]:
|
|
return parse_instruction_text(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def parse_instruction_text(text: str) -> list[ManualInstruction]:
|
|
instructions: list[ManualInstruction] = []
|
|
malformed: list[int] = []
|
|
for line_number, raw_line in enumerate(
|
|
text.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()
|
|
cut_match = CUT_PATTERN.fullmatch(operator_match.group(1))
|
|
if (copy_match is None or not new_label
|
|
or (cut_match and (not 0 < float(cut_match[1]) < 100
|
|
or copy_match.group(2).strip() == 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:
|
|
# This instruction acknowledges this source/target conflict,
|
|
# including decisions to keep/discard PDFs without merging.
|
|
# Leave other pending targets (and other copies) untouched.
|
|
result = item["result"]
|
|
if "delayed" in result:
|
|
pending = [entry for entry in result["delayed"]
|
|
if entry not in (["wrong-label", new_label_target],
|
|
["add-label", new_label_target])]
|
|
if pending:
|
|
result["delayed"] = pending
|
|
else:
|
|
result.pop("delayed")
|
|
if error in {f"wrg-lbl:{new_label_target}?",
|
|
f"wrg-lbl:{new_label_target}?delayed",
|
|
f"wrg-lbl:{new_label_target}?exists"}:
|
|
error = f"wrg-lbl-moved-to:{new_label_target}"
|
|
error = error.replace(f"(delayed){new_label_target}", f"(->){new_label_target}")
|
|
error = error.replace(f"(->){new_label_target}?", f"(->){new_label_target}")
|
|
result["error"] = error
|
|
|
|
|
|
def pending_manual_conflicts(results: dict[str, Any]) -> set[tuple[str, str, str]]:
|
|
"""Return the copy/source/target conflicts listed by correction's manual file."""
|
|
return {
|
|
(item["id"], label, target)
|
|
for label, batches in results.items()
|
|
for batch in batches
|
|
for item in batch
|
|
for kind, target in item.get("result", {}).get("delayed", [])
|
|
if kind in {"wrong-label", "add-label"}
|
|
}
|
|
|
|
|
|
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)]
|
|
if (instruction.should_merge or 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)
|
|
retained_conflicts = {
|
|
(item.copy_id, item.old_label, item.new_label) for item in instructions
|
|
}
|
|
dismissed_conflicts = pending_manual_conflicts(results) - retained_conflicts
|
|
cut_sources = [(item.copy_id, item.old_label) for item in instructions if item.cut]
|
|
if len(set(cut_sources)) != len(cut_sources):
|
|
raise CliError("Une seule coupe par label source est autorisée dans une résolution.")
|
|
|
|
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()
|
|
origins = {key: origins_for(workspace.root, *key, path)
|
|
for key, path in initial_paths.items()}
|
|
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]
|
|
if instruction.cut:
|
|
origins[key_old] = []
|
|
origins[key_new] = []
|
|
percent, keep = instruction.cut
|
|
first = source.parent / f"temp_{len(temp_files)}.pdf"
|
|
second = source.parent / f"temp_{len(temp_files) + 1}.pdf"
|
|
temp_files.extend((first, second))
|
|
split_pdf(source, percent, first, second)
|
|
retained, source = (first, second) if keep == 1 else (second, first)
|
|
current_paths[key_old] = retained
|
|
files_to_old.add(initial_paths[key_old])
|
|
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:
|
|
if not instruction.cut and instruction.operator == "->":
|
|
origins[key_new] = list(dict.fromkeys([
|
|
*origins[key_new], instruction.old_label]))
|
|
source_parts = page_parts(source)
|
|
destination_parts = page_parts(destination)
|
|
writer = PdfWriter()
|
|
try:
|
|
if instruction.pipe_first:
|
|
writer.append(source)
|
|
if destination.exists():
|
|
writer.append(destination)
|
|
else:
|
|
if destination.exists():
|
|
writer.append(destination)
|
|
writer.append(source)
|
|
writer.write(temp_output)
|
|
save_parts(temp_output, (source_parts + destination_parts
|
|
if instruction.pipe_first else destination_parts + source_parts))
|
|
except Exception:
|
|
temp_output.unlink(missing_ok=True)
|
|
metadata_path(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:
|
|
origins[key_new] = (
|
|
[instruction.old_label] if instruction.operator == "-x"
|
|
and not instruction.cut else []
|
|
)
|
|
copy_answer_pdf(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)
|
|
metadata_path(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)
|
|
move_answer_pdf(pdf, 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)
|
|
move_answer_pdf(current_path, final_name)
|
|
retained_origins = [origin for origin in origins[(copy_id, label)]
|
|
if (copy_id, origin) not in initial_paths
|
|
or initial_paths[(copy_id, origin)] not in files_to_old]
|
|
save_origins(workspace.root, copy_id, label, final_name, retained_origins)
|
|
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)
|
|
metadata_path(temporary).unlink(missing_ok=True)
|
|
|
|
for copy_id, label, target in dismissed_conflicts:
|
|
# A removed instruction acknowledges the conflict without touching PDFs.
|
|
set_suffix_and_clean_error(results, copy_id, label, None, target)
|
|
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 dismissed_conflicts:
|
|
print(f"{len(dismissed_conflicts)} removed conflict(s) dismissed; their PDFs were kept unchanged.")
|
|
if refaire_tasks:
|
|
print(
|
|
f"File {workspace.refaire_file.name} generated. Run "
|
|
f'`python -m copienator correct "{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())
|