diff --git a/Readme.org b/Readme.org index 20d5afa..47b9227 100644 --- a/Readme.org +++ b/Readme.org @@ -167,8 +167,11 @@ Les codes de sortie communs sont : | 130 | interruption par l'utilisateur | Le GUI distingue notamment un traitement partiel d'un échec. Les -premiers scripts migrés vers cette convention sont =export.py=, -=import.py=, =giving_names.py= et =grouping.py=. +scripts migrés vers cette convention sont actuellement : + +- =copies_tools.py=, =grouping.py= et =verify_groups.py= ; +- =post-correction.py= et =resolve_manual.py= ; +- =export.py=, =import.py= et =giving_names.py=. ** Correction d'un paquet de copies @@ -289,6 +292,11 @@ Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~ Regroupe les mêmes questions de différentes copies en groupes de tailles raisonnables. + 5. Facultatif : =python verify_groups.py Interro= + + Vérifie que chaque réponse PDF apparaît bien dans les métadonnées + des groupes. La commande renvoie un code non nul si une réponse est + absente ou si la vérification est incomplète. ** Correction et annotation diff --git a/copienator/__init__.py b/copienator/__init__.py index bdc56e3..dfff4d9 100644 --- a/copienator/__init__.py +++ b/copienator/__init__.py @@ -6,6 +6,7 @@ from .cli import ( evaluation_parser, evaluation_workspace, execute, + standard_parser, workspace_from_args, ) from .json_io import ( @@ -35,5 +36,6 @@ __all__ = [ "evaluation_workspace", "execute", "read_json", + "standard_parser", "workspace_from_args", ] diff --git a/copienator/cli.py b/copienator/cli.py index 509b6f8..68ec86b 100644 --- a/copienator/cli.py +++ b/copienator/cli.py @@ -30,13 +30,8 @@ class CliError(Exception): super().__init__(message) -def evaluation_parser(description: str) -> argparse.ArgumentParser: +def standard_parser(description: str) -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=description) - parser.add_argument( - "evaluation", - type=Path, - help="Evaluation directory", - ) parser.add_argument( "--verbose", action="store_true", @@ -45,6 +40,16 @@ def evaluation_parser(description: str) -> argparse.ArgumentParser: return parser +def evaluation_parser(description: str) -> argparse.ArgumentParser: + parser = standard_parser(description) + parser.add_argument( + "evaluation", + type=Path, + help="Evaluation directory", + ) + return parser + + def evaluation_workspace( path: str | Path, *, diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index e00d7a2..47a76cc 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -212,6 +212,16 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: requires=("Copies",), artifacts=("Par label",), ), + StepDefinition( + "verify_groups", + "Labels et regroupement", + "Vérifier les groupes produits", + "Vérifie que chaque réponse PDF apparaît dans les métadonnées des groupes.", + (python("default", "Vérification des groupes", "verify_groups.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + optional=True, + requires=("Copies", "Par label"), + ), StepDefinition( "correction", "Correction", diff --git a/copies_tools.py b/copies_tools.py index 9310a89..5eaf6d1 100644 --- a/copies_tools.py +++ b/copies_tools.py @@ -2,10 +2,19 @@ from __future__ import annotations import argparse import uuid +from collections.abc import Sequence from pathlib import Path from pypdf import PdfReader, PdfWriter +from copienator import ( + EvaluationWorkspace, + ExitCode, + execute, + standard_parser, + workspace_from_args, +) + def copy_pdfs(directory: Path) -> list[Path]: directory = directory.expanduser().resolve() @@ -91,19 +100,37 @@ def rename_all(directory: Path) -> list[tuple[Path, Path]]: return [(source, destination) for source, _temporary, destination in staged] -def main() -> None: - parser = argparse.ArgumentParser(description="Prepare scanned PDF copies.") +def build_parser() -> argparse.ArgumentParser: + parser = standard_parser("Prepare scanned PDF copies.") subparsers = parser.add_subparsers(dest="operation", required=True) for operation in ("rotate", "rename"): subparser = subparsers.add_parser(operation) - subparser.add_argument("directory", type=Path) - args = parser.parse_args() + subparser.add_argument("evaluation", type=Path, help="Evaluation directory") + subparser.add_argument( + "--verbose", + action="store_true", + default=argparse.SUPPRESS, + help="Show a traceback when an unexpected error occurs", + ) + return parser - if args.operation == "rotate": - rotate_all(args.directory) + +def run(workspace: EvaluationWorkspace, *, operation: str) -> ExitCode: + if operation == "rotate": + rotate_all(workspace.root) else: - rename_all(args.directory) + rename_all(workspace.root) + return ExitCode.SUCCESS + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute( + parser, + argv, + lambda args: run(workspace_from_args(args), operation=args.operation), + ) if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/post-correction.py b/post-correction.py index 8c176c0..a8f72a7 100644 --- a/post-correction.py +++ b/post-correction.py @@ -1,155 +1,118 @@ -import sys -import os -import time -from pathlib import Path +from __future__ import annotations + import argparse +import re +from collections.abc import Sequence +from pathlib import Path +from typing import Any -from copienator import atomic_write_json +from copienator import ( + EvaluationWorkspace, + ExitCode, + atomic_write_json, + evaluation_parser, + execute, + read_json, + workspace_from_args, +) -if len(sys.argv) < 2: - sys.exit("Usage: python script.py ") +WORD_LIST_FILE = Path(__file__).with_name("liste_francais.txt") +ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]") -def escape_latex_underscores(text): - r""" - Escape '_' outside LaTeX math environments. - Supports: - - $...$ - - $$...$$ - - \( ... \) - - \[ ... \] - """ - # Regex matching LaTeX math blocks +def build_parser() -> argparse.ArgumentParser: + return evaluation_parser("Clean encoding and LaTeX issues in correction.json.") + + +def escape_latex_underscores(text: str) -> str: + r"""Escape underscores outside LaTeX math environments.""" math_pattern = re.compile( - r'(\$\$.*?\$\$|' # $$...$$ - r'\$.*?\$|' # $...$ - r'\\\(.*?\\\)|' # \( ... \) - r'\\\[.*?\\\])', # \[ ... \] - re.DOTALL + r"(\$\$.*?\$\$|\$.*?\$|\\\(.*?\\\)|\\\[.*?\\\])", + re.DOTALL, ) - - parts = [] + parts: list[str] = [] last_end = 0 - for match in math_pattern.finditer(text): start, end = match.span() - - # Escape underscores outside math - outside = text[last_end:start].replace('_', r'\_') - parts.append(outside) - - # Keep math block unchanged + parts.append(text[last_end:start].replace("_", r"\_")) parts.append(match.group(0)) - last_end = end + parts.append(text[last_end:].replace("_", r"\_")) + return "".join(parts) - # Remaining text after last math block - outside = text[last_end:].replace('_', r'\_') - parts.append(outside) - return ''.join(parts) +def build_lookup_map(word_list_path: Path = WORD_LIST_FILE) -> dict[str, str]: + words = word_list_path.read_text(encoding="utf-8").splitlines() + lookup: dict[str, str] = {} + for word in words: + broken_key = ACCENT_PATTERN.sub("\x00", word) + if "\x00" in broken_key: + lookup[broken_key.lower()] = word + return lookup -arg_path = Path(sys.argv[1]) -tasks = [] # List of tuples: (filepath_str, label_str) -results = {} -INPUT_DIR = str(arg_path) -if not arg_path.exists(): - sys.exit(f"Directory {INPUT_DIR} not found.") - -import json -import ftfy -import re -import urllib.request - -with open('liste_francais.txt', 'r') as f: - french_words = f.read().splitlines() - -# 2. Pre-compute an O(1) lookup dictionary -# We simulate the corruption by replacing accents with null bytes (\x00) -lookup_map = {} -for word in french_words: - # Replace all French accents with \x00 to create the "broken" key - broken_key = re.sub(r'[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]', '\x00', word) - if '\x00' in broken_key: - lookup_map[broken_key] = word # e.g., "\x00cole" -> "école" - -# 3. Fast replace function -def fast_fix(text): - # Find words containing regular letters and null bytes - def replacer(match): +def fast_fix(text: str, lookup: dict[str, str]) -> str: + def replacer(match: re.Match[str]) -> str: broken_word = match.group(0) - # Return the fixed word from our map, or leave it if not found - # (Handles case-insensitivity by falling back to lowercase map) - fixed = lookup_map.get(broken_word.lower()) - # if not fixed: - # print(f"No match found for: {repr(broken_word)}") - return fixed or broken_word + return lookup.get(broken_word.lower(), broken_word) - return re.sub(r'[a-zA-Z\x00]+', replacer, text) - # return text + return re.sub(r"[a-zA-Z\x00]+", replacer, text) -INPUT_FILE = Path(INPUT_DIR) / "correction.json" -OUTPUT_FILE = Path(INPUT_DIR) / "correction.json" - -def fix_hex_corruption_safe(text): - # Only matches \x00 followed by hex if it results in an accented character - # or common Latin-1 symbols - return re.sub(r'\x00([eEfF][0-9a-fA-F])', - lambda m: chr(int(m.group(1), 16)), - text) - -def some_other_replacements(s): - s = s.replace("\neq", "\\neq") - s = s.replace("\not", "\\not") - return s +def fix_hex_corruption_safe(text: str) -> str: + return re.sub( + r"\x00([eEfF][0-9a-fA-F])", + lambda match: chr(int(match.group(1), 16)), + text, + ) -def clean_string(s: str) -> str: - # fix encoding issues - # s = ftfy.fix_text(s) - # print(s) - s = fix_hex_corruption_safe(s) - - s = s.replace('\x19', '\x00') - s = s.replace('\x18', '\x00') - s = s.replace('\x00\x00', '\x00') - - s = re.sub(r' \x00{1,2} ', ' à ', s) - - if '\x00' in s: - s = fast_fix(s) - s = s.replace('\x00', '') - s = some_other_replacements(s) - return escape_latex_underscores(s) +def some_other_replacements(text: str) -> str: + return text.replace("\neq", "\\neq").replace("\not", "\\not") -def clean_obj(obj): - if isinstance(obj, str): - return clean_string(obj) - - elif isinstance(obj, list): - return [clean_obj(x) for x in obj] - - elif isinstance(obj, dict): - r = {} - for k, v in obj.items(): - if k != "suffix": - r[k] = clean_obj(v) - else: - r[k] = v - return r - - else: - return obj +def clean_string(text: str, lookup: dict[str, str]) -> str: + text = fix_hex_corruption_safe(text) + text = text.replace("\x19", "\x00") + text = text.replace("\x18", "\x00") + text = text.replace("\x00\x00", "\x00") + text = re.sub(r" \x00{1,2} ", " à ", text) + if "\x00" in text: + text = fast_fix(text, lookup).replace("\x00", "") + return escape_latex_underscores(some_other_replacements(text)) -with open(INPUT_FILE, "r", encoding="utf-8") as f: - data = json.load(f) +def clean_obj(value: Any, lookup: dict[str, str]) -> Any: + if isinstance(value, str): + return clean_string(value, lookup) + if isinstance(value, list): + return [clean_obj(item, lookup) for item in value] + if isinstance(value, dict): + return { + key: item if key == "suffix" else clean_obj(item, lookup) + for key, item in value.items() + } + return value -data = clean_obj(data) -atomic_write_json(OUTPUT_FILE, data) +def run( + workspace: EvaluationWorkspace, + *, + word_list_path: Path = WORD_LIST_FILE, +) -> ExitCode: + workspace.require_files("correction.json") + lookup = build_lookup_map(word_list_path) + data = read_json(workspace.correction_file) + cleaned = clean_obj(data, lookup) + atomic_write_json(workspace.correction_file, cleaned) + print(f"Fixed JSON saved to {workspace.correction_file}") + return ExitCode.SUCCESS -print("Fixed JSON saved to", OUTPUT_FILE) + +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()) diff --git a/resolve_manual.py b/resolve_manual.py index b276842..495d258 100644 --- a/resolve_manual.py +++ b/resolve_manual.py @@ -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 ") +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()) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 7f62b48..b24448d 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -45,7 +45,12 @@ def load_script_module(filename: str, module_name: str): if spec is None or spec.loader is None: raise RuntimeError(f"Could not load {filename}") module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) + sys.modules[module_name] = module + try: + spec.loader.exec_module(module) + except Exception: + sys.modules.pop(module_name, None) + raise return module @@ -155,12 +160,24 @@ class StandardCliTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.modules = { + "copies_tools": load_script_module( + "copies_tools.py", "copienator_copies_tools_test" + ), "export": load_script_module("export.py", "copienator_export_test"), "import": load_script_module("import.py", "copienator_import_test"), "giving_names": load_script_module( "giving_names.py", "copienator_giving_names_test" ), "grouping": load_script_module("grouping.py", "copienator_grouping_test"), + "post_correction": load_script_module( + "post-correction.py", "copienator_post_correction_test" + ), + "resolve_manual": load_script_module( + "resolve_manual.py", "copienator_resolve_manual_test" + ), + "verify_groups": load_script_module( + "verify_groups.py", "copienator_verify_groups_test" + ), } def test_missing_evaluation_has_standard_exit_code(self) -> None: @@ -171,6 +188,10 @@ class StandardCliTests(unittest.TestCase): "import": [missing], "giving_names": [missing, "BGnot"], "grouping": [missing], + "post_correction": [missing], + "resolve_manual": [missing], + "verify_groups": [missing], + "copies_tools": ["rotate", missing], } for name, arguments in invocations.items(): with self.subTest(script=name), redirect_stderr(io.StringIO()): @@ -223,6 +244,21 @@ class StandardCliTests(unittest.TestCase): {"target": evaluation, "annotation_dir": "BGnot"}, ), "grouping": ("grouping", "default", {"target": evaluation}), + "post_correction": ( + "post_correction", + "default", + {"target": evaluation}, + ), + "resolve_manual": ( + "manual_resolution", + "default", + {"target": evaluation}, + ), + "verify_groups": ( + "verify_groups", + "default", + {"target": evaluation}, + ), } for module_name, (step_id, variant_id, values) in cases.items(): step = steps[step_id] @@ -232,6 +268,23 @@ class StandardCliTests(unittest.TestCase): parsed = self.modules[module_name].build_parser().parse_args(command[3:]) self.assertEqual(str(parsed.evaluation), evaluation) + for step_id in ("rotate", "rename"): + step = steps[step_id] + variant = step.variants[0] + command = build_command( + REPOSITORY, + step, + variant, + {"target": evaluation}, + evaluation, + ) + with self.subTest(script=f"copies_tools:{step_id}"): + parsed = self.modules["copies_tools"].build_parser().parse_args( + command[3:] + ) + self.assertEqual(parsed.operation, step_id) + self.assertEqual(str(parsed.evaluation), evaluation) + def test_export_main_copies_outputs(self) -> None: module = self.modules["export"] with tempfile.TemporaryDirectory() as directory: @@ -287,6 +340,124 @@ class StandardCliTests(unittest.TestCase): self.assertEqual(module.main([str(evaluation)]), 0) self.assertTrue((evaluation / "Par label").is_dir()) + def test_post_correction_main_cleans_json_atomically(self) -> None: + module = self.modules["post_correction"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + evaluation.mkdir() + atomic_write_json( + evaluation / "correction.json", + {"text": "outside_name and $math_name$", "suffix": "_new"}, + ) + self.assertEqual(module.main([str(evaluation)]), 0) + self.assertEqual( + read_json(evaluation / "correction.json"), + {"text": r"outside\_name and $math_name$", "suffix": "_new"}, + ) + + def test_manual_resolution_with_no_actions_is_safe(self) -> None: + module = self.modules["resolve_manual"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + atomic_write_json(evaluation / "correction.json", {}) + manual = evaluation / "manual_resolutions.txt" + manual.write_text("### Nothing to do\n", encoding="utf-8") + + self.assertEqual(module.main([str(evaluation)]), 0) + self.assertFalse(manual.exists()) + self.assertEqual(read_json(evaluation / "correction.json"), {}) + + def test_malformed_manual_resolution_is_not_deleted(self) -> None: + module = self.modules["resolve_manual"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + correction = evaluation / "correction.json" + atomic_write_json(correction, {}) + manual = evaluation / "manual_resolutions.txt" + manual.write_text("this is malformed\n", encoding="utf-8") + before = correction.read_bytes() + + with redirect_stderr(io.StringIO()): + self.assertEqual(module.main([str(evaluation)]), 1) + self.assertTrue(manual.exists()) + self.assertEqual(correction.read_bytes(), before) + + def test_documented_manual_resolution_operators_are_parsed(self) -> None: + module = self.modules["resolve_manual"] + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "manual_resolutions.txt" + path.write_text( + "\n".join( + f"Copie01 Old label {operator} New label" + for operator in module.OPERATORS + ), + encoding="utf-8", + ) + instructions = module.parse_instructions(path) + self.assertEqual( + [instruction.operator for instruction in instructions], + list(module.OPERATORS), + ) + + def test_replace_manual_resolution_creates_refaire_state(self) -> None: + module = self.modules["resolve_manual"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_dir = evaluation / "Copies" / "Copie01" + copy_dir.mkdir(parents=True) + (copy_dir / "Old.pdf").write_bytes(b"old answer") + (copy_dir / "New.pdf").write_bytes(b"replaced answer") + atomic_write_json( + evaluation / "correction.json", + { + "Old": [ + [ + { + "id": "01", + "result": {"error": "wrg-lbl:New?delayed"}, + } + ] + ], + "New": [[{"id": "01", "result": {"error": ""}}]], + }, + ) + (evaluation / "manual_resolutions.txt").write_text( + "Copie01 Old -x New\n", encoding="utf-8" + ) + + self.assertEqual(module.main([str(evaluation)]), 0) + self.assertEqual((copy_dir / "New_old.pdf").read_bytes(), b"replaced answer") + self.assertEqual((copy_dir / "New_new.pdf").read_bytes(), b"old answer") + self.assertEqual( + read_json(evaluation / "refaire.json"), + [["Copie01", ["New"]]], + ) + correction = read_json(evaluation / "correction.json") + self.assertEqual( + correction["Old"][0][0]["result"]["error"], + "wrg-lbl-moved-to:New", + ) + self.assertEqual(correction["New"][0][0]["result"]["suffix"], "_new") + + def test_verify_groups_reports_success_and_missing_answers(self) -> None: + module = self.modules["verify_groups"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_dir = evaluation / "Copies" / "Copie01" + group_dir = evaluation / "Par label" / "Ex 1" + copy_dir.mkdir(parents=True) + group_dir.mkdir(parents=True) + (copy_dir / "Ex 1.pdf").write_bytes(b"pdf") + + self.assertEqual(module.main([str(evaluation)]), 1) + atomic_write_json( + group_dir / "Group_1.json", + [["01", 0, 100, 1.0, "Ex 1"]], + ) + self.assertEqual(module.main([str(evaluation)]), 0) + class WorkflowTests(unittest.TestCase): def setUp(self) -> None: