From 0a86403ca6886243896cf36320a77d1bce265f0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Mon, 14 Sep 2026 22:20:17 +0200 Subject: [PATCH] miscs (Interro02) : horizontal cutting resolution --- copienator/annotation_actions.py | 7 +- copienator/annotation_data.py | 3 +- copienator/commands/annotating.py | 11 ++ copienator/commands/batch_status.py | 45 ++++++++ copienator/commands/correction.py | 17 ++- copienator/commands/cutleft.py | 4 +- copienator/commands/post_correction.py | 4 + copienator/commands/resolve_manual.py | 60 ++++++++-- copienator/commands/splitting_int.py | 50 +++++--- copienator/feedback_boxes.py | 13 +++ copienator/pdf_cut.py | 57 +++++++++ copienator_gui/app.py | 86 +++++++++++++- copienator_gui/batch_monitor.py | 81 +++++++++++++ copienator_gui/cut_helper.py | 131 +++++++++++++++++++++ copienator_gui/manual_resolution.py | 154 +++++++++++++++++++++++++ copienator_gui/notifications.py | 37 ++++++ copienator_gui/workflow.py | 25 +--- tests/test_batch_monitor.py | 69 +++++++++++ tests/test_batch_status_readiness.py | 54 +++++++++ tests/test_feedback_boxes.py | 82 +++++++++++++ tests/test_gui_convenience.py | 60 ++++++++++ tests/test_gui_core.py | 11 +- tests/test_gui_cut_helper.py | 56 +++++++++ tests/test_gui_manual_resolution.py | 74 ++++++++++++ tests/test_manual_cut.py | 125 ++++++++++++++++++++ tests/test_manual_resolution_state.py | 65 +++++++++++ tests/test_prompting.py | 33 ++++++ tests/test_splitting_geometry.py | 92 +++++++++++++++ 28 files changed, 1447 insertions(+), 59 deletions(-) create mode 100644 copienator/feedback_boxes.py create mode 100644 copienator/pdf_cut.py create mode 100644 copienator_gui/batch_monitor.py create mode 100644 copienator_gui/cut_helper.py create mode 100644 copienator_gui/manual_resolution.py create mode 100644 copienator_gui/notifications.py create mode 100644 tests/test_batch_monitor.py create mode 100644 tests/test_batch_status_readiness.py create mode 100644 tests/test_feedback_boxes.py create mode 100644 tests/test_gui_cut_helper.py create mode 100644 tests/test_gui_manual_resolution.py create mode 100644 tests/test_manual_cut.py create mode 100644 tests/test_manual_resolution_state.py create mode 100644 tests/test_prompting.py create mode 100644 tests/test_splitting_geometry.py diff --git a/copienator/annotation_actions.py b/copienator/annotation_actions.py index 1fc8575..be97b91 100644 --- a/copienator/annotation_actions.py +++ b/copienator/annotation_actions.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Any from .json_io import read_json +from .feedback_boxes import valid_feedback_box Log = Callable[[str], None] @@ -25,8 +26,10 @@ def apply_checkbox_actions( continue result = labels_data[label]["result"] feedbacks = result.get("feedback", []) - global_feedbacks = [item for item in feedbacks if not item.get("box_2d")] - local_feedbacks = [item for item in feedbacks if item.get("box_2d")] + # Match the renderer's fallback for invalid boxes so checkbox indices + # still address the right comment when returned annotations are read. + global_feedbacks = [item for item in feedbacks if not valid_feedback_box(item.get("box_2d"))] + local_feedbacks = [item for item in feedbacks if valid_feedback_box(item.get("box_2d"))] local_feedbacks.sort(key=lambda item: item["box_2d"][0]) for action in label_actions: diff --git a/copienator/annotation_data.py b/copienator/annotation_data.py index b7ce3ff..8084810 100644 --- a/copienator/annotation_data.py +++ b/copienator/annotation_data.py @@ -8,6 +8,7 @@ from typing import Any from PIL import Image from .json_io import read_json +from .feedback_boxes import valid_feedback_box from .workspace import EvaluationWorkspace AnnotationData = dict[str, dict[str, dict[str, Any]]] @@ -69,7 +70,7 @@ def _scaled_result(result: dict[str, Any], coordinates: GroupCoordinates | None) return scaled for feedback in scaled.get("feedback", []): box = feedback.get("box_2d") - if not box or len(box) != 4: + if not box or not valid_feedback_box(box): continue box[0] = int(box[0] * coordinates.height) // 1000 box[2] = int(box[2] * coordinates.height) // 1000 diff --git a/copienator/commands/annotating.py b/copienator/commands/annotating.py index e102ec1..799a1f1 100644 --- a/copienator/commands/annotating.py +++ b/copienator/commands/annotating.py @@ -32,6 +32,7 @@ from copienator import ( from copienator.annotation_data import load_annotation_data from copienator.answer_info import build_answer_info from copienator.filesystem import staged_directory +from copienator.feedback_boxes import valid_feedback_box from copienator.utils import natural_key MARGIN_LEFT = 300 @@ -326,6 +327,16 @@ def compose_label_image(base_img, label, result, hmin, # Filter deleted items (used by reading_annotations.py) feedbacks = [f for f in feedbacks if "to_delete" not in f] + # Never guess where an invalid rectangle belongs, or lose its comment. + # Use a copy so rendering cannot mutate saved correction data. + normalized = [] + for feedback in feedbacks: + box = feedback.get("box_2d") + if box is not None and not valid_feedback_box(box): + print(f"Warning: Copie{with_id or ''} {label}: invalid feedback box {box!r}; displaying the comment without a rectangle.") + feedback = {**feedback, "box_2d": None} + normalized.append(feedback) + feedbacks = normalized global_fb = [f for f in feedbacks if not f.get('box_2d')] local_fb = [f for f in feedbacks if f.get('box_2d')] diff --git a/copienator/commands/batch_status.py b/copienator/commands/batch_status.py index 5f7b051..5ab4950 100644 --- a/copienator/commands/batch_status.py +++ b/copienator/commands/batch_status.py @@ -9,10 +9,13 @@ from google import genai from copienator import configuration as config from copienator import ( CliError, + EvaluationWorkspace, ExitCode, atomic_write_bytes, execute, + read_json, standard_parser, + workspace_from_args, ) @@ -43,6 +46,41 @@ def list_jobs(*, client=None) -> ExitCode: return ExitCode.SUCCESS +def check_evaluation_jobs(workspace: EvaluationWorkspace, *, client=None) -> ExitCode: + """Only report readiness after checking every job recorded for this evaluation.""" + if not workspace.batch_jobs_file.is_file(): + print("Impossible de vérifier les batchs : batch_jobs.json est absent.") + return ExitCode.PARTIAL + manifest = read_json(workspace.batch_jobs_file) + jobs = manifest.get("jobs") if isinstance(manifest, dict) else None + if not isinstance(jobs, dict): + raise CliError(f"Invalid batch manifest: {workspace.batch_jobs_file}") + if not jobs: + print("Aucun job enregistré pour cette évaluation.") + return ExitCode.PARTIAL + if any(not isinstance(entry, dict) or not isinstance(entry.get("name"), str) + or not entry["name"].strip() for entry in jobs.values()): + raise CliError(f"Invalid batch job in {workspace.batch_jobs_file}") + client = client or _client() + ready = True + for tier, entry in jobs.items(): + job = client.batches.get(name=entry["name"]) + state = job.state.name if hasattr(job.state, "name") else job.state + print(f"{tier} — {entry['name']}: {state}") + if state != "JOB_STATE_SUCCEEDED": + ready = False + if getattr(job, "error", None): + print(f" Erreur : {job.error}") + elif not getattr(getattr(job, "dest", None), "file_name", None): + ready = False + print(" Le fichier de résultats n’est pas encore disponible.") + if ready: + print("Tous les batchs de l’évaluation ont réussi. Les résultats sont prêts à récupérer.") + return ExitCode.SUCCESS + print("Les résultats ne sont pas tous prêts. Consultez à nouveau cette étape plus tard.") + return ExitCode.PARTIAL + + def download_job( job_name: str, *, @@ -73,6 +111,8 @@ def build_parser() -> argparse.ArgumentParser: parser = standard_parser("List or download Gemini correction batch jobs") parser.add_argument("--download", metavar="JOB_NAME") parser.add_argument("--output", type=Path, help="Downloaded JSONL destination") + parser.add_argument("--evaluation", type=Path, + help="Check readiness of jobs recorded in this evaluation's batch_jobs.json") return parser @@ -80,6 +120,11 @@ def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() def handle(args: argparse.Namespace) -> ExitCode: + if args.evaluation is not None: + if args.download or args.output is not None: + raise CliError("--evaluation cannot be combined with --download or --output", + ExitCode.INVALID_ARGUMENTS) + return check_evaluation_jobs(workspace_from_args(args)) if args.output is not None and not args.download: raise CliError("--output requires --download", ExitCode.INVALID_ARGUMENTS) if args.download: diff --git a/copienator/commands/correction.py b/copienator/commands/correction.py index f29815a..72f6e9c 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 import prompting +from copienator.feedback_boxes import valid_feedback_box from copienator import ( CliError, EvaluationWorkspace, @@ -267,6 +268,9 @@ def correct_boxes_with_gemini(pid, label, pdf_path, original_feedbacks, for f in corrected_feedbacks: b = f.get("box_2d") if b: + if not valid_feedback_box(b) or any(value < 0 or value > 1000 for value in b): + f["box_2d"] = None + continue ymin_s, xmin_s, ymax_s, xmax_s = b # Y mapping: Add the group Y-offset (yming), then normalize to total_height @@ -475,6 +479,9 @@ def process_single_task(task_tuple, precomputed_response=None): for (i,f) in enumerate(res["feedback"]): b = f.get("box_2d") if b: + if not valid_feedback_box(b): + needs_correction.append(i) + continue ymin, _xmin, ymax, xmax = b ymin = ymin * total_height // 1000 ymax = ymax * total_height // 1000 @@ -484,9 +491,11 @@ def process_single_task(task_tuple, precomputed_response=None): pid, label, group_name) continue - if (ymin < yming - 50 or ymax > ymaxg + 50 or xmax / 1000 > width_r): + if (ymin < yming - 50 or ymax > ymaxg + 50 + or ymin > ymaxg + 50 or ymax < yming - 50 + or _xmin < 0 or xmax / 1000 > width_r): needs_correction.append(i) - break + continue if ymin < yming - 5: ymin = yming - 5 b[0] = ymin * 1000 // total_height @@ -852,7 +861,7 @@ def run_configured(args: argparse.Namespace) -> ExitCode: # Check for remaining unresolved delayed tasks unresolved_delayed = [] with io_lock: - for label, batches in results.items(): + for label, batches in sorted(results.items()): for batch in batches: for p in batch: res = p.get("result", {}) @@ -869,7 +878,7 @@ def run_configured(args: argparse.Namespace) -> ExitCode: manual_path = INPUT_DIR / "manual_resolutions.txt" atomic_write_text( manual_path, - "### Use -> x>, -x, ss, sx, xx, xs\n" + "### Use -> x>, -x, ss, sx, xx, xs, c{43}1>, c{43}2x\n" + "\n".join(unresolved_delayed) + "\n", ) diff --git a/copienator/commands/cutleft.py b/copienator/commands/cutleft.py index 425b359..5d1542a 100644 --- a/copienator/commands/cutleft.py +++ b/copienator/commands/cutleft.py @@ -65,7 +65,9 @@ def stitch_images(image_list: list[Image.Image]) -> Image.Image | None: @lru_cache(maxsize=3) def _get_pdf_pages_cached(pdf_path: Path) -> list[Image.Image]: - return convert_from_path(pdf_path) + # Label coordinates use the full, displayed MediaBox, including on PDFs + # previously cropped by crop-margins. Keep this in sync with split-answers. + return convert_from_path(pdf_path, use_cropbox=False) def get_pdf_pages(pdf_path: Path) -> list[Image.Image]: diff --git a/copienator/commands/post_correction.py b/copienator/commands/post_correction.py index 0a17c3e..f070a0e 100644 --- a/copienator/commands/post_correction.py +++ b/copienator/commands/post_correction.py @@ -9,6 +9,7 @@ from typing import Any from copienator import ( EvaluationWorkspace, ExitCode, + atomic_write_bytes, atomic_write_json, evaluation_parser, execute, @@ -104,6 +105,9 @@ def run( lookup = build_lookup_map(word_list_path) data = read_json(workspace.correction_file) cleaned = clean_obj(data, lookup) + backup = workspace.root / "correction_precleanup.json" + atomic_write_bytes(backup, workspace.correction_file.read_bytes()) + print(f"Original JSON backed up to {backup}") atomic_write_json(workspace.correction_file, cleaned) print(f"Fixed JSON saved to {workspace.correction_file}") return ExitCode.SUCCESS diff --git a/copienator/commands/resolve_manual.py b/copienator/commands/resolve_manual.py index 5b16468..2cc7161 100644 --- a/copienator/commands/resolve_manual.py +++ b/copienator/commands/resolve_manual.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import Any from pypdf import PdfWriter +from copienator.pdf_cut import split_pdf from copienator import ( CliError, @@ -22,7 +23,8 @@ from copienator import ( ) OPERATORS = ("-x", "->", "x>", "ss", "sx", "xx", "xs") -OPERATOR_PATTERN = re.compile(r"\s+(-x|->|x>|ss|sx|xx|xs)\s+") +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+(.+)") @@ -34,6 +36,11 @@ class ManualInstruction: 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(">") @@ -48,10 +55,14 @@ def build_parser() -> argparse.ArgumentParser: 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( - path.read_text(encoding="utf-8").splitlines(), start=1 + text.splitlines(), start=1 ): line = raw_line.strip() if not line or line.startswith("###"): @@ -64,7 +75,10 @@ def parse_instructions(path: Path) -> list[ManualInstruction]: 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: + 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( @@ -97,15 +111,25 @@ def set_suffix_and_clean_error( 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}", - ) + # 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 get_actual_pdf(copies_dir: Path, copy_id: str, label: str) -> Path: @@ -154,6 +178,9 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: raise CliError("correction.json must contain a JSON object") results: dict[str, Any] = loaded instructions = parse_instructions(workspace.manual_resolutions_file) + 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] = {} @@ -174,6 +201,15 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode: key_new = (instruction.copy_id, instruction.new_label) source = initial_paths[key_old] destination = current_paths[key_new] + if instruction.cut: + 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}" diff --git a/copienator/commands/splitting_int.py b/copienator/commands/splitting_int.py index 6421cab..a92db80 100644 --- a/copienator/commands/splitting_int.py +++ b/copienator/commands/splitting_int.py @@ -80,15 +80,9 @@ def _save_cropped_page( y1: float, output_path: Path, ) -> None: - page = document[page_number] - rotated_rectangle = page.rect * page.transformation_matrix - visual_crop = pymupdf.Rect( - rotated_rectangle.x0 + x0, - y0, - rotated_rectangle.x0 + x1, - y1, - ) - unrotated_clip = visual_crop * page.derotation_matrix + # The source has been normalized by _prepare_split_pages: no rotation or + # CropBox translation remains in the coordinates passed to show_pdf_page. + visual_crop = pymupdf.Rect(x0, y0, x1, y1) cropped = pymupdf.open() try: target_page = cropped.new_page(width=visual_crop.width, height=visual_crop.height) @@ -96,14 +90,32 @@ def _save_cropped_page( target_page.rect, document, page_number, - rotate=-page.rotation, - clip=unrotated_clip, + clip=visual_crop, ) cropped.save(output_path) finally: cropped.close() +def _prepare_split_pages(document: pymupdf.Document) -> list[pymupdf.Rect]: + """Use the label preview's full-page coordinates; retain visible bounds. + + Work only on the in-memory document. Baking rotation into the content after + restoring the MediaBox avoids show_pdf_page's rotated CropBox offsets. + Intersecting with the saved bounds later preserves prior margin cropping. + """ + visible_bounds = [] + for page in document: + crop = page.cropbox + media = page.mediabox + full_crop = pymupdf.Rect(media.x0, 0, media.x1, media.height) + page.set_cropbox(full_crop) + crop -= (full_crop.x0, full_crop.y0, full_crop.x0, full_crop.y0) + visible_bounds.append(crop * page.rotation_matrix) + page.remove_rotation() + return visible_bounds + + def _render_split_outputs( input_pdf: Path, coords_list: list[Coordinate], @@ -112,6 +124,7 @@ def _render_split_outputs( """Render every current answer into an otherwise empty staging directory.""" document = pymupdf.open(input_pdf) try: + visible_bounds = _prepare_split_pages(document) parsed = _parse_coordinates(coords_list) parts_by_label: defaultdict[str, list[Path]] = defaultdict(list) with tempfile.TemporaryDirectory(prefix="copienator-split-") as temp_directory: @@ -160,16 +173,20 @@ def _render_split_outputs( page = document[page_number] y0 = (y_start / 1000) * page.rect.height if page_number == start_page else 0 y1 = (end_y / 1000) * page.rect.height if page_number == end_page else page.rect.height - if y1 <= y0 + 1: + clip = pymupdf.Rect( + fraction_x0 * page.rect.width, y0, + fraction_x1 * page.rect.width, y1, + ) & visible_bounds[page_number] & page.rect + if clip.is_empty or clip.height <= 1 or clip.width <= 1: continue part_path = temporary / f"part-{index}-{page_number}.pdf" _save_cropped_page( document, page_number, - fraction_x0 * page.rect.width, - y0, - fraction_x1 * page.rect.width, - y1, + clip.x0, + clip.y0, + clip.x1, + clip.y1, part_path, ) parts_by_label[clean_label].append(part_path) @@ -264,4 +281,3 @@ def main(argv: Sequence[str] | None = None) -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/copienator/feedback_boxes.py b/copienator/feedback_boxes.py new file mode 100644 index 0000000..c8757a3 --- /dev/null +++ b/copienator/feedback_boxes.py @@ -0,0 +1,13 @@ +"""Geometry checks shared by correction and annotation rendering.""" + +import math +from numbers import Real + + +def valid_feedback_box(box) -> bool: + return ( + isinstance(box, (list, tuple)) and len(box) == 4 + and all(isinstance(value, Real) and not isinstance(value, bool) + and math.isfinite(value) for value in box) + and box[0] < box[2] and box[1] < box[3] + ) diff --git a/copienator/pdf_cut.py b/copienator/pdf_cut.py new file mode 100644 index 0000000..2dada03 --- /dev/null +++ b/copienator/pdf_cut.py @@ -0,0 +1,57 @@ +"""Split a PDF along its cumulative visible page height, in reading order.""" + +import math +from pathlib import Path + +import pymupdf + +from copienator.commands.splitting_int import _prepare_split_pages + +# Percentages emitted by the GUI have six decimal places. Recover exact page +# boundaries despite rounding, without turning nearby in-page cuts into breaks. +BOUNDARY_TOLERANCE_PERCENT = 0.000001 + + +def cut_position(heights: list[float], percent: float) -> tuple[int, float]: + """Return (page index, offset); offset zero denotes an exact page break.""" + if not heights or any(height <= 0 for height in heights): + raise ValueError("Le PDF doit contenir des pages non vides.") + if not math.isfinite(percent) or not 0 < percent < 100: + raise ValueError("Le pourcentage de coupe doit être strictement entre 0 et 100.") + total = sum(heights) + position = total * percent / 100 + start = 0.0 + for index, height in enumerate(heights): + if index and abs(percent - start / total * 100) <= BOUNDARY_TOLERANCE_PERCENT: + return index, 0.0 + end = start + height + if index + 1 < len(heights) and abs(percent - end / total * 100) <= BOUNDARY_TOLERANCE_PERCENT: + return index + 1, 0.0 + if position < end: + return index, position - start + start = end + raise ValueError("Coupe hors du document.") + + +def split_pdf(source: Path, percent: float, first_path: Path, second_path: Path) -> None: + """Preserve whole pages; clip only the page actually crossed by the cut.""" + with pymupdf.open(source) as document, pymupdf.open() as first, pymupdf.open() as second: + index, offset = cut_position([page.rect.height for page in document], percent) + if index: + first.insert_pdf(document, from_page=0, to_page=index - 1) + if offset == 0: + second.insert_pdf(document, from_page=index) + else: + with pymupdf.open() as page_document: + page_document.insert_pdf(document, from_page=index, to_page=index) + visible = _prepare_split_pages(page_document)[0] + for target, clip in ( + (first, pymupdf.Rect(visible.x0, visible.y0, visible.x1, visible.y0 + offset)), + (second, pymupdf.Rect(visible.x0, visible.y0 + offset, visible.x1, visible.y1)), + ): + page = target.new_page(width=clip.width, height=clip.height) + page.show_pdf_page(page.rect, page_document, 0, clip=clip) + if index + 1 < len(document): + second.insert_pdf(document, from_page=index + 1) + first.save(first_path) + second.save(second_path) diff --git a/copienator_gui/app.py b/copienator_gui/app.py index be143c1..92f62c4 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -21,6 +21,9 @@ from copienator.platform import ( ) from .diagnostics import collect_diagnostics +from .batch_monitor import BatchMonitor +from .notifications import notify_desktop +from .manual_resolution import ManualResolutionPanel from .refaire import SECTION as REFAIRE_SECTION from .refaire import ( RefaireSelection, @@ -274,6 +277,10 @@ class CopienatorApp(tk.Tk): self.extra_var = tk.StringVar() self.command_var = tk.StringVar(value="Sélectionnez une étape.") self.info_var = tk.StringVar(value="Choisissez un dossier d’évaluation.") + self.batch_watch_status = tk.StringVar(value="Vérification automatique arrêtée.") + self.batch_monitor = BatchMonitor( + self, self._batch_results_ready, self._batch_watch_status_changed, self._append_console + ) self._build_ui(show_personal_steps) self.extra_var.trace_add("write", lambda *_args: self._update_command_preview()) @@ -543,6 +550,7 @@ class CopienatorApp(tk.Tk): if not evaluation or not evaluation.is_dir(): messagebox.showerror("Dossier invalide", "Choisissez un dossier d’évaluation existant.") return + self.batch_monitor.stop() self._save_current_form() self.state_store.load(evaluation) for step in self.steps: @@ -744,7 +752,7 @@ class CopienatorApp(tk.Tk): return self._rendering = True redo = step.section == REFAIRE_SECTION - expanded_form = redo or step.id in {"page_splitter", "crop_blank_margins", "cutleft", "labels"} + expanded_form = redo or step.id in {"page_splitter", "crop_blank_margins", "cutleft", "labels", "manual_resolution"} self.console.configure(height=8 if expanded_form else 14) self.rowconfigure(1, weight=4 if expanded_form else 3) self.rowconfigure(2, weight=1 if expanded_form else 2) @@ -830,6 +838,18 @@ class CopienatorApp(tk.Tk): attach_tooltip(widget, spec.help) row += 1 + if step.id == "manual_resolution": + def manual_evaluation(): + target = self.arg_vars.get("target") + value = target.get().strip() if target else "" + return (self.repository / value).resolve() if value else self.evaluation + + self.manual_panel = ManualResolutionPanel(self.form, manual_evaluation) + self.manual_panel.grid(row=row, column=0, columnspan=2, sticky="nsew", pady=8) + if "target" in self.arg_vars: + self.arg_vars["target"].trace_add("write", lambda *_args: self.manual_panel.reload()) + row += 1 + show_extra = bool(step.extra_arguments_help) and ( not step.extra_arguments_variants or variant.id in step.extra_arguments_variants @@ -852,6 +872,23 @@ class CopienatorApp(tk.Tk): self._update_controls() def _render_context_controls(self, step: StepDefinition, row: int) -> int: + if step.id == "batch_status": + controls = ttk.Frame(self.form) + controls.grid(row=row, column=0, columnspan=2, sticky="w", pady=8) + self.watch_batches_button = ttk.Button( + controls, text="Vérifier toutes les 5 minutes", command=self._start_batch_watch + ) + self.watch_batches_button.pack(side="left") + self.stop_watch_batches_button = ttk.Button( + controls, text="Arrêter les vérifications", command=self.batch_monitor.stop + ) + self.stop_watch_batches_button.pack(side="left", padx=8) + ttk.Label(self.form, textvariable=self.batch_watch_status).grid( + row=row + 1, column=0, columnspan=2, sticky="w" + ) + ttk.Label(self.form, text="Gardez Copienator ouvert. Une notification de bureau sera envoyée lorsque tous les résultats seront prêts.", + wraplength=650).grid(row=row + 2, column=0, columnspan=2, sticky="w", pady=8) + return row + 3 if step.id == "rename": self.validate_rename_button = ttk.Button( self.form, text="Valider sans renommer", command=self._validate_rename_step @@ -1299,6 +1336,8 @@ class CopienatorApp(tk.Tk): if not step or not evaluation or not self.state_store.evaluation: messagebox.showerror("Évaluation absente", "Chargez d’abord un dossier d’évaluation.") return + if step.id == "batch_status" and self.batch_monitor.active: + return if self.active_step_id or self.runner.running: messagebox.showwarning("Traitement en cours", "Interrompez le traitement actuel avant d’en lancer un autre.") return @@ -1448,6 +1487,39 @@ class CopienatorApp(tk.Tk): self._finish_process(int(return_code), bool(interrupted)) self.after(60, self._poll_runner) + def _batch_watch_status_changed(self, message: str) -> None: + self.batch_watch_status.set(message) + self._update_controls() + + def _start_batch_watch(self) -> None: + if self.active_step_id or self.runner.running or not self.state_store.evaluation: + return + step = self.step_by_id["batch_status"] + command = build_command(self.repository, step, step.variants[0], {}, + str(self.state_store.evaluation)) + environment = build_runner_environment( + os.environ, self.api_key_var.get(), self.proxy_var.get(), self.use_proxy_var.get() + ) + self.batch_monitor.start(command, self.repository, environment, + self.state_store.workspace.log_path("batch_status_watch")) + + def _batch_results_ready(self) -> None: + evaluation = self.state_store.evaluation + if evaluation is None: + return + self.state_store.update_step("batch_status", status="success", return_code=0) + self._populate_tree() + message = f"{evaluation.name} : tous les résultats batch sont prêts à récupérer." + self.info_var.set(message) + try: + notify_desktop("Copienator — batchs terminés", message) + except (OSError, RuntimeError) as exc: + self.bell() + self._append_console(f"Notification de bureau indisponible : {exc}\n{message}\n") + if (self.current_step and self.current_step.id == "batch_status" + and not self.active_step_id and not self.runner.running): + self._move_selection_from("batch_status", 1) + def _finish_process(self, return_code: int, interrupted: bool) -> None: step_id = self.active_step_id if not step_id: @@ -1509,6 +1581,10 @@ class CopienatorApp(tk.Tk): self._populate_tree() self._update_controls() if status == "success": + if step_id == "batch_status" and self.batch_monitor.active: + self.batch_monitor.stop() + self._batch_results_ready() + return self.after_idle(lambda completed_id=step_id: self._move_selection_from(completed_id, 1)) def _send_input(self) -> None: @@ -1552,13 +1628,18 @@ class CopienatorApp(tk.Tk): def _update_controls(self) -> None: running = bool(self.active_step_id) or self.runner.running + if self.current_step and self.current_step.id == "batch_status": + self.watch_batches_button.configure(state="disabled" if running or self.batch_monitor.active or not self.state_store.evaluation else "normal") + self.stop_watch_batches_button.configure(state="normal" if self.batch_monitor.active else "disabled") if running and self.current_step and self.current_step.id in {"page_splitter", "cutleft"}: self.marked_copies_button.configure(state="disabled") if self.current_step and self.current_step.id == "rename": self.validate_rename_button.configure( state="disabled" if running or not self.state_store.evaluation else "normal" ) - self.run_button.configure(state="disabled" if running or not self.current_step else "normal") + watching_current = bool(self.current_step and self.current_step.id == "batch_status" + and self.batch_monitor.active) + self.run_button.configure(state="disabled" if running or not self.current_step or watching_current else "normal") self.skip_button.configure(state="normal" if not running and self.current_step and self.current_step.optional else "disabled") self.interrupt_button.configure(state="normal" if running else "disabled") self.force_button.configure(state="normal" if running else "disabled") @@ -1593,4 +1674,5 @@ class CopienatorApp(tk.Tk): except OSError: pass self._save_current_form() + self.batch_monitor.stop() self.destroy() diff --git a/copienator_gui/batch_monitor.py b/copienator_gui/batch_monitor.py new file mode 100644 index 0000000..2ab685f --- /dev/null +++ b/copienator_gui/batch_monitor.py @@ -0,0 +1,81 @@ +"""Non-blocking, cancellable batch checks while the desktop GUI is open.""" + +import queue + +from .runner import ProcessRunner + +CHECK_INTERVAL_MS = 5 * 60 * 1000 + + +class BatchMonitor: + def __init__(self, scheduler, on_ready, on_status, on_output): + self.scheduler = scheduler + self.on_ready = on_ready + self.on_status = on_status + self.on_output = on_output + self.active = False + self.timer = None + self.runner = None + + def start(self, command, cwd, environment, log_path): + if self.active: + return + self.arguments = (command, cwd, environment, log_path) + self.active = True + self._check() + + def stop(self): + self.active = False + if self.timer is not None: + self.scheduler.after_cancel(self.timer) + self.timer = None + if self.runner is not None: + try: + self.runner.force_stop() + except OSError as exc: + self.on_output(f"Arrêt de la vérification : {exc}\n") + self.runner = None + self.on_status("Vérification automatique arrêtée.") + + def _later(self): + self.timer = self.scheduler.after(CHECK_INTERVAL_MS, self._check) + + def _check(self): + self.timer = None + if not self.active: + return + self.runner = ProcessRunner() + self.on_status("Vérification des batchs en cours…") + try: + self.runner.start(*self.arguments) + except (OSError, RuntimeError) as exc: + self.on_output(f"Vérification impossible : {exc}\n") + self.runner = None + self.on_status("Échec de la vérification. Nouvel essai dans 5 minutes.") + self._later() + return + self.timer = self.scheduler.after(100, self._poll) + + def _poll(self): + self.timer = None + if not self.active or self.runner is None: + return + while True: + try: + event, payload = self.runner.events.get_nowait() + except queue.Empty: + break + if event in {"output", "runner_error"}: + self.on_output(str(payload)) + elif event == "finished": + code, interrupted = payload + self.runner = None + if code == 0 and not interrupted: + self.active = False + self.on_status("Tous les résultats batch sont prêts.") + self.on_ready() + else: + self.on_status("Résultats pas encore prêts. Nouvelle vérification dans 5 minutes.") + self._later() + return + self.timer = self.scheduler.after(100, self._poll) diff --git a/copienator_gui/cut_helper.py b/copienator_gui/cut_helper.py new file mode 100644 index 0000000..10ea2f1 --- /dev/null +++ b/copienator_gui/cut_helper.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import tkinter as tk +from pathlib import Path +from tkinter import messagebox, ttk + +import pymupdf +from PIL import Image, ImageTk + +from copienator.pdf_cut import cut_position + +PAGE_GAP = 28 +SNAP_PIXELS = 12 + + +def percentage_at_y(y: float, heights: list[float], scale: float) -> float: + """Convert canvas y to document height, snapping across inter-page gaps.""" + top = 0.0 + cumulative = 0.0 + total = sum(heights) + for index, height in enumerate(heights): + bottom = top + height * scale + if index + 1 < len(heights) and bottom - SNAP_PIXELS <= y <= bottom + PAGE_GAP + SNAP_PIXELS: + return (cumulative + height) / total * 100 + if y <= bottom: + return max(0.0, min(100.0, (cumulative + (y - top) / scale) / total * 100)) + cumulative += height + top = bottom + PAGE_GAP + return 100.0 + + +def y_at_percentage(percent: float, heights: list[float], scale: float) -> float: + if percent <= 0: + return 0.0 + if percent >= 100: + return sum(heights) * scale + (len(heights) - 1) * PAGE_GAP + index, offset = cut_position(heights, percent) + top = sum(heights[:index]) * scale + index * PAGE_GAP + return top - PAGE_GAP / 2 if index and offset == 0 else top + offset * scale + + +def cut_operator(percent: float, keep: int, mode: str) -> str: + value = f"{percent:.6f}".rstrip("0").rstrip(".") + return f"c{{{value}}}{keep}{mode}" + + +class CutHelper(tk.Toplevel): + def __init__(self, parent, path: Path, on_accept, initial=None): + # Load the source before creating a window so invalid PDFs leave no dialog. + with pymupdf.open(path) as document: + if not len(document): + raise ValueError("Le PDF est vide.") + self.heights = [page.rect.height for page in document] + width = min(850, parent.winfo_screenwidth() - 100) + self.scale = min(1.5, width / max(page.rect.width for page in document)) + rendered = [] + for page in document: + pix = page.get_pixmap(matrix=pymupdf.Matrix(self.scale, self.scale), alpha=False, + colorspace=pymupdf.csRGB) + rendered.append(Image.frombytes("RGB", (pix.width, pix.height), pix.samples)) + super().__init__(parent) + self.title(f"Cut — {path.name}") + self.geometry(f"{width + 45}x{min(850, parent.winfo_screenheight() - 100)}") + self.transient(parent.winfo_toplevel()) + self.on_accept = on_accept + self.percent = initial[0] if initial else 50.0 + self.keep = tk.IntVar(value=initial[1] if initial else 1) + self.mode = tk.StringVar(value=initial[2] if initial else ">") + self.caption = tk.StringVar() + ttk.Label(self, text="Déplacez la barre rouge. Entrée : afficher la commande ; Échap : annuler.", + wraplength=width).pack(anchor="w", padx=8, pady=5) + controls = ttk.Frame(self) + controls.pack(fill="x", padx=8) + ttk.Label(controls, text="Conserver à la source :").pack(side="left") + ttk.Radiobutton(controls, text="1 — début", variable=self.keep, value=1).pack(side="left") + ttk.Radiobutton(controls, text="2 — fin", variable=self.keep, value=2).pack(side="left") + ttk.Radiobutton(controls, text="Ajouter à la cible", variable=self.mode, value=">").pack(side="left") + ttk.Radiobutton(controls, text="Remplacer", variable=self.mode, value="x").pack(side="left") + ttk.Label(self, textvariable=self.caption).pack(anchor="w", padx=8, pady=5) + viewport = ttk.Frame(self) + viewport.pack(fill="both", expand=True) + self.canvas = tk.Canvas(viewport, background="#555555", highlightthickness=0) + scrollbar = ttk.Scrollbar(viewport, command=self.canvas.yview) + self.canvas.configure(yscrollcommand=scrollbar.set) + scrollbar.pack(side="right", fill="y") + self.canvas.pack(fill="both", expand=True) + self.images = [ImageTk.PhotoImage(image, master=self) for image in rendered] + top = 0.0 + self.width = width + for index, image in enumerate(self.images): + self.canvas.create_image(0, top, anchor="nw", image=image) + top += self.heights[index] * self.scale + if index + 1 < len(self.images): + top += PAGE_GAP + self.canvas.configure(scrollregion=(0, 0, width, top)) + self.bar = self.canvas.create_line(0, 0, width, 0, fill="#ff3030", width=4) + self.canvas.bind("", self.move_bar) + self.canvas.bind("", self.move_bar) + self.canvas.bind("", lambda event: self.canvas.yview_scroll(-3, "units")) + self.canvas.bind("", lambda event: self.canvas.yview_scroll(3, "units")) + self.canvas.bind("", lambda event: self.canvas.yview_scroll(-1 if event.delta > 0 else 1, "units")) + self.bind("", self.accept) + self.bind("", lambda event: self.destroy()) + self.keep.trace_add("write", lambda *_: self.draw_bar()) + self.mode.trace_add("write", lambda *_: self.draw_bar()) + self.draw_bar() + self.canvas.yview_moveto(max(0, (y_at_percentage(self.percent, self.heights, self.scale) - 200) / top)) + self.focus_set() + self.grab_set() + + def move_bar(self, event): + self.percent = percentage_at_y(self.canvas.canvasy(event.y), self.heights, self.scale) + self.draw_bar() + + def draw_bar(self): + y = y_at_percentage(self.percent, self.heights, self.scale) + self.canvas.coords(self.bar, 0, y, self.width, y) + text = cut_operator(self.percent, self.keep.get(), self.mode.get()) + if 0 < self.percent < 100: + index, offset = cut_position(self.heights, self.percent) + text += f" — entre les pages {index} et {index + 1}" if offset == 0 else f" — page {index + 1}" + self.caption.set(text) + + def accept(self, event=None): + operator = cut_operator(self.percent, self.keep.get(), self.mode.get()) + rounded = float(operator.split("{")[1].split("}")[0]) + if not 0 < rounded < 100: + messagebox.showerror("Coupe invalide", "Chaque partie doit contenir une portion du PDF.", parent=self) + return + self.destroy() + self.on_accept(operator) diff --git a/copienator_gui/manual_resolution.py b/copienator_gui/manual_resolution.py new file mode 100644 index 0000000..a92d96d --- /dev/null +++ b/copienator_gui/manual_resolution.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import tkinter as tk +from pathlib import Path +from tkinter import messagebox, ttk + +from copienator import CliError +from copienator.commands.resolve_manual import get_actual_pdf, parse_instruction_text +from copienator.platform import open_path +from .cut_helper import CutHelper + + +HELP = """La correction propose x> pour un mauvais label et -> pour une réponse supplémentaire lorsque le PDF cible existe déjà. Vérifiez les deux PDF avant de choisir. + +Format : Copie01 Label source OP Label cible| +Conservez Copie suivi du numéro et les labels exacts (sans .pdf). Les espaces dans les labels sont acceptés ; entourez l’opérateur d’espaces. + +-> : fusionner dans la cible, conserver la source. +x> : fusionner dans la cible, archiver la source. +-x : remplacer la cible par une copie de la source, conserver la source. +xx : remplacer la cible par une copie de la source, archiver la source. +ss : conserver les deux PDF sans fusion. +sx : conserver la source, archiver la cible, sans fusion. +xs : archiver la source, conserver la cible, sans fusion. + +c{43}1> : couper la source à 43 %, conserver le début et ajouter la fin à la cible. +c{43}2x : couper la source à 43 %, conserver la fin et remplacer la cible par le début. +1 conserve le début, 2 conserve la fin ; > fusionne l’autre partie avec la cible, x la remplace. Le pourcentage porte sur la hauteur cumulée des pages visibles. Les décimales sont acceptées. Une seule coupe par label source est autorisée. +Cut permet de placer la coupe, avec accrochage entre les pages. Entrée ferme l’aperçu et affiche la commande à recopier ci-dessus : aucun fichier n’est modifié. Une séparation entre pages conserve les pages entières. La source complète est archivée en _old ; les deux labels modifiés sont générés en _new et ajoutés à refaire.json. + +Pour les fusions : « Source x> Cible| » place la cible avant la source ; « Source x> |Cible » place la source avant la cible. Même règle avec -> et c{…}1>/c{…}2> (pour la partie transférée) ; sans |, la cible vient en premier. +Vous pouvez changer l’opérateur, déplacer |, corriger les labels ou retirer une instruction. Les lignes vides et celles commençant par ### sont ignorées. Retirer/commenter une ligne ne résout pas son conflit. + +Enregistrez dans l’éditeur, puis cliquez sur Recharger et enfin sur Exécuter. Seul le fichier enregistré est appliqué. L’archivage utilise le suffixe _old ; les PDF créés utilisent _new. Après succès, manual_resolutions.txt est supprimé et correction.json est mis à jour. Si des PDF sont créés, refaire.json est généré : relancez la correction avec --refaire.""" + + +class ManualResolutionPanel(ttk.Frame): + def __init__(self, parent, get_evaluation): + super().__init__(parent) + self.get_evaluation = get_evaluation + self.pdf_buttons = [] + self.cut_buttons = [] + actions = ttk.Frame(self) + actions.pack(fill="x") + self.editor_button = ttk.Button( + actions, text="Ouvrir dans un éditeur de texte", command=self.open_editor + ) + self.editor_button.pack(side="left") + ttk.Button(actions, text="Recharger", command=self.reload).pack(side="left", padx=6) + self.cut_result = tk.StringVar() + result = ttk.Frame(self) + result.pack(fill="x", pady=4) + ttk.Entry(result, textvariable=self.cut_result, state="readonly").pack(side="left", fill="x", expand=True) + ttk.Button(result, text="Copier la commande Cut", command=self.copy_cut_command).pack(side="left", padx=6) + self.status = ttk.Label(self, wraplength=650) + self.status.pack(fill="x", pady=4) + preview = ttk.Frame(self) + preview.pack(fill="both", expand=True) + self.text = tk.Text(preview, height=10, width=50, wrap="word", state="disabled") + scroll = ttk.Scrollbar(preview, command=self.text.yview) + self.text.configure(yscrollcommand=scroll.set) + scroll.pack(side="right", fill="y") + self.text.pack(fill="both", expand=True) + ttk.Label(self, text=HELP, wraplength=650, justify="left").pack(fill="x", pady=8) + self.reload() + + def manual_path(self) -> Path | None: + evaluation = self.get_evaluation() + return evaluation / "manual_resolutions.txt" if evaluation else None + + def open_editor(self): + path = self.manual_path() + if path is None or not path.is_file(): + self.reload() + return + try: + # .txt is opened in the desktop's associated text editor. + open_path(path) + except (OSError, RuntimeError) as exc: + messagebox.showerror("Ouverture impossible", str(exc)) + + def open_pdf(self, evaluation, copy_id, label): + path = get_actual_pdf(evaluation / "Copies", copy_id, label) + try: + if not path.is_file(): + raise FileNotFoundError(f"PDF introuvable : {path}") + open_path(path) + except (OSError, RuntimeError) as exc: + messagebox.showerror("Ouverture du PDF", str(exc)) + + def reload(self): + for button in self.pdf_buttons + self.cut_buttons: + button.destroy() + self.pdf_buttons.clear() + self.cut_buttons.clear() + self.cut_result.set("") + self.text.configure(state="normal") + self.text.delete("1.0", "end") + path = self.manual_path() + self.editor_button.configure(state="disabled") + try: + if path is None: + raise FileNotFoundError("Chargez une évaluation.") + content = path.read_text(encoding="utf-8") + except (OSError, UnicodeError) as exc: + self.status.configure(text=f"manual_resolutions.txt indisponible : {exc}") + else: + self.editor_button.configure(state="normal") + malformed = [] + for number, raw in enumerate(content.splitlines(keepends=True), 1): + self.text.insert("end", raw.rstrip("\r\n")) + try: + instructions = parse_instruction_text(raw) + except CliError: + malformed.append(str(number)) + else: + if instructions: + instruction = instructions[0] + for title, label in (("PDF source", instruction.old_label), + ("PDF cible", instruction.new_label)): + button = ttk.Button( + self.text, text=title, + command=lambda label=label, copy_id=instruction.copy_id, root=path.parent: self.open_pdf(root, copy_id, label), + ) + self.pdf_buttons.append(button) + self.text.window_create("end", window=button, padx=8) + button = ttk.Button(self.text, text="Cut", + command=lambda item=instruction, root=path.parent: self.open_cut(root, item)) + self.cut_buttons.append(button) + self.text.window_create("end", window=button, padx=8) + if raw.endswith("\n"): + self.text.insert("end", "\n") + detail = (" — lignes invalides : " + ", ".join(malformed)) if malformed else ( + f" — {len(self.pdf_buttons) // 2} instruction(s)" + ) + self.status.configure(text=str(path) + detail) + finally: + self.text.configure(state="disabled") + + def copy_cut_command(self): + if self.cut_result.get(): + self.clipboard_clear() + self.clipboard_append(self.cut_result.get()) + + def open_cut(self, evaluation, instruction): + source = get_actual_pdf(evaluation / "Copies", instruction.copy_id, instruction.old_label) + def accepted(operator): + target = ("|" + instruction.new_label) if instruction.pipe_first else (instruction.new_label + "|") + self.cut_result.set(f"Copie{instruction.copy_id} {instruction.old_label} {operator} {target}") + try: + initial = (*instruction.cut, instruction.operator[-1]) if instruction.cut else None + CutHelper(self, source, accepted, initial) + except (OSError, RuntimeError, ValueError) as exc: + messagebox.showerror("Ouverture du PDF", str(exc)) diff --git a/copienator_gui/notifications.py b/copienator_gui/notifications.py new file mode 100644 index 0000000..b972cb1 --- /dev/null +++ b/copienator_gui/notifications.py @@ -0,0 +1,37 @@ +"""Launch native desktop notifications without blocking Tk.""" + +import base64 +import json +import os +import subprocess +import sys + +from copienator.platform import find_executable + + +def notify_desktop(title: str, message: str) -> None: + if sys.platform.startswith("linux"): + executable = find_executable("notify-send") + if not executable: + raise RuntimeError("Installez notify-send (libnotify) pour les notifications de bureau.") + command = [executable, "--app-name=Copienator", "--", title, message] + elif sys.platform == "darwin": + command = ["/usr/bin/osascript", "-e", + f"display notification {json.dumps(message, ensure_ascii=False)} with title {json.dumps(title, ensure_ascii=False)}"] + elif os.name == "nt": + # Encode the script and quote strings as literals; no shell interpolation. + quote = lambda value: "'" + value.replace("'", "''") + "'" + script = ( + "Add-Type -AssemblyName System.Windows.Forms;" + "$notice = New-Object System.Windows.Forms.NotifyIcon;" + "$notice.Icon = [System.Drawing.SystemIcons]::Information;" + "$notice.Visible = $true;" + f"$notice.ShowBalloonTip(10000, {quote(title)}, {quote(message)}, " + "[System.Windows.Forms.ToolTipIcon]::Info);" + "Start-Sleep -Seconds 12; $notice.Dispose()" + ) + command = ["powershell.exe", "-NoProfile", "-NonInteractive", "-WindowStyle", "Hidden", + "-EncodedCommand", base64.b64encode(script.encode("utf-16-le")).decode("ascii")] + else: + raise RuntimeError("Notifications de bureau indisponibles sur ce système.") + subprocess.Popen(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index 6bb75f8..d5a9378 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -352,7 +352,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: "bool", "--overwrite", help="Relance les corrections demandées même lorsqu’un résultat existe déjà.", - variants=("live", "batch", "hybrid", "refaire"), + variants=("live", "batch", "hybrid", "refaire", "integrate"), ), ArgumentSpec( "limit", @@ -394,24 +394,8 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: "batch_status", "Correction", "Consulter l’état des batchs", - "Affiche les jobs Gemini en cours. L’identifiant de téléchargement est facultatif.", + "Vérifie les jobs enregistrés pour l’évaluation. Passe à la récupération uniquement lorsque tous ont réussi et que leurs résultats sont disponibles ; sinon, reste sur cette étape.", (python("default", "État des batchs", "batch-status"),), - arguments=( - ArgumentSpec( - "download", - "Télécharger le job", - "text", - "--download", - help="Saisissez l’identifiant complet d’un job Gemini terminé pour télécharger son fichier de résultats.", - ), - ArgumentSpec( - "output", - "Fichier JSONL de destination", - "path", - "--output", - help="Choisissez le fichier JSONL dans lequel enregistrer le job téléchargé. Ce champ n’est utilisé que si un identifiant de job est fourni.", - ), - ), optional=True, skip_for_live_correction=True, ), @@ -429,7 +413,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: "post_correction", "Correction", "Nettoyer la correction", - "Corrige certains problèmes d’encodage et prépare le texte pour LaTeX.", + "Sauvegarde correction.json dans correction_precleanup.json, puis corrige certains problèmes d’encodage et prépare le texte pour LaTeX. La sauvegarde est remplacée à chaque nettoyage.", (python("default", "Post-correction", "post-correction"),), arguments=(arg_target("le dossier de l’évaluation"),), requires=("correction.json",), @@ -788,6 +772,9 @@ def build_command( positionals: list[str] = [] options: list[str] = [] + if step.id == "batch_status": + # Use the loaded evaluation without introducing another input field. + options.extend(("--evaluation", evaluation_arg)) for spec in step.arguments: if spec.variants and variant.id not in spec.variants: continue diff --git a/tests/test_batch_monitor.py b/tests/test_batch_monitor.py new file mode 100644 index 0000000..415c007 --- /dev/null +++ b/tests/test_batch_monitor.py @@ -0,0 +1,69 @@ +import queue +import unittest +from unittest.mock import Mock, patch + +from copienator_gui.batch_monitor import BatchMonitor, CHECK_INTERVAL_MS +from copienator_gui.notifications import notify_desktop + + +class BatchMonitorTests(unittest.TestCase): + def setUp(self): + self.scheduler = Mock() + self.scheduler.after.side_effect = lambda delay, callback: (delay, callback) + self.ready = Mock() + self.status = Mock() + self.monitor = BatchMonitor(self.scheduler, self.ready, self.status, Mock()) + self.runner = Mock() + self.runner.events = queue.Queue() + self.factory = patch("copienator_gui.batch_monitor.ProcessRunner", return_value=self.runner) + self.factory.start() + self.addCleanup(self.factory.stop) + + def start(self): + self.monitor.start(["check"], "/tmp", {}, None) + + def test_checks_immediately_retries_in_five_minutes_and_notifies_once(self): + self.start() + self.runner.start.assert_called_once() + self.start() + self.runner.start.assert_called_once() + self.runner.events.put(("finished", (4, False))) + self.monitor._poll() + self.assertEqual(self.monitor.timer[0], CHECK_INTERVAL_MS) + self.assertEqual(CHECK_INTERVAL_MS, 300000) + self.ready.assert_not_called() + self.monitor.timer[1]() + self.assertEqual(self.runner.start.call_count, 2) + self.runner.events.put(("finished", (0, False))) + self.monitor._poll() + self.assertFalse(self.monitor.active) + self.assertIsNone(self.monitor.timer) + self.ready.assert_called_once() + self.monitor._poll() + self.ready.assert_called_once() + + def test_stop_cancels_timer_and_running_check_without_notification(self): + self.start() + timer = self.monitor.timer + self.monitor.stop() + self.scheduler.after_cancel.assert_called_once_with(timer) + self.runner.force_stop.assert_called_once() + self.runner.events.put(("finished", (0, False))) + self.monitor._poll() + self.monitor._check() + self.ready.assert_not_called() + self.runner.start.assert_called_once() + + def test_start_failure_retries_without_reporting_readiness(self): + self.runner.start.side_effect = OSError("unavailable") + self.start() + self.assertEqual(self.monitor.timer[0], CHECK_INTERVAL_MS) + self.ready.assert_not_called() + + def test_linux_notification_passes_text_as_arguments(self): + with patch("copienator_gui.notifications.sys.platform", "linux"), patch( + "copienator_gui.notifications.find_executable", return_value="/usr/bin/notify-send" + ), patch("copienator_gui.notifications.subprocess.Popen") as launch: + notify_desktop("Copienator", "Interro02 : résultats prêts") + self.assertEqual(launch.call_args.args[0], ["/usr/bin/notify-send", "--app-name=Copienator", + "--", "Copienator", "Interro02 : résultats prêts"]) diff --git a/tests/test_batch_status_readiness.py b/tests/test_batch_status_readiness.py new file mode 100644 index 0000000..588cde5 --- /dev/null +++ b/tests/test_batch_status_readiness.py @@ -0,0 +1,54 @@ +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, call, patch + +from copienator import CliError, EvaluationWorkspace, ExitCode, atomic_write_json +from copienator.commands.batch_status import check_evaluation_jobs, main + + +class BatchReadinessTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.workspace = EvaluationWorkspace(Path(self.temp.name)) + self.client = Mock() + + def manifest(self, jobs): + atomic_write_json(self.workspace.batch_jobs_file, {"jobs": jobs}) + + def test_only_recorded_jobs_are_checked_and_all_must_have_results(self): + self.manifest({"flash": {"name": "batches/flash"}, "pro": {"name": "batches/pro"}}) + succeeded = SimpleNamespace(state="JOB_STATE_SUCCEEDED", + dest=SimpleNamespace(file_name="files/result")) + for state in ("JOB_STATE_PENDING", "JOB_STATE_RUNNING", "JOB_STATE_FAILED", + "JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED", "UNKNOWN", "JOB_STATE_SUCCEEDED"): + with self.subTest(state=state): + self.client.reset_mock() + self.client.batches.get.side_effect = [succeeded, SimpleNamespace( + state=SimpleNamespace(name=state), dest=SimpleNamespace(file_name="files/pro"))] + result = check_evaluation_jobs(self.workspace, client=self.client) + self.assertEqual(result, ExitCode.SUCCESS if state == "JOB_STATE_SUCCEEDED" else ExitCode.PARTIAL) + self.assertEqual(self.client.batches.get.call_args_list, + [call(name="batches/flash"), call(name="batches/pro")]) + self.client.batches.list.assert_not_called() + self.client.files.download.assert_not_called() + self.client.batches.get.side_effect = [succeeded, SimpleNamespace( + state="JOB_STATE_SUCCEEDED", dest=None)] + self.assertEqual(check_evaluation_jobs(self.workspace, client=self.client), ExitCode.PARTIAL) + + def test_missing_empty_and_invalid_manifest_cannot_report_success(self): + self.assertEqual(check_evaluation_jobs(self.workspace, client=self.client), ExitCode.PARTIAL) + self.manifest({}) + self.assertEqual(check_evaluation_jobs(self.workspace, client=self.client), ExitCode.PARTIAL) + for jobs in ([], {"flash": {}}, {"flash": {"name": ""}}, {"flash": None}): + self.manifest(jobs) + with self.assertRaises(CliError): + check_evaluation_jobs(self.workspace, client=self.client) + self.client.batches.get.assert_not_called() + + def test_cli_returns_readiness_code_for_selected_evaluation(self): + with patch("copienator.commands.batch_status.check_evaluation_jobs", return_value=ExitCode.PARTIAL) as check: + self.assertEqual(main(["--evaluation", str(self.workspace.root)]), ExitCode.PARTIAL) + self.assertEqual(check.call_args.args[0].root, self.workspace.root) diff --git a/tests/test_feedback_boxes.py b/tests/test_feedback_boxes.py new file mode 100644 index 0000000..e897ac1 --- /dev/null +++ b/tests/test_feedback_boxes.py @@ -0,0 +1,82 @@ +import copy +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from PIL import Image + +from copienator import EvaluationWorkspace, atomic_write_json +from copienator.annotation_data import GroupCoordinates, _scaled_result +from copienator.annotation_actions import apply_checkbox_actions +from copienator.commands import annotating, correction +from copienator.feedback_boxes import valid_feedback_box + + +class FeedbackBoxTests(unittest.TestCase): + def test_checkbox_for_promoted_feedback_deletes_the_correct_comment(self): + feedback = [{"text": "Invalid local", "box_2d": [753, 680, 287, 946]}, + {"text": "Existing global", "box_2d": None}] + apply_checkbox_actions({"Ex 5": {"result": {"feedback": feedback}}}, + [{"label": "Ex 5", "type": "del_global", "index": 0}], lambda _: None) + self.assertTrue(feedback[0]["to_delete"]) + self.assertNotIn("to_delete", feedback[1]) + + def test_bad_boxes_keep_their_comment_as_global_feedback_without_mutating_data(self): + for box in ([753, 680, 287, 946], [10, 50, 20, 30], [10, 20, 10, 30], + [1, 2, 3], [None, 0, 10, 20], [float("nan"), 0, 10, 20]): + with self.subTest(box=box): + result = {"score": 2, "feedback": [{"text": "Important comment", "box_2d": box}]} + original_box = result["feedback"][0]["box_2d"] + scaled = _scaled_result(result, GroupCoordinates(2415, 3297, 1655, 3297)) + callback = Mock() + render = Mock(return_value=Image.new("RGBA", (200, 40), "white")) + with patch.object(annotating, "render_score_text", return_value=Image.new("RGBA", (200, 40))): + image, _ = annotating.compose_label_image(Image.new("RGBA", (800, 100)), "Ex 5", scaled, + 2415, render_fn=render, draw_callback=callback) + self.assertIsNotNone(image) + render.assert_called_once_with("Important comment", unittest.mock.ANY) + self.assertFalse(any(call.args[0] == "local_rect" for call in callback.call_args_list)) + self.assertIs(result["feedback"][0]["box_2d"], original_box) + + def test_valid_boxes_remain_local(self): + result = {"feedback": [{"text": "Local", "box_2d": [10, 20, 30, 40]}]} + before = copy.deepcopy(result) + callback = Mock() + with patch.object(annotating, "render_score_text", return_value=Image.new("RGBA", (200, 40))): + annotating.compose_label_image(Image.new("RGBA", (800, 100)), "Ex 5", result, 0, + render_fn=Mock(return_value=Image.new("RGBA", (200, 40))), + draw_callback=callback) + self.assertTrue(any(call.args[0] == "local_rect" for call in callback.call_args_list)) + self.assertEqual(result, before) + self.assertTrue(valid_feedback_box([10, 20, 30, 40])) + + def test_invalid_auxiliary_response_loses_only_its_rectangle(self): + returned = [{"text": "Keep this", "box_2d": [753, 680, 287, 946]}] + with patch.object(correction.prompting, "request_for_box_correction", return_value=([], {})), patch.object( + correction, "call_gemini_with_retries", return_value=json.dumps(returned) + ): + feedback = correction.correct_boxes_with_gemini("26", "Ex 5", Path("unused.pdf"), [], 0, 1000, 1, 1000) + self.assertEqual(feedback, [{"text": "Keep this", "box_2d": None}]) + + def test_correction_requests_repair_for_inverted_boxes_and_falls_back_without_losing_comments(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + group = root / "Par label" / "Ex 5" / "Group_1.jpg" + group.parent.mkdir(parents=True) + group.touch() + atomic_write_json(group.with_suffix(".json"), [["26", 0, 1000, 1, "Ex 5"]]) + args = correction.build_parser().parse_args([str(root)]) + correction.configure_runtime(EvaluationWorkspace(root), [(str(group), "Ex 5")], args, api_client=Mock()) + response = [{"id": "26", "result": {"score": 2, "error": "", "feedback": [ + {"text": "First", "box_2d": [753, 680, 287, 946]}, + {"text": "Second", "box_2d": [100, 900, 200, 100]}]}}] + with patch.object(correction.prompting, "generate_request", return_value=([], {})), patch.object( + correction, "correct_boxes_with_gemini", side_effect=RuntimeError("repair failed") + ) as repair: + correction.process_single_task((str(group), "Ex 5"), json.dumps(response)) + repair.assert_called_once() + feedback = correction.results["Ex 5"][0][0]["result"]["feedback"] + self.assertEqual([f["text"] for f in feedback], ["First", "Second"]) + self.assertTrue(all(f["box_2d"] is None for f in feedback)) diff --git a/tests/test_gui_convenience.py b/tests/test_gui_convenience.py index 8211830..8d70d18 100644 --- a/tests/test_gui_convenience.py +++ b/tests/test_gui_convenience.py @@ -68,6 +68,66 @@ class GuiConvenienceTests(unittest.TestCase): self.app.open_evaluation_button.invoke() opened.assert_called_once_with(self.evaluation) + def test_manual_resolution_panel_follows_command_target(self): + manual = self.evaluation / "manual_resolutions.txt" + manual.write_text("Copie01 A -> B|\n", encoding="utf-8") + self.app.tree.selection_set("manual_resolution") + self.app.update() + self.assertEqual(len(self.app.manual_panel.pdf_buttons), 2) + other = self.repository / "Other" + other.mkdir() + other_manual = other / "manual_resolutions.txt" + other_manual.write_text("Copie02 C xs D\n", encoding="utf-8") + self.app.arg_vars["target"].set(str(other)) + self.app.update() + self.assertIn("Copie02", self.app.manual_panel.text.get("1.0", "end")) + with patch("copienator_gui.manual_resolution.open_path") as opened: + self.app.manual_panel.editor_button.invoke() + opened.assert_called_once_with(other_manual) + + def test_batch_status_waits_until_all_jobs_are_ready(self): + self.app.state_store.update_step("correction", variant="batch") + self.app.state_store.update_step("batch_status", visited=True) + self.app.tree.selection_set("batch_status") + self.app.update() + command = self.app._make_command() + self.assertIn("--evaluation", command) + self.assertFalse(self.app.arg_vars) + self.app.active_step_id = "batch_status" + self.app._finish_process(4, False) + self.app.update() + self.assertEqual(self.app.current_step.id, "batch_status") + self.app.active_step_id = "batch_status" + self.app._finish_process(0, False) + self.app.update() + self.assertEqual(self.app.current_step.id, "fetch_batches") + + def test_batch_watch_buttons_use_loaded_evaluation_and_stop_on_reload(self): + self.app.state_store.update_step("correction", variant="batch") + self.app.tree.selection_set("batch_status") + self.app.update() + with patch.object(self.app.batch_monitor, "start") as start: + self.app.watch_batches_button.invoke() + self.assertEqual(start.call_args.args[0][-2:], ["--evaluation", str(self.evaluation)]) + self.app.batch_monitor.active = True + self.app._update_controls() + self.assertEqual(str(self.app.stop_watch_batches_button.cget("state")), "normal") + self.app.stop_watch_batches_button.invoke() + self.assertFalse(self.app.batch_monitor.active) + self.app.batch_monitor.active = True + self.app._load_evaluation() + self.assertFalse(self.app.batch_monitor.active) + + def test_background_batch_readiness_notifies_without_changing_other_step(self): + self.app.tree.selection_set("inputs") + self.app.update() + with patch("copienator_gui.app.notify_desktop") as notify: + self.app._batch_results_ready() + notify.assert_called_once() + self.assertIn(self.evaluation.name, notify.call_args.args[1]) + self.assertEqual(self.app.current_step.id, "inputs") + self.assertEqual(self.app.state_store.step("batch_status")["status"], "success") + def test_optional_blank_crop_can_target_a_copy_or_be_skipped(self): copies = self.evaluation/"Copies" copies.mkdir() diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 6a7088f..b068599 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -1340,11 +1340,20 @@ class StandardCliTests(unittest.TestCase): evaluation / "correction.json", {"text": "outside_name and $math_name$", "suffix": "_new"}, ) + original = (evaluation / "correction.json").read_bytes() + backup = evaluation / "correction_precleanup.json" + backup.write_bytes(b"previous backup") self.assertEqual(module.main([str(evaluation)]), 0) + self.assertEqual(backup.read_bytes(), original) self.assertEqual( read_json(evaluation / "correction.json"), {"text": r"outside\_name and $math_name$", "suffix": "_new"}, ) + cleaned = (evaluation / "correction.json").read_bytes() + with patch.object(module, "atomic_write_bytes", side_effect=OSError("backup failed")): + self.assertNotEqual(module.main([str(evaluation)]), 0) + self.assertEqual((evaluation / "correction.json").read_bytes(), cleaned) + self.assertEqual(backup.read_bytes(), original) def test_manual_resolution_with_no_actions_is_safe(self) -> None: module = self.modules["resolve_manual"] @@ -1891,7 +1900,7 @@ class WorkflowTests(unittest.TestCase): ) self.assertEqual( command_arguments(status), - ["--download", "files/job-1", "--output", "/tmp/result.jsonl"], + ["--evaluation", self.evaluation], ) grouped = self.command( diff --git a/tests/test_gui_cut_helper.py b/tests/test_gui_cut_helper.py new file mode 100644 index 0000000..e7ae647 --- /dev/null +++ b/tests/test_gui_cut_helper.py @@ -0,0 +1,56 @@ +import os +import tempfile +import tkinter as tk +import unittest +from pathlib import Path +from types import SimpleNamespace + +import pymupdf + +from copienator_gui.cut_helper import CutHelper, PAGE_GAP +from copienator_gui.manual_resolution import ManualResolutionPanel + + +@unittest.skipUnless(os.environ.get("DISPLAY"), "requires a display") +class CutHelperTests(unittest.TestCase): + def test_cut_button_drag_to_page_gap_and_enter_only_produces_command(self): + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + source = directory / "Copies" / "Copie16" / "A.pdf" + source.parent.mkdir(parents=True) + with pymupdf.open() as document: + for height in (200, 400): + page = document.new_page(width=300, height=height) + page.insert_text((20, 40), "Student answer") + document.save(source) + manual = directory / "manual_resolutions.txt" + manual.write_text("Copie16 A -> B|\n") + original = source.read_bytes() + root = tk.Tk() + try: + panel = ManualResolutionPanel(root, lambda: directory) + panel.pack() + root.update() + self.assertEqual(len(panel.cut_buttons), 1) + panel.cut_buttons[0].invoke() + helper = next(child for child in panel.winfo_children() if isinstance(child, CutHelper)) + helper.canvas.yview_moveto(0) + root.update() + helper.move_bar(SimpleNamespace(y=200 * helper.scale + PAGE_GAP / 2)) + self.assertIn("entre les pages 1 et 2", helper.caption.get()) + helper.keep.set(2) + helper.mode.set("x") + helper.focus_force() + root.update() + helper.event_generate("") + root.update() + self.assertFalse(helper.winfo_exists()) + self.assertEqual(panel.cut_result.get(), "Copie16 A c{33.333333}2x B|") + panel.copy_cut_command() + self.assertEqual(root.clipboard_get(), panel.cut_result.get()) + self.assertEqual(source.read_bytes(), original) + self.assertEqual(manual.read_text(), "Copie16 A -> B|\n") + panel.reload() + self.assertEqual(len(panel.cut_buttons), 1) + finally: + root.destroy() diff --git a/tests/test_gui_manual_resolution.py b/tests/test_gui_manual_resolution.py new file mode 100644 index 0000000..88444fe --- /dev/null +++ b/tests/test_gui_manual_resolution.py @@ -0,0 +1,74 @@ +import os +import tempfile +import tkinter as tk +import unittest +from pathlib import Path +from unittest.mock import call, patch + +from copienator_gui.manual_resolution import ManualResolutionPanel + + +@unittest.skipUnless(os.environ.get("DISPLAY"), "requires a display") +class ManualResolutionPanelTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.evaluation = Path(self.temp.name) + self.root = tk.Tk() + self.addCleanup(self.root.destroy) + self.path = self.evaluation / "manual_resolutions.txt" + + def panel(self): + panel = ManualResolutionPanel(self.root, lambda: self.evaluation) + panel.pack() + self.root.update() + return panel + + def test_preview_editor_and_each_pdf_pair_use_shared_resolution_rules(self): + content = "### Instructions\nCopie01 Ex 1 x> |Ex 2\n\nCopie02 Ex 3 ss Ex 4|\n" + self.path.write_text(content, encoding="utf-8") + paths = [] + for copy, label in (("01", "Ex 1_new"), ("01", "Ex 2_old"), + ("02", "Ex 3"), ("02", "Ex 4")): + path = self.evaluation / "Copies" / f"Copie{copy}" / f"{label}.pdf" + path.parent.mkdir(parents=True, exist_ok=True) + path.touch() + paths.append(path) + panel = self.panel() + self.assertEqual(panel.text.get("1.0", "end-1c"), content) + self.assertEqual(len(panel.pdf_buttons), 4) + self.assertEqual([button.cget("text") for button in panel.pdf_buttons], + ["PDF source", "PDF cible"] * 2) + with patch("copienator_gui.manual_resolution.open_path") as opened: + panel.editor_button.invoke() + for button in panel.pdf_buttons: + button.invoke() + self.assertEqual(opened.call_args_list, [call(self.path), *map(call, paths)]) + + def test_reload_keeps_invalid_lines_visible_and_removes_stale_actions(self): + self.path.write_text("Copie01 A -> B|\n", encoding="utf-8") + panel = self.panel() + self.path.write_text("bad instruction\nCopie02 C sx D\n", encoding="utf-8") + panel.reload() + self.assertIn("bad instruction", panel.text.get("1.0", "end")) + self.assertIn("lignes invalides : 1", panel.status.cget("text")) + self.assertEqual(len(panel.pdf_buttons), 2) + self.path.unlink() + panel.reload() + self.assertFalse(panel.pdf_buttons) + self.assertEqual(str(panel.editor_button.cget("state")), "disabled") + + def test_missing_source_does_not_prevent_opening_destination(self): + self.path.write_text("Copie01 A -> B|\n", encoding="utf-8") + destination = self.evaluation / "Copies" / "Copie01" / "B.pdf" + destination.parent.mkdir(parents=True) + destination.touch() + panel = self.panel() + with patch("copienator_gui.manual_resolution.open_path") as opened, patch( + "copienator_gui.manual_resolution.messagebox.showerror" + ) as error: + panel.pdf_buttons[0].invoke() + opened.assert_not_called() + panel.pdf_buttons[1].invoke() + opened.assert_called_once_with(destination) + self.assertIn("A.pdf", error.call_args.args[1]) diff --git a/tests/test_manual_cut.py b/tests/test_manual_cut.py new file mode 100644 index 0000000..dc31949 --- /dev/null +++ b/tests/test_manual_cut.py @@ -0,0 +1,125 @@ +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import numpy as np +import pymupdf + +from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json +from copienator.commands.resolve_manual import parse_instruction_text, resolve_manual +from copienator.pdf_cut import cut_position, split_pdf +from copienator_gui.cut_helper import PAGE_GAP, cut_operator, percentage_at_y, y_at_percentage + + +def make_pdf(path, heights=(300, 700), rotation=0, cropped=False): + with pymupdf.open() as document: + for index, height in enumerate(heights): + page = document.new_page(width=200, height=height) + page.draw_rect(pymupdf.Rect(0, 0, 200, height / 2), color=None, fill=(1, 0, 0)) + page.draw_rect(pymupdf.Rect(0, height / 2, 200, height), color=None, fill=(0, 0, 1)) + page.insert_text((30, 30), f"Page {index + 1}") + if cropped: + page.set_cropbox(pymupdf.Rect(10, 20, 180, height - 30)) + page.set_rotation(rotation) + document.save(path) + + +class ManualCutTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + + def test_parser_accepts_new_operators_and_rejects_invalid_cuts(self): + for keep in (1, 2): + for action in (">", "x"): + item = parse_instruction_text(f"Copie16 Ex 4 : 1) c{{43.125}}{keep}{action} |Ex 4 : 2)")[0] + self.assertEqual(item.cut, (43.125, keep)) + self.assertEqual(item.should_merge, action == ">") + self.assertTrue(item.pipe_first) + for operator in ("c{0}1>", "c{100}1>", "c{-1}1>", "c{101}1>", "c{43}3>", + "c{43}1s", "c{NaN}1>"): + with self.assertRaises(CliError): + parse_instruction_text(f"Copie16 A {operator} B") + with self.assertRaises(CliError): + parse_instruction_text("Copie16 A c{43}1> A") + + def test_page_boundary_preserves_whole_pages_without_clipping(self): + source, first, second = [self.root / name for name in ("source.pdf", "first.pdf", "second.pdf")] + make_pdf(source, heights=(100, 200), rotation=180) + before = source.read_bytes() + with patch("pymupdf.Page.show_pdf_page", side_effect=AssertionError("must not clip whole pages")): + split_pdf(source, 33.333333, first, second) + for path, height in ((first, 100), (second, 200)): + with pymupdf.open(path) as pdf: + self.assertEqual(len(pdf), 1) + self.assertEqual(pdf[0].rect.height, height) + self.assertEqual(pdf[0].rotation, 180) + self.assertEqual(source.read_bytes(), before) + + def test_in_page_cut_preserves_visible_pixels_for_cropped_rotated_pages(self): + source, first, second = [self.root / name for name in ("source.pdf", "first.pdf", "second.pdf")] + for rotation in (0, 90, 180, 270): + with self.subTest(rotation=rotation): + make_pdf(source, heights=(400,), rotation=rotation, cropped=True) + with pymupdf.open(source) as pdf: + pix = pdf[0].get_pixmap() + expected = np.frombuffer(pix.samples, np.uint8).reshape(pix.height, pix.width, 3) + split_pdf(source, 50, first, second) + for path, pixels in ((first, expected[:len(expected)//2]), (second, expected[len(expected)//2:])): + with pymupdf.open(path) as pdf: + pix = pdf[0].get_pixmap() + actual = np.frombuffer(pix.samples, np.uint8).reshape(pix.height, pix.width, 3) + self.assertEqual(actual.shape, pixels.shape) + self.assertLess(np.abs(actual.astype(float) - pixels).mean(), 0.1) + + def test_cut_within_page_preserves_subsequent_pages(self): + source, first, second = [self.root / name for name in ("source.pdf", "first.pdf", "second.pdf")] + make_pdf(source) + split_pdf(source, 15, first, second) + with pymupdf.open(first) as pdf: + self.assertEqual([page.rect.height for page in pdf], [150]) + with pymupdf.open(second) as pdf: + self.assertEqual([page.rect.height for page in pdf], [150, 700]) + + def test_resolver_archives_source_and_recorrrects_both_labels(self): + for keep in (1, 2): + for mode in (">", "x"): + for pipe_first in (False, True): + with self.subTest(keep=keep, mode=mode, pipe_first=pipe_first): + root = self.root / f"{keep}{mode == '>'}{pipe_first}" + copies = root / "Copies" / "Copie16" + copies.mkdir(parents=True) + source, target = copies / "A.pdf", copies / "B.pdf" + make_pdf(source) + make_pdf(target, heights=(80,)) + original, destination = source.read_bytes(), target.read_bytes() + atomic_write_json(root / "correction.json", { + label: [[{"id": "16", "result": {}}]] for label in ("A", "B") + }) + new_label = "|B" if pipe_first else "B|" + (root / "manual_resolutions.txt").write_text(f"Copie16 A c{{30}}{keep}{mode} {new_label}\n") + self.assertEqual(resolve_manual(EvaluationWorkspace(root)), 0) + self.assertEqual((copies / "A_old.pdf").read_bytes(), original) + self.assertEqual((copies / "B_old.pdf").read_bytes(), destination) + retained, moved = (300, 700) if keep == 1 else (700, 300) + with pymupdf.open(copies / "A_new.pdf") as pdf: + self.assertEqual([page.rect.height for page in pdf], [retained]) + with pymupdf.open(copies / "B_new.pdf") as pdf: + expected = ([moved, 80] if pipe_first else [80, moved]) if mode == ">" else [moved] + self.assertEqual([page.rect.height for page in pdf], expected) + self.assertEqual(read_json(root / "refaire.json"), [["Copie16", ["A", "B"]]]) + self.assertTrue(all(read_json(root / "correction.json")[label][0][0]["result"]["suffix"] == "_new" for label in ("A", "B"))) + self.assertFalse(list(copies.glob("temp_*.pdf"))) + + def test_helper_snaps_to_gap_and_round_trips_percentage(self): + heights = [100, 200] + for y in (95, 100, 100 + PAGE_GAP / 2, 100 + PAGE_GAP + 5): + percent = percentage_at_y(y, heights, 1) + self.assertEqual(cut_position(heights, percent), (1, 0)) + self.assertEqual(y_at_percentage(percent, heights, 1), 100 + PAGE_GAP / 2) + self.assertEqual(cut_operator(43, 1, ">"), "c{43}1>") + self.assertEqual(cut_operator(100/3, 2, "x"), "c{33.333333}2x") + self.assertEqual(cut_position(heights, 33.333333), (1, 0)) + self.assertNotEqual(cut_position(heights, 33.3)[1], 0) diff --git a/tests/test_manual_resolution_state.py b/tests/test_manual_resolution_state.py new file mode 100644 index 0000000..311b4dd --- /dev/null +++ b/tests/test_manual_resolution_state.py @@ -0,0 +1,65 @@ +import copy +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +import pymupdf + +from copienator import EvaluationWorkspace, atomic_write_json, read_json +from copienator.commands import correction, resolve_manual + + +class ManualResolutionStateTests(unittest.TestCase): + def test_acknowledgement_clears_only_this_target_and_copy(self): + for error in ("wrg-lbl:B?", "wrg-lbl:B?delayed", "wrg-lbl:B?exists", "al:(->)B?(->)C?", "al:(delayed)B"): + with self.subTest(error=error): + source = {"id": "01", "result": {"error": error, "delayed": [ + ["wrong-label", "B"], ["add-label", "B"], ["add-label", "C"]]}} + other = {"id": "02", "result": {"error": error, "delayed": [["wrong-label", "B"]]}} + before_other = copy.deepcopy(other) + results = {"A": [[source, other]]} + resolve_manual.set_suffix_and_clean_error(results, "01", "A", None, "B") + self.assertEqual(source["result"]["delayed"], [["add-label", "C"]]) + self.assertNotIn("B?", source["result"]["error"]) + self.assertEqual(other, before_other) + resolve_manual.set_suffix_and_clean_error(results, "01", "A", None, "C") + self.assertNotIn("delayed", source["result"]) + + def fixture(self, root, operator): + copies = root / "Copies" / "Copie01" + copies.mkdir(parents=True) + for label in ("A", "B"): + with pymupdf.open() as doc: + page = doc.new_page(width=200, height=200) + page.insert_text((20, 40), label) + doc.save(copies / f"{label}.pdf") + atomic_write_json(root / "correction.json", { + "A": [[{"id": "01", "result": {"error": "wrg-lbl:B?", "delayed": [["wrong-label", "B"]]}}]], + "B": [[{"id": "01", "result": {"error": ""}}]], + }) + (root / "manual_resolutions.txt").write_text(f"Copie01 A {operator} B|\n") + return EvaluationWorkspace(root) + + def test_every_resolution_acknowledges_pending_conflict(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) + result = read_json(workspace.correction_file)["A"][0][0]["result"] + self.assertNotIn("delayed", result) + self.assertFalse(workspace.manual_resolutions_file.exists()) + + def test_refaire_does_not_recreate_a_resolved_source_conflict(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = self.fixture(Path(temporary), "x>") + resolve_manual.resolve_manual(workspace) + (workspace.groups_dir / "B").mkdir(parents=True) + args = correction.build_parser().parse_args([str(workspace.root), "--refaire"]) + correction.configure_runtime(workspace, [], args, api_client=Mock()) + with patch.object(correction.grouping, "get_pdf_height", return_value=400), patch.object( + correction.grouping, "create_jpg" + ), patch.object(correction, "process_single_task", return_value=[]): + self.assertEqual(correction.run_configured(args), 0) + self.assertFalse(workspace.manual_resolutions_file.exists()) + self.assertNotIn("delayed", correction.results["A"][0][0]["result"]) diff --git a/tests/test_prompting.py b/tests/test_prompting.py new file mode 100644 index 0000000..9fac940 --- /dev/null +++ b/tests/test_prompting.py @@ -0,0 +1,33 @@ +import unittest +from unittest.mock import patch + +from copienator import prompting + + +class PromptingTests(unittest.TestCase): + def test_perspective_is_inserted_with_alternative_method_guidance(self) -> None: + with patch.object( + prompting, "get_label_text_content", return_value="Question" + ), patch.object( + prompting, "get_label_sol_content", return_value="Solution" + ), patch.object( + prompting, "get_label_persp_content", return_value="Barème détaillé" + ): + prompt = prompting.make_prompt("evaluation", "Ex 1") + + self.assertIn(prompting.PERSPECTIVE_GUIDANCE, prompt) + self.assertIn("Barème détaillé", prompt) + + def test_alternative_method_guidance_is_omitted_without_perspective(self) -> None: + with patch.object( + prompting, "get_label_text_content", return_value="Question" + ), patch.object( + prompting, "get_label_sol_content", return_value="Solution" + ), patch.object(prompting, "get_label_persp_content", return_value=None): + prompt = prompting.make_prompt("evaluation", "Ex 1") + + self.assertNotIn(prompting.PERSPECTIVE_GUIDANCE, prompt) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_splitting_geometry.py b/tests/test_splitting_geometry.py new file mode 100644 index 0000000..9560952 --- /dev/null +++ b/tests/test_splitting_geometry.py @@ -0,0 +1,92 @@ +"""Compare extracted answers against the same Poppler view used for labels.""" + +import tempfile +import unittest +from pathlib import Path + +import numpy as np +import pymupdf +from pdf2image import convert_from_path + +from copienator.commands.splitting_int import _render_split_outputs + + +class SplittingGeometryTests(unittest.TestCase): + def make_source(self, path, rotation, cropped=True): + with pymupdf.open() as document: + page = document.new_page(width=800, height=1000) + # Asymmetric colours exercise both translation and orientation. + for row in range(20): + for column in range(8): + colour = (row / 20, column / 8, (row + column) % 7 / 7) + page.draw_rect( + pymupdf.Rect(column * 100, row * 50, + (column + 1) * 100, (row + 1) * 50), + color=None, fill=colour, + ) + page.insert_text((200, 500), "Middle of the answer") + if cropped: + page.set_cropbox(pymupdf.Rect(20, 80, 760, 920)) + page.set_rotation(rotation) + document.save(path) + + def assert_matches_preview(self, answer, preview, bounds): + actual = np.asarray(convert_from_path(answer, dpi=72)[0]).astype(float) + expected = np.asarray(preview.crop(bounds)).astype(float) + self.assertEqual(actual.shape, expected.shape) + self.assertLess(np.abs(actual - expected).mean(), 0.5) + + def test_cropped_and_uncropped_pages_at_every_rotation(self): + visible = { + 0: (20, 80, 760, 920), + 90: (80, 20, 920, 760), + 180: (40, 80, 780, 920), + 270: (80, 40, 920, 780), + } + for rotation in (0, 90, 180, 270): + for cropped in (False, True): + for full_answer in (False, True): + with self.subTest(rotation=rotation, cropped=cropped, full=full_answer): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "copy.pdf" + self.make_source(source, rotation, cropped) + before = source.read_bytes() + preview = convert_from_path(source, dpi=72, use_cropbox=False)[0] + width, height = preview.size + bounds = visible[rotation] if cropped else (0, 0, width, height) + if full_answer: + coordinates = [("A", 0, 0, 10, 0, 100)] + else: + coordinates = [("A", 0, 250, 260, 0, 100), + ("_", 0, 711, 721, 0, 100)] + bounds = (bounds[0], max(bounds[1], height // 4), + bounds[2], min(bounds[3], height * 3 // 4)) + _render_split_outputs(source, coordinates, root) + self.assert_matches_preview(root / "A.pdf", preview, bounds) + self.assertEqual(source.read_bytes(), before) + + def test_answer_continues_to_bottom_then_next_page(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "copy.pdf" + first = root / "first.pdf" + second = root / "second.pdf" + self.make_source(first, 180) + self.make_source(second, 0) + with pymupdf.open() as document: + for path in (first, second): + with pymupdf.open(path) as part: + document.insert_pdf(part) + document.save(source) + previews = convert_from_path(source, dpi=72, use_cropbox=False) + _render_split_outputs(source, [("A", 0, 700, 720, 0, 100), + ("_", 1, 211, 230, 500, 600)], root) + rendered = convert_from_path(root / "A.pdf", dpi=72) + self.assertEqual(len(rendered), 2) + for actual, preview, bounds in zip(rendered, previews, + [(40, 700, 780, 920), (20, 80, 760, 250)]): + expected = np.asarray(preview.crop(bounds)).astype(float) + actual = np.asarray(actual).astype(float) + self.assertEqual(actual.shape, expected.shape) + self.assertLess(np.abs(actual - expected).mean(), 0.5)