Standardisation 2
This commit is contained in:
+249
-165
@@ -1,196 +1,280 @@
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
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 atomic_write_json
|
||||
from copienator import (
|
||||
CliError,
|
||||
EvaluationWorkspace,
|
||||
ExitCode,
|
||||
atomic_write_json,
|
||||
evaluation_parser,
|
||||
execute,
|
||||
read_json,
|
||||
workspace_from_args,
|
||||
)
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
sys.exit("Usage: python resolve_manual.py <InputDir>")
|
||||
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+(.+)")
|
||||
|
||||
input_dir = Path(sys.argv[1])
|
||||
manual_file = input_dir / "manual_resolutions.txt"
|
||||
correction_file = input_dir / "correction.json"
|
||||
refaire_file = input_dir / "refaire.json"
|
||||
copies_dir = input_dir / "Copies"
|
||||
|
||||
if not manual_file.exists():
|
||||
sys.exit(f"No {manual_file.name} found. Nothing to resolve.")
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ManualInstruction:
|
||||
copy_id: str
|
||||
old_label: str
|
||||
operator: str
|
||||
new_label: str
|
||||
pipe_first: bool
|
||||
|
||||
with open(correction_file, "r", encoding="utf-8") as f:
|
||||
results = json.load(f)
|
||||
@property
|
||||
def should_merge(self) -> bool:
|
||||
return self.operator.endswith(">")
|
||||
|
||||
def set_suffix_and_clean_error(pid, label, suffix, new_lbl_target=None):
|
||||
"""Updates correction.json to set suffixes and clear resolved delayed tags."""
|
||||
if label in results:
|
||||
for batch in results[label]:
|
||||
for item in batch:
|
||||
if item["id"] == pid:
|
||||
if suffix:
|
||||
item["result"]["suffix"] = suffix
|
||||
err = item["result"].get("error", "")
|
||||
if new_lbl_target:
|
||||
if f"wrg-lbl:{new_lbl_target}?delayed" in err:
|
||||
item["result"]["error"] = f"wrg-lbl-moved-to:{new_lbl_target}"
|
||||
if f"(delayed){new_lbl_target}" in err:
|
||||
item["result"]["error"] = err.replace(f"(delayed){new_lbl_target}", f"(->){new_lbl_target}")
|
||||
@property
|
||||
def should_copy(self) -> bool:
|
||||
return not self.should_merge and "s" not in self.operator
|
||||
|
||||
def get_actual_pdf(pid, label):
|
||||
"""Finds the effective PDF considering possible suffixes."""
|
||||
base = copies_dir / f"Copie{pid}" / f"{label}.pdf"
|
||||
if base.exists(): return base
|
||||
if base.with_name(f"{label}_new.pdf").exists(): return base.with_name(f"{label}_new.pdf")
|
||||
if base.with_name(f"{label}_old.pdf").exists(): return base.with_name(f"{label}_old.pdf")
|
||||
|
||||
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):
|
||||
if stem.endswith("_new"): return stem[:-4]
|
||||
if stem.endswith("_old"): return stem[:-4]
|
||||
|
||||
def safe_strip_suffix(stem: str) -> str:
|
||||
if stem.endswith(("_new", "_old")):
|
||||
return stem[:-4]
|
||||
return stem
|
||||
|
||||
instructions = []
|
||||
with open(manual_file, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("###"): continue
|
||||
|
||||
# Regex to split on the operator properly handles spaces in labels
|
||||
match = re.search(r'\s+(-x|->|x>)\s+', line)
|
||||
if not match:
|
||||
print(f"Skipping malformed line: {line}")
|
||||
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
|
||||
|
||||
op = match.group(1)
|
||||
left = line[:match.start()].strip()
|
||||
right = line[match.end():].strip()
|
||||
|
||||
m_left = re.match(r'Copie(\d+)\s+(.+)', left)
|
||||
if not m_left:
|
||||
continue
|
||||
|
||||
pid = m_left.group(1)
|
||||
old_label = m_left.group(2).strip()
|
||||
new_part = right
|
||||
|
||||
pipe_first = new_part.startswith("|")
|
||||
# pipe_last = new_part.endswith("|")
|
||||
new_label = new_part.strip("|").strip()
|
||||
|
||||
instructions.append((pid, old_label, op, new_label, pipe_first))
|
||||
|
||||
# State trackers
|
||||
initial_paths = {} # Tracks the source files at script start (enables safe Swapping)
|
||||
current_paths = {} # Tracks the "latest active file" (which may be a temp file for chained merges)
|
||||
files_to_old = set()
|
||||
temp_files = []
|
||||
|
||||
# Pre-load existing paths
|
||||
for pid, old_label, _, new_label, _ in instructions:
|
||||
for lbl in (old_label, new_label):
|
||||
if (pid, lbl) not in initial_paths:
|
||||
p = get_actual_pdf(pid, lbl)
|
||||
initial_paths[(pid, lbl)] = p
|
||||
current_paths[(pid, lbl)] = p
|
||||
|
||||
# Evaluate instructions
|
||||
for pid, old_label, op, new_label, pipe_first in instructions:
|
||||
should_merge = op[1] == ">"
|
||||
src_pdf = initial_paths[(pid, old_label)]
|
||||
dest_pdf = current_paths[(pid, new_label)]
|
||||
|
||||
temp_out = copies_dir / f"Copie{pid}" / f"temp_{len(temp_files)}.pdf"
|
||||
|
||||
if op[0] == "x":
|
||||
files_to_old.add(initial_paths[(pid, old_label)])
|
||||
if op[1] == "x":
|
||||
files_to_old.add(initial_paths[(pid, new_label)])
|
||||
|
||||
if should_merge:
|
||||
if not dest_pdf.exists() or not src_pdf.exists():
|
||||
print("Debug : should_merge but, {} or {} doesn't exist"
|
||||
.format(src_pdf, dest_pdf))
|
||||
input("You should Ctrl-C and fix.")
|
||||
# MERGE
|
||||
merger = PdfWriter()
|
||||
if pipe_first:
|
||||
merger.append(src_pdf)
|
||||
merger.append(dest_pdf)
|
||||
else:
|
||||
merger.append(dest_pdf)
|
||||
merger.append(src_pdf)
|
||||
merger.write(temp_out)
|
||||
merger.close()
|
||||
current_paths[(pid, new_label)] = temp_out
|
||||
temp_files.append(temp_out)
|
||||
# Original destination is now embedded in the merge, back it up
|
||||
files_to_old.add(initial_paths[(pid, new_label)])
|
||||
else:
|
||||
if op[1] != "s" and op[0] != "s": # xx or -x
|
||||
shutil.copy(src_pdf, temp_out)
|
||||
current_paths[(pid, new_label)] = temp_out
|
||||
temp_files.append(temp_out)
|
||||
|
||||
|
||||
# Commit Filesystem Changes
|
||||
|
||||
# 1. Archive deprecated files to _old.pdf
|
||||
for pdf in files_to_old:
|
||||
if pdf.exists():
|
||||
pid_str = pdf.parent.name.replace("Copie", "")
|
||||
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:
|
||||
if old_name.exists(): old_name.unlink()
|
||||
old_name.unlink(missing_ok=True)
|
||||
shutil.move(str(pdf), str(old_name))
|
||||
set_suffix_and_clean_error(results, copy_id, label, "_old")
|
||||
|
||||
set_suffix_and_clean_error(pid_str, label, "_old")
|
||||
for instruction in instructions:
|
||||
set_suffix_and_clean_error(
|
||||
results,
|
||||
instruction.copy_id,
|
||||
instruction.old_label,
|
||||
None,
|
||||
instruction.new_label,
|
||||
)
|
||||
|
||||
# 2. Clear all delayed errors for the instructions
|
||||
for pid, old_label, op, new_label, pf in instructions:
|
||||
set_suffix_and_clean_error(pid, old_label, None, new_label)
|
||||
|
||||
refaire_tasks = []
|
||||
|
||||
# 3. Rename active temp files to _new.pdf and queue for --refaire
|
||||
for (pid, label), current_path in current_paths.items():
|
||||
if "temp_" in current_path.name:
|
||||
final_name = copies_dir / f"Copie{pid}" / f"{label}_new.pdf"
|
||||
if final_name.exists(): final_name.unlink()
|
||||
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)
|
||||
|
||||
set_suffix_and_clean_error(pid, label, "_new")
|
||||
used_temps = set(current_paths.values())
|
||||
for temporary in temp_files:
|
||||
if temporary not in used_temps:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
# Push to refaire_tasks uniquely
|
||||
added = False
|
||||
for t in refaire_tasks:
|
||||
if t[0] == f"Copie{pid}":
|
||||
if label not in t[1]:
|
||||
t[1].append(label)
|
||||
added = True
|
||||
break
|
||||
if not added:
|
||||
refaire_tasks.append([f"Copie{pid}", [label]])
|
||||
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()
|
||||
|
||||
# 4. Clean up any unused temp files (overwritten by chained replacements)
|
||||
used_temps = set(current_paths.values())
|
||||
for temp in temp_files:
|
||||
if temp not in used_temps and temp.exists():
|
||||
temp.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
|
||||
|
||||
# Finalize JSONs
|
||||
atomic_write_json(correction_file, results)
|
||||
|
||||
if refaire_tasks:
|
||||
atomic_write_json(refaire_file, refaire_tasks)
|
||||
def run(workspace: EvaluationWorkspace) -> ExitCode:
|
||||
return resolve_manual(workspace)
|
||||
|
||||
manual_file.unlink(missing_ok=True)
|
||||
|
||||
print("Manual resolutions successfully applied.")
|
||||
if refaire_tasks:
|
||||
print(f"File {refaire_file.name} generated. Run `python correction.py \"{input_dir}\" --refaire` to process updates.")
|
||||
else:
|
||||
print("No new corrections required.")
|
||||
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())
|
||||
|
||||
Reference in New Issue
Block a user