From aa40e58dd17fcade8739499878ea7edc18d51677 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Thu, 20 Aug 2026 15:01:17 +0200 Subject: [PATCH] Standardisation 6 --- Readme.org | 13 +- page_splitter.py | 281 +++++++++++++++++++++++++---------------- plotting.py | 197 +++++++++++++++++------------ tests/test_gui_core.py | 119 +++++++++++++++++ 4 files changed, 419 insertions(+), 191 deletions(-) diff --git a/Readme.org b/Readme.org index 4178cd5..29458aa 100644 --- a/Readme.org +++ b/Readme.org @@ -171,7 +171,8 @@ scripts migrés vers cette convention sont actuellement : - =copies_tools.py=, =grouping.py= et =verify_groups.py= ; - =post-correction.py= et =resolve_manual.py= ; -- =cutleft.py= et =splitting_int.py= ; +- =page_splitter.py=, =cutleft.py=, =plotting.py= et + =splitting_int.py= ; - =annotating.py=, =annotating_with_checks.py= et =annotating_by_label.py= ; - =reading_annotations.py= et =reading_grouped_annotations.py= ; @@ -252,6 +253,12 @@ Mettre les copies scannées au format pdf dans =Interro=. + de déplacer la délimitation à droite/gauche Fix issues with =python page_splitter.py Interro14/Copies/Copie01.pdf= + + Le PDF transformé est construit dans un dossier temporaire. La + copie produite et la sauvegarde dans =Copies Originales= sont + ensuite installées avec rollback : une erreur conserve les deux + versions précédentes. Une relance ciblée lit directement la + sauvegarde originale sans la déplacer au préalable. 4. =python cutleft.py Interro= Découpe la partie gauche des copies, là où il devrait y avoir les @@ -287,6 +294,10 @@ Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~ Pour modifier une seule copie : =python plotting.py Interro/Copies/Copie01.pdf= + Les coordonnées agrégées sont écrites atomiquement dans le JSON de + la copie. Fermer la fenêtre avant la fin d'une copie ne remplace pas + son JSON par un résultat incomplet. + It also generates les =Copie01.json=, à partir des =Copie01_01.json= En cas de soucis, (par exemple les pages ne sont pas dans le bon ordre) - Réordonner les pages du fichier pdf diff --git a/page_splitter.py b/page_splitter.py index 343bf1d..39d970a 100644 --- a/page_splitter.py +++ b/page_splitter.py @@ -1,9 +1,15 @@ +from __future__ import annotations + +import argparse import glob import os import re import shutil -import sys +import tempfile import tkinter as tk +import uuid +from collections.abc import Sequence +from pathlib import Path from tkinter import messagebox import fitz # PyMuPDF @@ -11,16 +17,78 @@ from PIL import Image, ImageDraw, ImageTk from pypdf import PdfReader, PdfWriter from config import PAGE_SPLITTER_KB +from copienator import ( + CliError, + EvaluationWorkspace, + ExitCode, + execute, + target_parser, + workspace_from_target, +) from platform_utils import launch_pdf_arranger - # --- Constants --- # Conversion factor: 1 cm to points (1 inch = 2.54 cm, 72 points = 1 inch) CM_TO_POINTS = (1 / 2.54) * 72 -def list_pdf_files(directory): - l = list(reversed(sorted(glob.glob(os.path.join(directory, "*.pdf"))))) - return [u for u in l if "enonce" not in u] +def list_pdf_files(directory: str | Path) -> list[Path]: + paths = sorted(Path(directory).glob("*.pdf"), key=lambda path: path.name.casefold()) + return [path for path in paths if "enonce" not in path.name.casefold()] + + +def _temporary_sibling(path: Path, purpose: str) -> Path: + return path.with_name(f".{path.name}.{purpose}.{uuid.uuid4().hex}.tmp") + + +def commit_processed_pdf( + workspace: EvaluationWorkspace, + original_path: Path, + generated_path: Path, +) -> Path: + """Commit a processed copy and its original backup with rollback.""" + backup_path = workspace.original_copies_dir / original_path.name + output_path = workspace.copies_dir / original_path.name + workspace.original_copies_dir.mkdir(parents=True, exist_ok=True) + workspace.copies_dir.mkdir(parents=True, exist_ok=True) + + staged_backup = None + if original_path.resolve() != backup_path.resolve(): + staged_backup = _temporary_sibling(backup_path, "new-original") + shutil.copy2(original_path, staged_backup) + saved_backup = _temporary_sibling(backup_path, "old-original") + saved_output = _temporary_sibling(output_path, "old-output") + backup_replaced = False + output_replaced = False + try: + if staged_backup is not None: + if backup_path.exists(): + backup_path.replace(saved_backup) + staged_backup.replace(backup_path) + backup_replaced = True + if output_path.exists(): + output_path.replace(saved_output) + generated_path.replace(output_path) + output_replaced = True + if original_path.resolve() not in { + backup_path.resolve(), + output_path.resolve(), + }: + original_path.unlink() + except Exception: + if output_replaced and output_path.exists(): + output_path.unlink() + if saved_output.exists(): + saved_output.replace(output_path) + if backup_replaced and backup_path.exists(): + backup_path.unlink() + if saved_backup.exists(): + saved_backup.replace(backup_path) + raise + finally: + for temporary in (staged_backup, saved_backup, saved_output): + if temporary is not None and temporary.exists(): + temporary.unlink() + return output_path class PDFPreviewer: @@ -30,26 +98,36 @@ class PDFPreviewer: return False self.pdf_path = self.inputs.pop() self.file_rotation = 0 - self.base_name = os.path.splitext(os.path.basename(self.pdf_path))[0] - self.split_dir = f"{self.base_name}_split" - self.reorder_dir = f"{self.base_name}_reorder" - - # Create a temporary output file - self.final_file = f"{self.base_name}_temp.pdf" + self.base_name = self.pdf_path.stem + self._temporary_directory = tempfile.TemporaryDirectory( + prefix=f".{self.base_name}.page-splitter.", + dir=self.workspace.root, + ) + working_dir = Path(self._temporary_directory.name) + self.split_dir = working_dir / "split" + self.reorder_dir = working_dir / "reorder" + self.final_file = working_dir / f"{self.base_name}.pdf" self.current_page_index = 0 self.page_settings = [] self.processing = False # Flag to prevent multiple finish calls try: self.doc = fitz.open(self.pdf_path) - except Exception as e: + except (OSError, RuntimeError, ValueError) as e: + self.failed = True + self._temporary_directory.cleanup() messagebox.showerror("Error", f"Failed to open PDF file: {e}") self.master.destroy() return - self.master.title(f"PDF Splitter - {os.path.basename(self.pdf_path)}") + self.master.title(f"PDF Splitter - {self.pdf_path.name}") return True - def __init__(self, master, path): + def __init__( + self, + master: tk.Tk, + workspace: EvaluationWorkspace, + inputs: list[Path], + ) -> None: """ Initializes the application. @@ -57,40 +135,16 @@ class PDFPreviewer: master (tk.Tk): The root Tkinter window. pdf_path (str): The path to the input PDF file. """ - if not os.path.exists(path): - messagebox.showerror("Error", f"File not found: {path}") - master.destroy() - return - - if os.path.isdir(path): - self.inputs = list_pdf_files(path) - else: - # Check for existing original in backup and restore if found - dir_name = os.path.dirname(os.path.abspath(path)) - file_name = os.path.basename(path) - if os.path.basename(dir_name) == "Copies": - dir_name = os.path.dirname(dir_name) - path = os.path.join(dir_name, file_name) - backup_path = os.path.join(dir_name, "Copies Originales", file_name) - - if os.path.exists(backup_path): - try: - shutil.move(backup_path, path) - print(f"Restored original file from: {backup_path}") - except Exception as e: - messagebox.showerror("Error", f"Failed to restore original file: {e}") - master.destroy() - return - - self.inputs = [path] - + self.workspace = workspace + self.inputs = inputs self.output_dir = None self.master = master self.num = 0 self.global_rotation = 0 # Rotation appliquée à tous les fichiers self.history = [] + self.failed = False if not self.setup_next_file(): - print(f"Aucun fichier PDF valide trouvé dans : {path}") + print(f"No PDF files found in {workspace.root}") master.destroy() return @@ -249,7 +303,7 @@ class PDFPreviewer: # Re-open the file from disk to reset changes (like moved pages) try: self.doc = fitz.open(self.pdf_path) - except Exception as e: + except (OSError, RuntimeError, ValueError) as e: messagebox.showerror("Error", f"Failed to reopen PDF file: {e}") self.master.destroy() return @@ -328,7 +382,8 @@ class PDFPreviewer: self._initialize_current_page_settings() self.load_page() else: - self.finish_and_process() + if not self.finish_and_process(): + return self.history.append(self.pdf_path) if self.setup_next_file(): self._initialize_current_page_settings() @@ -336,59 +391,24 @@ class PDFPreviewer: else: self.master.destroy() - def finish_and_process(self): - """Starts the PDF splitting process and moves files.""" - self.split_pdf() - # print("Debug : ", self.page_settings) - # input("Splitting done. Continue ?") - self.reorder_pdfs() - # input("Reorder done. Continue ?") - self.concate_files() - - # Logic to move original to backup and replace with new file + def finish_and_process(self) -> bool: + """Render and transactionally install the processed PDF.""" try: - abs_path = os.path.abspath(self.pdf_path) - dir_name = os.path.dirname(abs_path) - file_name = os.path.basename(abs_path) - - backup_dir = os.path.join(dir_name, "Copies Originales") - copies_dir = os.path.join(dir_name, "Copies") - os.makedirs(backup_dir, exist_ok=True) - os.makedirs(copies_dir, exist_ok=True) - - backup_path = os.path.join(backup_dir, file_name) - copies_path = os.path.join(copies_dir, file_name) - - # Remove backup if it already exists (overwrite) - if os.path.exists(backup_path): - os.remove(backup_path) - - # Move the original file to "Copies Originales" - shutil.move(self.pdf_path, backup_path) - - # Move the temp output file to replace the original - shutil.move(self.final_file, copies_path) - - # print(f"Original moved to {backup_path}, new file saved at {self.pdf_path}") - - except Exception as e: - messagebox.showerror("Error", f"Failed to move/replace files: {e}") - - self.remove_dirs() - - def _restore_original(self, path): - """Restores the original file from the 'Copies Originales' backup.""" - dir_name = os.path.dirname(os.path.abspath(path)) - file_name = os.path.basename(path) - backup_path = os.path.join(dir_name, "Copies Originales", file_name) - - if os.path.exists(backup_path): - try: - # Moving overwrites the generated PDF with the original backup - shutil.move(backup_path, path) - print(f"Restored original file from: {backup_path}") - except Exception as e: - print(f"Failed to restore original file: {e}") + self.split_pdf() + self.reorder_pdfs() + self.concate_files() + commit_processed_pdf(self.workspace, self.pdf_path, self.final_file) + except Exception as exc: # noqa: BLE001 - interactive boundary + self.failed = True + self.processing = False + print(f"Failed to process {self.pdf_path}: {exc}") + messagebox.showerror("Error", f"Failed to process PDF: {exc}") + self._temporary_directory.cleanup() + self.master.destroy() + return False + else: + self._temporary_directory.cleanup() + return True def go_to_previous_file(self, event=None): """Goes back to the beginning of the previously completed file.""" @@ -398,14 +418,16 @@ class PDFPreviewer: # Close the currently open document to avoid lock issues if hasattr(self, 'doc'): self.doc.close() + if hasattr(self, "_temporary_directory"): + self._temporary_directory.cleanup() # 1. Push current file back onto the stack so it processes next self.inputs.append(self.pdf_path) - # 2. Get the previous file, restore its original state, and push to stack + # 2. Reprocess the previous file from its preserved original backup prev_file = self.history.pop() - self._restore_original(prev_file) - self.inputs.append(prev_file) + backup = self.workspace.original_copies_dir / Path(prev_file).name + self.inputs.append(backup if backup.is_file() else Path(prev_file)) # 3. Reload environment (setup_next_file will pop prev_file back off the stack) self.setup_next_file() @@ -426,13 +448,9 @@ class PDFPreviewer: for pdf in pdf_files: try: os.remove(pdf) - except Exception as e: + except OSError as e: print(f"Error deleting {pdf}: {e}") - def remove_dirs(self): - shutil.rmtree(self.split_dir) - shutil.rmtree(self.reorder_dir) - def split_pdf(self): """Splits each page of the PDF according to the saved settings.""" print("Starting PDF processing...") @@ -588,13 +606,56 @@ class PDFPreviewer: print(f"Created merged PDF: {self.final_file}") -if __name__ == "__main__": - if len(sys.argv) != 2: - print("Usage: python script_name.py ") - sys.exit(1) +def _selected_inputs( + workspace: EvaluationWorkspace, + target: Path, +) -> list[Path]: + if target.is_file(): + if target.suffix.casefold() != ".pdf": + raise CliError(f"Target is not a PDF: {target}", ExitCode.INVALID_ARGUMENTS) + backup = workspace.original_copies_dir / target.name + return [backup if backup.is_file() else target] - pdf_file_path = sys.argv[1] + directory = target + if target == workspace.copies_dir: + candidates = list_pdf_files(workspace.copies_dir) + candidates = [ + ( + workspace.original_copies_dir / path.name + if (workspace.original_copies_dir / path.name).is_file() + else path + ) + for path in candidates + ] + else: + candidates = list_pdf_files(directory) + return list(reversed(candidates)) + +def run(workspace: EvaluationWorkspace, target: Path) -> ExitCode: + inputs = _selected_inputs(workspace, target) + if not inputs: + print(f"No PDF files found in {target}") + return ExitCode.SUCCESS root = tk.Tk() - app = PDFPreviewer(root, pdf_file_path) + application = PDFPreviewer(root, workspace, inputs) root.mainloop() + return ExitCode.FAILURE if application.failed else ExitCode.SUCCESS + + +def build_parser() -> argparse.ArgumentParser: + return target_parser("Interactively split and reorder scanned PDF pages") + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + + def handle(args: argparse.Namespace) -> ExitCode: + workspace, target = workspace_from_target(args) + return run(workspace, target) + + return execute(parser, argv, handle) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/plotting.py b/plotting.py index 72b805a..6937632 100644 --- a/plotting.py +++ b/plotting.py @@ -1,24 +1,29 @@ -import json +from __future__ import annotations + +import argparse import queue -import sys import threading import tkinter as tk +from collections.abc import Sequence from pathlib import Path from tkinter import messagebox from PIL import Image, ImageDraw, ImageFont, ImageTk +from copienator import ( + EvaluationWorkspace, + ExitCode, + atomic_write_json, + execute, + read_json, + target_parser, + workspace_from_target, +) from platform_utils import open_path - -print("o to open pdf, O original pdf, e to emacs part, p to go back, i to interro, click for coordinates") +from utils import natural_key, read_all_labels # --- Configuration & Globals --- padding = 60 -valid_labels_set = None - -# Queue payload: (pil_image, json_path, metadata) -# metadata is a dict: {'copie': str, 'part': int, 'schema': dict} -image_queue = queue.Queue(maxsize=5) try: font = ImageFont.truetype("DejaVuSans.ttf", size=30) @@ -55,6 +60,14 @@ def convert_list(l, group_id, json_schema): ll.append(ee) return ll + +def normalized_labels(entries): + return [ + str(value["label"]).removeprefix("|").removesuffix("|") + for value in entries + if str(value["label"]) != "_" + ] + def prepare_image(image_path: str, bounding_boxes, all_labels, nb_pages, last_label_index): im = Image.open(image_path) im.load() @@ -94,7 +107,7 @@ def prepare_image(image_path: str, bounding_boxes, all_labels, nb_pages, last_la # --- Processing Logic (Worker Thread) --- -def worker_thread(base_dir, files_to_process, all_labels): +def _worker_items(base_dir, files_to_process, all_labels, output_queue): """ Iterates through files, prepares VISUALS only, and puts metadata in queue. Does NOT write final JSON files anymore. @@ -111,9 +124,8 @@ def worker_thread(base_dir, files_to_process, all_labels): json_schema_path = base_dir / 'Cutleft' / f"{copie}_schema.json" try: - with open(json_schema_path, 'r') as f: - json_schema = json.load(f) - except: + json_schema = read_json(json_schema_path) + except (OSError, TypeError, ValueError): print("No json_schema : ", json_schema_path) continue @@ -124,11 +136,10 @@ def worker_thread(base_dir, files_to_process, all_labels): bb_list = [] json_name = "" try: - with open(json_path, 'r') as f: - json_result = json.load(f) + json_result = read_json(json_path) bb_list = json_result.get("list", []) json_name = json_result.get("name", "") - except Exception as e: + except Exception as e: # noqa: BLE001 - malformed user-editable JSON print(f"Warning: {json_path.name} is malformed! Loading blank. {e}") # We do NOT skip; we continue so the user can fix it in the GUI @@ -138,7 +149,7 @@ def worker_thread(base_dir, files_to_process, all_labels): prepare_image(str(img_path), bb_list, all_labels, nb_pages, last_label_index) error_msg = None - except Exception as e: + except Exception as e: # noqa: BLE001 - keep the item editable in the GUI print(f"Error processing {img_path.name}: {e}") pil_image = Image.open(str(img_path)) error_msg = str(e) @@ -151,15 +162,24 @@ def worker_thread(base_dir, files_to_process, all_labels): "error": error_msg } - image_queue.put((pil_image, json_path, metadata)) + output_queue.put((pil_image, json_path, metadata)) - # Sentinel to indicate finished - image_queue.put((None, None, None)) +def worker_thread(base_dir, files_to_process, all_labels, output_queue): + """Prepare queue items and always terminate the GUI stream.""" + failure = None + try: + _worker_items(base_dir, files_to_process, all_labels, output_queue) + except Exception as exc: # noqa: BLE001 - worker boundary + failure = str(exc) + print(f"Plotting worker failed: {exc}") + finally: + metadata = {"worker_error": failure} if failure else None + output_queue.put((None, None, metadata)) # --- GUI Logic (Main Thread) --- class ImageViewer: - def __init__(self, root, base_dir): + def __init__(self, root, workspace, valid_labels, input_queue): self.root = root self.root.resizable(False, False) # If you resize, coordinates will be wrong @@ -171,7 +191,10 @@ class ImageViewer: root.geometry(f"+{x}+{y}") - self.base_dir = base_dir + self.workspace = workspace + self.base_dir = workspace.root + self.valid_labels = valid_labels + self.image_queue = input_queue self.root.title("Bounding Box Viewer") self.label = tk.Label(root, text="Waiting for images...") self.label.pack(expand=True, fill="both") @@ -192,6 +215,7 @@ class ImageViewer: self.history = [] self.forward_stack = [] self.current_pil_image = None + self.failed = False from config import PLOTTING_KB @@ -202,7 +226,8 @@ class ImageViewer: self.root.bind(PLOTTING_KB["open pdf"], self.on_open_pdf) self.root.bind(PLOTTING_KB["open original pdf"], self.on_open_ori_pdf) self.root.bind(PLOTTING_KB["open eval"], self.on_open_interro) - self.root.bind('', lambda e: self.root.quit()) + self.root.bind('', lambda _event: self.close()) + self.root.protocol("WM_DELETE_WINDOW", self.close) self.label.bind('', self.on_click) self.poll_queue() @@ -214,10 +239,15 @@ class ImageViewer: if self.forward_stack: pil_image, json_path, metadata = self.forward_stack.pop() else: - pil_image, json_path, metadata = image_queue.get_nowait() + pil_image, json_path, metadata = self.image_queue.get_nowait() # Handle End of Stream if pil_image is None: + if metadata and metadata.get("worker_error"): + self.failed = True + messagebox.showerror( + "Processing Error", metadata["worker_error"] + ) self.save_current_batch() # Save any remaining data print("All images processed.") self.root.quit() @@ -241,10 +271,12 @@ class ImageViewer: if self.active_copie_name and self.accumulated_results: main_json_path = self.base_dir / "Copies" / f"{self.active_copie_name}.json" print(f"Writing aggregated result to {main_json_path}") - with open(main_json_path, 'w') as f: - json.dump(self.accumulated_results, f) + atomic_write_json(main_json_path, self.accumulated_results) self.accumulated_results = None + def close(self): + self.root.quit() + def on_previous(self, event): if self.is_viewing and self.history: @@ -288,8 +320,7 @@ class ImageViewer: num_added = 0 # ADD THIS LINE try: - with open(self.current_json_path, 'r') as f: - current_data = json.load(f) + current_data = read_json(self.current_json_path) # Perform the conversion now, post-edit converted_items = convert_list( @@ -298,11 +329,10 @@ class ImageViewer: self.current_meta["schema"] ) - labels = [v["label"] for v in current_data["list"]] - labels = [label for label in labels if label != "_"] - labels = [label[1:] for label in labels if label[0] == "|"] - labels = [label[:-1] for label in labels if label[-1] == "|"] - false_labels = [label for label in labels if label not in valid_labels_set] + labels = normalized_labels(current_data["list"]) + false_labels = [ + label for label in labels if label not in self.valid_labels + ] if false_labels: msg = f"Wrong label in {self.current_json_path.name}: {false_labels}\n\n\tPlease press 'e' to fix it, then press Enter again." @@ -318,7 +348,7 @@ class ImageViewer: if "name" in current_data and current_data["name"] != "Continued": self.accumulated_results["name"] = current_data["name"] - except Exception as e: + except Exception as e: # noqa: BLE001 - interactive validation boundary # Warn user and STOP (do not advance to next image) msg = f"Error reading {self.current_json_path.name}:\n\n{e}\n\nPlease press 'e' to fix it, then press Enter again." print(msg) @@ -387,51 +417,58 @@ class ImageViewer: self.root.clipboard_clear() self.root.clipboard_append(box_str) -from utils import natural_key, read_all_labels +def _selected_images(workspace: EvaluationWorkspace, target: Path) -> list[Path]: + workspace.require_directories("Cutleft", "Copies") + if target.is_file(): + stem = target.stem + exact = workspace.cutleft_dir / f"{stem}.jpg" + if exact.is_file(): + return [exact] + return sorted( + workspace.cutleft_dir.glob(f"{stem}_*.jpg"), + key=natural_key, + ) + return sorted(workspace.cutleft_dir.glob("*.jpg"), key=natural_key) + + +def run(workspace: EvaluationWorkspace, target: Path) -> ExitCode: + workspace.require_files("labels") + all_labels = read_all_labels(workspace.root) + files_to_process = _selected_images(workspace, target) + if not files_to_process: + print(f"No Cutleft images found for {target}") + return ExitCode.PARTIAL + + print( + "o to open pdf, O original pdf, e to edit part, p to go back, " + "i to open the statement, click for coordinates" + ) + input_queue = queue.Queue(maxsize=5) + worker = threading.Thread( + target=worker_thread, + args=(workspace.root, files_to_process, all_labels, input_queue), + daemon=True, + ) + worker.start() + root = tk.Tk() + application = ImageViewer(root, workspace, set(all_labels), input_queue) + root.mainloop() + return ExitCode.PARTIAL if application.failed else ExitCode.SUCCESS + + +def build_parser() -> argparse.ArgumentParser: + return target_parser("Interactively verify detected label coordinates") + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + + def handle(args: argparse.Namespace) -> ExitCode: + workspace, target = workspace_from_target(args) + return run(workspace, target) + + return execute(parser, argv, handle) + if __name__ == "__main__": - if len(sys.argv) < 2: - print("Usage: python plotting.py ") - sys.exit(1) - - input_path = Path(sys.argv[1]) - files_to_process = [] - - - if input_path.is_file(): - # Correctly identify base_dir if we are in 'Copies' or 'Cutleft' - if input_path.parent.name in ["Copies", "Cutleft"]: - base_dir = input_path.parent.parent - else: - base_dir = input_path.parent - - stem = input_path.stem - cutleft_dir = base_dir / "Cutleft" - img_path = cutleft_dir / f"{stem}.jpg" - - if img_path.exists(): - files_to_process = [img_path] - else: - # We're given something like Copie01.pdf, look for its split image parts - files_to_process = sorted(list(cutleft_dir.glob(f"{stem}_*.jpg")), key=natural_key) - else: - base_dir = input_path - cutleft_dir = base_dir / "Cutleft" - if not cutleft_dir.exists(): - print(f"Error: {cutleft_dir} does not exist.") - sys.exit(1) - files_to_process = sorted(cutleft_dir.glob("*.jpg")) - - try: - all_labels = read_all_labels(base_dir) - except FileNotFoundError: - all_labels = [] - valid_labels_set = set(all_labels) - - t = threading.Thread(target=worker_thread, args=(base_dir, files_to_process, all_labels)) - t.daemon = True - t.start() - - root = tk.Tk() - app = ImageViewer(root, base_dir) - root.mainloop() + raise SystemExit(main()) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index faf7252..83d457b 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -306,6 +306,10 @@ class StandardCliTests(unittest.TestCase): "splitting_int": load_script_module( "splitting_int.py", "splitting_int" ), + "page_splitter": load_script_module( + "page_splitter.py", "page_splitter" + ), + "plotting": load_script_module("plotting.py", "plotting"), "copies_tools": load_script_module( "copies_tools.py", "copienator_copies_tools_test" ), @@ -345,6 +349,8 @@ class StandardCliTests(unittest.TestCase): "reading_grouped_annotations": [missing], "cutleft": [missing], "splitting_int": [missing], + "page_splitter": [missing], + "plotting": [missing], } for name, arguments in invocations.items(): with self.subTest(script=name), redirect_stderr(io.StringIO()): @@ -447,6 +453,16 @@ class StandardCliTests(unittest.TestCase): "default", {"target": evaluation}, ), + "page_splitter": ( + "page_splitter", + "default", + {"target": evaluation}, + ), + "plotting": ( + "plotting", + "default", + {"target": evaluation}, + ), } for module_name, (step_id, variant_id, values) in cases.items(): step = steps[step_id] @@ -654,6 +670,109 @@ class StandardCliTests(unittest.TestCase): self.assertTrue(answer.is_file()) self.assertEqual(len(PdfReader(answer).pages), 1) + def test_page_splitter_commits_original_and_generated_copy(self) -> None: + module = self.modules["page_splitter"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + evaluation.mkdir() + original = evaluation / "Copie01.pdf" + generated = evaluation / ".generated.pdf" + original.write_bytes(b"original") + generated.write_bytes(b"processed") + workspace = EvaluationWorkspace(evaluation) + + output = module.commit_processed_pdf(workspace, original, generated) + self.assertEqual(output, evaluation / "Copies" / "Copie01.pdf") + self.assertEqual(output.read_bytes(), b"processed") + self.assertEqual( + (evaluation / "Copies Originales" / "Copie01.pdf").read_bytes(), + b"original", + ) + self.assertFalse(original.exists()) + + def test_page_splitter_commit_failure_rolls_back_both_files(self) -> None: + module = self.modules["page_splitter"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + original = evaluation / "Copie01.pdf" + backup = evaluation / "Copies Originales" / "Copie01.pdf" + output = evaluation / "Copies" / "Copie01.pdf" + backup.parent.mkdir(parents=True) + output.parent.mkdir() + original.write_bytes(b"new original") + backup.write_bytes(b"previous original") + output.write_bytes(b"previous output") + missing_generated = evaluation / "missing-generated.pdf" + + with self.assertRaises(FileNotFoundError): + module.commit_processed_pdf( + EvaluationWorkspace(evaluation), original, missing_generated + ) + self.assertEqual(original.read_bytes(), b"new original") + self.assertEqual(backup.read_bytes(), b"previous original") + self.assertEqual(output.read_bytes(), b"previous output") + + def test_page_splitter_reprocesses_from_preserved_original(self) -> None: + module = self.modules["page_splitter"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + generated = evaluation / "Copies" / "Copie01.pdf" + original = evaluation / "Copies Originales" / "Copie01.pdf" + generated.parent.mkdir(parents=True) + original.parent.mkdir() + generated.write_bytes(b"processed") + original.write_bytes(b"original") + workspace = EvaluationWorkspace(evaluation) + self.assertEqual( + module._selected_inputs(workspace, generated), + [original], + ) + + def test_plotting_batch_save_is_atomic(self) -> None: + module = self.modules["plotting"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + viewer = module.ImageViewer.__new__(module.ImageViewer) + viewer.base_dir = evaluation + viewer.active_copie_name = "Copie01" + viewer.accumulated_results = { + "name": "Test", + "list": [{"label": "Ex 1"}], + } + viewer.save_current_batch() + self.assertEqual( + read_json(evaluation / "Copies" / "Copie01.json"), + {"name": "Test", "list": [{"label": "Ex 1"}]}, + ) + self.assertIsNone(viewer.accumulated_results) + + def test_plotting_validates_plain_and_directional_labels(self) -> None: + module = self.modules["plotting"] + self.assertEqual( + module.normalized_labels( + [ + {"label": "Ex 1"}, + {"label": "|Ex 2"}, + {"label": "Ex 3|"}, + {"label": "_"}, + ] + ), + ["Ex 1", "Ex 2", "Ex 3"], + ) + + def test_plotting_worker_failure_still_terminates_its_queue(self) -> None: + module = self.modules["plotting"] + output_queue = queue.Queue() + with patch.object( + module, "_worker_items", side_effect=RuntimeError("worker failed") + ): + module.worker_thread(Path("."), [], [], output_queue) + image, json_path, metadata = output_queue.get_nowait() + self.assertIsNone(image) + self.assertIsNone(json_path) + self.assertEqual(metadata, {"worker_error": "worker failed"}) + def test_post_correction_main_cleans_json_atomically(self) -> None: module = self.modules["post_correction"] with tempfile.TemporaryDirectory() as directory: