DS09 : Conflict resolution, and other
This commit is contained in:
@@ -0,0 +1,196 @@
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from pypdf import PdfWriter
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
sys.exit("Usage: python resolve_manual.py <InputDir>")
|
||||
|
||||
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.")
|
||||
|
||||
with open(correction_file, "r", encoding="utf-8") as f:
|
||||
results = json.load(f)
|
||||
|
||||
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}")
|
||||
|
||||
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")
|
||||
return base
|
||||
|
||||
def safe_strip_suffix(stem):
|
||||
if stem.endswith("_new"): return stem[:-4]
|
||||
if stem.endswith("_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}")
|
||||
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", "")
|
||||
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()
|
||||
shutil.move(str(pdf), str(old_name))
|
||||
|
||||
set_suffix_and_clean_error(pid_str, label, "_old")
|
||||
|
||||
# 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()
|
||||
shutil.move(str(current_path), str(final_name))
|
||||
|
||||
set_suffix_and_clean_error(pid, label, "_new")
|
||||
|
||||
# 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]])
|
||||
|
||||
# 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()
|
||||
|
||||
# Finalize JSONs
|
||||
with open(correction_file, "w", encoding="utf-8") as f:
|
||||
json.dump(results, f, indent=2)
|
||||
|
||||
if refaire_tasks:
|
||||
with open(refaire_file, "w", encoding="utf-8") as f:
|
||||
json.dump(refaire_tasks, f, indent=2)
|
||||
|
||||
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.")
|
||||
Reference in New Issue
Block a user