From 96c00fa9412b8a8453c5c87428f1381524809da8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Tue, 22 Sep 2026 12:20:23 +0200 Subject: [PATCH] original-label-> information --- copienator/annotation_data.py | 6 ++++ copienator/commands/annotating.py | 4 +++ copienator/commands/correction.py | 3 ++ copienator/commands/resolve_manual.py | 16 +++++++++ tests/test_manual_resolution_state.py | 52 +++++++++++++++++++++++++++ 5 files changed, 81 insertions(+) diff --git a/copienator/annotation_data.py b/copienator/annotation_data.py index 8084810..ee96a9f 100644 --- a/copienator/annotation_data.py +++ b/copienator/annotation_data.py @@ -10,6 +10,7 @@ from PIL import Image from .json_io import read_json from .feedback_boxes import valid_feedback_box from .workspace import EvaluationWorkspace +from .answer_origins import origins_for AnnotationData = dict[str, dict[str, dict[str, Any]]] RefaireList = list[list[Any]] @@ -208,4 +209,9 @@ def load_annotation_data( data = {copy_id: data[copy_id]} if copy_id in data else {} if not data: warnings.append(f"Copy id {copy_id} was not found in correction.json") + for student_id, answers in data.items(): + for label, answer in answers.items(): + answer["result"]["content_origins"] = origins_for( + workspace.root, student_id, label, answer["pdf_path"] + ) return AnnotationLoadResult(data, warnings) diff --git a/copienator/commands/annotating.py b/copienator/commands/annotating.py index 799a1f1..9713ac1 100644 --- a/copienator/commands/annotating.py +++ b/copienator/commands/annotating.py @@ -324,6 +324,10 @@ def compose_label_image(base_img, label, result, hmin, if error == "empty-answer" and not with_empty: return None, 0 + origins = result.get("content_origins", []) + if origins: + error = " ".join([error, *(f"{origin}->" for origin in origins)]).strip() + # Filter deleted items (used by reading_annotations.py) feedbacks = [f for f in feedbacks if "to_delete" not in f] diff --git a/copienator/commands/correction.py b/copienator/commands/correction.py index a2ff681..4916a9d 100644 --- a/copienator/commands/correction.py +++ b/copienator/commands/correction.py @@ -18,6 +18,7 @@ from google import genai from copienator import configuration as config from copienator.commands import grouping +from copienator.answer_origins import save_origins from copienator import prompting from copienator.feedback_boxes import valid_feedback_box from copienator import ( @@ -435,6 +436,7 @@ def handle_label_errors(pid, label, res, pdf_path): if not base_add_pdf_path.exists() and not add_pdf_path.exists(): shutil.copy(str(pdf_path), str(add_pdf_path)) + save_origins(INPUT_DIR, pid, add_label, add_pdf_path, [label]) tprint(f"\t\tCopying Copie{pid} : {label} -> {add_label}") idx = reserve_next_group_idx(add_label) tprint(f"\t\tMaking {add_label} group {idx+1}") @@ -704,6 +706,7 @@ def resolve_delayed_moves(): resolved_any = True shutil.copy(str(pdf_path), str(add_pdf_path)) + save_origins(INPUT_DIR, pid, target_label, add_pdf_path, [label]) idx = reserve_next_group_idx(target_label) height = grouping.get_pdf_height(str(add_pdf_path)) grouping.create_jpg(target_label, idx, [(pid, str(add_pdf_path), height)], GROUPS_DIR) diff --git a/copienator/commands/resolve_manual.py b/copienator/commands/resolve_manual.py index a8d5a44..4ea8346 100644 --- a/copienator/commands/resolve_manual.py +++ b/copienator/commands/resolve_manual.py @@ -10,6 +10,7 @@ from typing import Any from pypdf import PdfWriter from copienator.pdf_cut import split_pdf +from copienator.answer_origins import origins_for, save_origins from copienator import ( CliError, @@ -204,6 +205,8 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: _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: @@ -212,6 +215,8 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: 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" @@ -232,6 +237,9 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: 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])) writer = PdfWriter() try: if instruction.pipe_first: @@ -252,6 +260,10 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: 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 [] + ) shutil.copy(source, temp_output) current_paths[key_new] = temp_output temp_files.append(temp_output) @@ -287,6 +299,10 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: 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)) + 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: diff --git a/tests/test_manual_resolution_state.py b/tests/test_manual_resolution_state.py index 9d658c2..bcb5277 100644 --- a/tests/test_manual_resolution_state.py +++ b/tests/test_manual_resolution_state.py @@ -8,10 +8,62 @@ import pymupdf from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json from copienator.commands import correction, resolve_manual +from copienator.answer_origins import origins_for, save_origins +from copienator.annotation_data import load_annotation_data from copienator_gui.app import has_manual_conflicts class ManualResolutionStateTests(unittest.TestCase): + def test_automatic_additional_content_records_origin_but_wrong_label_does_not(self): + for kind in ("additional-answer", "wrong-label"): + with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary: + workspace = self.fixture(Path(temporary), "->") + copies = workspace.copies_dir / "Copie01" + (copies / "B.pdf").unlink() + (workspace.root / "labels").write_text("A\nB\n") + with ( + patch.object(correction, "INPUT_DIR", workspace.root, create=True), + patch.object(correction, "COPIES_DIR", workspace.copies_dir, create=True), + patch.object(correction, "GROUPS_DIR", workspace.groups_dir, create=True), + patch.object(correction, "enonce_total", return_value=""), + patch.object(correction.prompting, "request_for_additional_answer", return_value=([], {})), + patch.object(correction.prompting, "request_for_wrong_label", return_value=([], {})), + patch.object(correction, "call_gemini_with_retries", return_value='["B"]' if kind == "additional-answer" else '"B"'), + patch.object(correction, "reserve_next_group_idx", return_value=0), + patch.object(correction.grouping, "get_pdf_height", return_value=200), + patch.object(correction.grouping, "create_jpg"), + ): + correction.handle_label_errors("01", "A", {"error": kind}, copies / "A.pdf") + self.assertEqual( + origins_for(workspace.root, "01", "B", copies / "B_new.pdf"), + ["A"] if kind == "additional-answer" else [], + ) + + def test_target_provenance_only_for_uncut_retained_sources(self): + for operator in (*resolve_manual.OPERATORS, "c{50}1>", "c{50}2x"): + with self.subTest(operator=operator), tempfile.TemporaryDirectory() as temporary: + workspace = self.fixture(Path(temporary), operator) + resolve_manual.resolve_manual(workspace) + target = resolve_manual.get_actual_pdf(workspace.copies_dir, "01", "B") + self.assertEqual( + origins_for(workspace.root, "01", "B", target), + ["A"] if operator in ("->", "-x") else [], + ) + + def test_provenance_survives_new_correction_but_not_replaced_pdf(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = self.fixture(Path(temporary), "->") + target = workspace.copies_dir / "Copie01" / "B.pdf" + save_origins(workspace.root, "01", "B", target, ["A", "A"]) + atomic_write_json(workspace.correction_file, { + "B": [[{"id": "01", "result": {"score": 4, "error": "", "feedback": []}}]] + }) + self.assertEqual(origins_for(workspace.root, "01", "B", target), ["A"]) + loaded = load_annotation_data(workspace) + self.assertEqual(loaded.data["01"]["B"]["result"]["content_origins"], ["A"]) + target.write_bytes(b"replacement") + self.assertEqual(origins_for(workspace.root, "01", "B", target), []) + def test_merge_into_missing_target_uses_source_and_schedules_redo(self): for operator in ("->", "x>"): for target in ("B|", "|B"):