From 644e287586aa665cba058b9f0eb84a117e0f020a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Thu, 20 Aug 2026 14:53:06 +0200 Subject: [PATCH] Standardisation 5 --- Readme.org | 10 + copienator/filesystem.py | 13 +- cutleft.py | 525 +++++++++++++++++++-------------------- splitting_int.py | 425 +++++++++++++++++-------------- tests/test_gui_core.py | 141 +++++++++++ 5 files changed, 648 insertions(+), 466 deletions(-) diff --git a/Readme.org b/Readme.org index d16016b..4178cd5 100644 --- a/Readme.org +++ b/Readme.org @@ -171,6 +171,7 @@ 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= ; - =annotating.py=, =annotating_with_checks.py= et =annotating_by_label.py= ; - =reading_annotations.py= et =reading_grouped_annotations.py= ; @@ -260,6 +261,10 @@ Mettre les copies scannées au format pdf dans =Interro=. Rerun on a single file with =python cutleft.py Interro/Copies/Copie01.pdf= + Les images et le fichier =_schema.json= d'une copie sont remplacés + ensemble. Fermer l'outil juste après la dernière validation ne peut + donc plus interrompre un thread de sauvegarde en arrière-plan. + ** Labelisation et regroupement Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~ @@ -291,6 +296,11 @@ Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~ Découpe les copies suivant les exercices Peut-être appelé avec une seule copie. + + Les réponses d'une copie sont préparées dans un dossier temporaire, + puis remplacent ensemble le dossier précédent. En cas d'erreur, + l'ancienne version est conservée. Les réponses devenues obsolètes + restent archivées dans le sous-dossier =Missing=. 4. =python grouping.py Interro= Regroupe les mêmes questions de différentes copies en groupes de diff --git a/copienator/filesystem.py b/copienator/filesystem.py index 3bdb6d8..e5331af 100644 --- a/copienator/filesystem.py +++ b/copienator/filesystem.py @@ -2,6 +2,7 @@ from __future__ import annotations import shutil import uuid +from collections.abc import Iterable from contextlib import contextmanager from pathlib import Path @@ -44,7 +45,11 @@ def staged_directory(destination: str | Path): @contextmanager -def staged_files(destination: str | Path): +def staged_files( + destination: str | Path, + *, + remove: Iterable[str] = (), +): """Stage a set of files and merge them into a directory with rollback.""" target = Path(destination) target.parent.mkdir(parents=True, exist_ok=True) @@ -56,9 +61,15 @@ def staged_files(destination: str | Path): try: yield staging staged = sorted(path for path in staging.iterdir() if path.is_file()) + staged_names = {path.name for path in staged} + removed_names = set(remove) - staged_names target.mkdir(parents=True, exist_ok=True) backup.mkdir() try: + for name in sorted(removed_names): + destination_path = target / name + if destination_path.is_file() or destination_path.is_symlink(): + destination_path.replace(backup / name) for source in staged: destination_path = target / source.name if destination_path.exists() or destination_path.is_symlink(): diff --git a/cutleft.py b/cutleft.py index beccf19..12a7b6a 100644 --- a/cutleft.py +++ b/cutleft.py @@ -1,366 +1,343 @@ -import sys -from functools import lru_cache -import os -import time -import json # Added for schema output +from __future__ import annotations + import argparse +import threading +import time import tkinter as tk +from collections.abc import Sequence +from functools import lru_cache +from pathlib import Path +from queue import Empty, Queue from threading import Thread -from queue import Queue, Empty + from pdf2image import convert_from_path from PIL import Image, ImageTk -# --- Configuration --- +from copienator import ( + CliError, + EvaluationWorkspace, + ExitCode, + atomic_write_json, + execute, + target_parser, + workspace_from_target, +) +from copienator.filesystem import staged_files + DELIMITER_WIDTH = 5 DELIMITER_COLOR = (0, 0, 0) OUTPUT_SIZE = (1800, 1000) - -parser = argparse.ArgumentParser(description="PDF Cropper") -parser.add_argument("path", help="Directory path or PDF file path") -parser.add_argument("--fullpage", action="store_true", help="Process all files in full page mode (1 page per output file)") -args = parser.parse_args() - -path_arg = args.path -fullpage_mode = args.fullpage -files = [] -INPUT_DIR = "" -COPIES_DIR = "" - -if os.path.isfile(path_arg) and path_arg.lower().endswith('.pdf'): - COPIES_DIR = os.path.abspath(os.path.dirname(path_arg)) - # If the file is inside a "Copies" folder, set INPUT_DIR to the parent - if os.path.basename(COPIES_DIR).lower() == 'copies': - INPUT_DIR = os.path.dirname(COPIES_DIR) - else: - INPUT_DIR = COPIES_DIR - files = [os.path.basename(path_arg)] -elif os.path.isdir(path_arg): - # Support passing either the base dir or the Copies dir directly - abs_path = os.path.abspath(path_arg) - if os.path.basename(abs_path).lower() == 'copies': - COPIES_DIR = abs_path - INPUT_DIR = os.path.dirname(abs_path) - else: - INPUT_DIR = abs_path - COPIES_DIR = os.path.join(INPUT_DIR, 'Copies') - - if os.path.exists(COPIES_DIR): - files = sorted([f for f in os.listdir(COPIES_DIR) if f.lower().endswith('.pdf') and - "nonc" not in f.lower()]) - else: - sys.exit(f"Error: Could not find 'Copies' directory inside {INPUT_DIR}") -else: - sys.exit("Error: Input must be a directory or a PDF file.") - - -OUTPUT_DIR = os.path.join(INPUT_DIR, 'Cutleft') - -if not os.path.exists(OUTPUT_DIR): - os.makedirs(OUTPUT_DIR) - -# --- Processing Logic --- - -def distribute_pages(total_pages, max_per_file=5): - """ - Calculates how to split pages into chunks <= max_per_file, - balancing the number of columns per file. - Example: 12 pages, max 5 -> [4, 4, 4] - """ - if total_pages == 0: - return [] - - # Calculate minimum number of files needed - num_files = (total_pages + max_per_file - 1) // max_per_file - - # Calculate base size and remainder - base_count = total_pages // num_files - remainder = total_pages % num_files - - distribution = [] - for i in range(num_files): - # Distribute remainder to the first few files - count = base_count + (1 if i < remainder else 0) - distribution.append(count) - - return distribution - -def stitch_images(image_list): - """Helper to stitch a list of images horizontally with delimiters.""" - if not image_list: - return None - - num_images = len(image_list) - total_width = sum(img.width for img in image_list) + (num_images - 1) * DELIMITER_WIDTH - max_height = max(img.height for img in image_list) - - combined = Image.new('RGB', (total_width, max_height), color=(255, 255, 255)) - - x_offset = 0 - for idx, img in enumerate(image_list): - combined.paste(img, (x_offset, 0)) - x_offset += img.width - if idx < num_images - 1: - delimiter = Image.new('RGB', (DELIMITER_WIDTH, max_height), color=DELIMITER_COLOR) - combined.paste(delimiter, (x_offset, 0)) - x_offset += DELIMITER_WIDTH - - return combined - -import threading pdf_cache_lock = threading.Lock() +def distribute_pages(total_pages: int, max_per_file: int = 5) -> list[int]: + """Distribute pages into balanced chunks no larger than max_per_file.""" + if total_pages == 0: + return [] + number_of_files = (total_pages + max_per_file - 1) // max_per_file + base_count, remainder = divmod(total_pages, number_of_files) + return [ + base_count + (1 if index < remainder else 0) + for index in range(number_of_files) + ] + + +def stitch_images(image_list: list[Image.Image]) -> Image.Image | None: + if not image_list: + return None + total_width = sum(image.width for image in image_list) + total_width += (len(image_list) - 1) * DELIMITER_WIDTH + max_height = max(image.height for image in image_list) + combined = Image.new("RGB", (total_width, max_height), color="white") + x_offset = 0 + for index, image in enumerate(image_list): + combined.paste(image, (x_offset, 0)) + x_offset += image.width + if index < len(image_list) - 1: + delimiter = Image.new( + "RGB", (DELIMITER_WIDTH, max_height), color=DELIMITER_COLOR + ) + combined.paste(delimiter, (x_offset, 0)) + x_offset += DELIMITER_WIDTH + return combined + + @lru_cache(maxsize=3) -def _get_pdf_pages_cached(filename): - pdf_path = os.path.join(COPIES_DIR, filename) +def _get_pdf_pages_cached(pdf_path: Path) -> list[Image.Image]: return convert_from_path(pdf_path) -def get_pdf_pages(filename): - """Thread-safe wrapper for the cached PDF conversion.""" + +def get_pdf_pages(pdf_path: Path) -> list[Image.Image]: + """Thread-safe wrapper around the small PDF conversion cache.""" with pdf_cache_lock: - return _get_pdf_pages_cached(filename) + return _get_pdf_pages_cached(pdf_path) -def process_single_pdf(filename, shift_offset=0, max_per_file=5): - """ - Converts PDF to stitched images. - Returns a tuple: (preview_image_resized, list_of_split_images, schema_dict) - """ + +def process_single_pdf( + pdf_path: Path, + shift_offset: int = 0, + max_per_file: int = 5, +) -> tuple[Image.Image, list[Image.Image], dict[str, object]] | None: + """Convert one PDF into a preview, full-resolution splits and metadata.""" try: - pages = get_pdf_pages(filename) cropped_images = [] - - for img in pages: - width, height = img.size - + for image in get_pdf_pages(pdf_path): + width, height = image.size if max_per_file == 1: - # If Single Page mode, take the full width (ignore shift/crop) - left = 0 - right = width + left, right = 0, width else: - # Original "Cutleft" logic (approx 1/3 width) - left = 100 + shift_offset - right = (width // 3) + 100 + shift_offset - - # Ensure crop box is valid - left = max(0, left) - right = min(width, right) - + left = max(0, 100 + shift_offset) + right = min(width, width // 3 + 100 + shift_offset) if right > left: - crop_box = (left, 0, right, height) - cropped = img.crop(crop_box) - cropped_images.append(cropped) - + cropped_images.append(image.crop((left, 0, right, height))) if not cropped_images: return None - # 1. Generate Schema / Distribution - col_distribution = distribute_pages(len(cropped_images), max_per_file=max_per_file) - - # 2. Generate Split Images (Full Resolution) + distribution = distribute_pages(len(cropped_images), max_per_file) split_images = [] - current_idx = 0 - for count in col_distribution: - chunk = cropped_images[current_idx : current_idx + count] - stitched_chunk = stitch_images(chunk) - split_images.append(stitched_chunk) - current_idx += count - - # 3. Generate Preview (All stitched together, Resized) + current_index = 0 + for count in distribution: + stitched = stitch_images(cropped_images[current_index : current_index + count]) + if stitched is not None: + split_images.append(stitched) + current_index += count full_stitch = stitch_images(cropped_images) - preview_resized = full_stitch.resize(OUTPUT_SIZE, Image.BILINEAR) - - schema = { - "original_filename": filename, + if full_stitch is None: + return None + preview = full_stitch.resize(OUTPUT_SIZE, Image.Resampling.BILINEAR) + schema: dict[str, object] = { + "original_filename": pdf_path.name, "total_pages": len(cropped_images), "number_of_files": len(split_images), - "columns_per_file": col_distribution + "columns_per_file": distribution, } - - return (preview_resized, split_images, schema) - - except Exception as e: - print(f"Error processing {filename}: {e}") + return preview, split_images, schema + except Exception as exc: # noqa: BLE001 - interactive item failure + print(f"Error processing {pdf_path.name}: {exc}") return None -def save_results(result_tuple, filename): - """ - Saves the split images and the schema JSON. - """ - _, splits, schema = result_tuple - base_name = os.path.splitext(filename)[0] - # --- Cleanup: Delete existing files for this PDF --- - for f in os.listdir(OUTPUT_DIR): - file_path = os.path.join(OUTPUT_DIR, f) - if f == f"{base_name}_schema.json": - os.remove(file_path) - elif f.startswith(f"{base_name}_") and f.endswith(".jpg"): - suffix = f[len(base_name)+1:-4] - if suffix.isdigit(): - os.remove(file_path) - - # Save Images - for i, img in enumerate(splits): - suffix = f"_{i+1:02d}" - output_filename = f"{base_name}{suffix}.jpg" - output_path = os.path.join(OUTPUT_DIR, output_filename) - img.save(output_path, "JPEG", quality=95) - print(f"Saved: {output_filename}") - - # Save Schema - json_filename = f"{base_name}_schema.json" - json_path = os.path.join(OUTPUT_DIR, json_filename) - with open(json_path, 'w') as f: - json.dump(schema, f, indent=4) - print(f"Saved schema: {json_filename}") +def _previous_cutleft_outputs(output_dir: Path, base_name: str) -> set[str]: + if not output_dir.is_dir(): + return set() + result = {f"{base_name}_schema.json"} + for path in output_dir.glob(f"{base_name}_*.jpg"): + suffix = path.stem.removeprefix(f"{base_name}_") + if suffix.isdigit(): + result.add(path.name) + return result -# --- GUI Application --- +def save_results( + result: tuple[Image.Image, list[Image.Image], dict[str, object]], + pdf_path: Path, + output_dir: Path, +) -> None: + """Atomically replace every Cutleft output associated with one copy.""" + _, splits, schema = result + base_name = pdf_path.stem + previous = _previous_cutleft_outputs(output_dir, base_name) + with staged_files(output_dir, remove=previous) as staging: + for index, image in enumerate(splits, start=1): + filename = f"{base_name}_{index:02d}.jpg" + image.save(staging / filename, "JPEG", quality=95) + atomic_write_json(staging / f"{base_name}_schema.json", schema) + for index in range(1, len(splits) + 1): + print(f"Saved: {base_name}_{index:02d}.jpg") + print(f"Saved schema: {base_name}_schema.json") + class ImageReviewer: - def __init__(self, file_list, default_max_per_file=5): - self.files = file_list + def __init__( + self, + files: list[Path], + output_dir: Path, + default_max_per_file: int = 5, + ) -> None: + self.files = files + self.output_dir = output_dir self.index = 0 self.current_shift = 0 self.default_max_per_file = default_max_per_file self.current_max_per_file = default_max_per_file - self.current_preview = None # Only stores the resized preview for GUI + self.current_preview: Image.Image | None = None self.is_processing = False + self.manual_queue: Queue[ + tuple[Image.Image, list[Image.Image], dict[str, object]] | None + ] = Queue() - # Queue for manual re-processing results - self.manual_queue = Queue() - - # Setup GUI self.root = tk.Tk() self.root.title("PDF Cropper") self.root.geometry("+100+100") - self.label_img = tk.Label(self.root) self.label_img.pack() - self.label_info = tk.Label(self.root, text="", font=("Arial", 12, "bold")) self.label_info.pack(pady=5) + self.root.bind("", self.on_next) + self.root.bind("n", lambda _event: self.on_shift(50)) + self.root.bind("N", lambda _event: self.on_shift(100)) + self.root.bind("t", lambda _event: self.on_shift(-50)) + self.root.bind("1", lambda _event: self.on_set_max_pages(1)) - # Bindings - self.root.bind('', self.on_next) - self.root.bind('n', lambda e: self.on_shift(50)) - self.root.bind('N', lambda e: self.on_shift(100)) - self.root.bind('t', lambda e: self.on_shift(-50)) - self.root.bind('1', lambda e: self.on_set_max_pages(1)) - - # Start background pre-fetcher - self.bg_thread = Thread(target=self.prefetch_worker, daemon=True) - self.bg_thread.start() - - # Load first image + Thread(target=self.prefetch_worker, daemon=True).start() self.load_current_image() - self.root.lift() self.root.focus_force() self.root.mainloop() - def on_set_max_pages(self, count): + def on_set_max_pages(self, count: int) -> None: if self.is_processing: return self.current_max_per_file = count print(f"Setting max pages per file: {count}") self.trigger_processing(self.files[self.index], self.current_shift) - def prefetch_worker(self): - """Background thread to load the NEXT file's PDF pages into RAM.""" - idx_to_process = -1 + def prefetch_worker(self) -> None: + processed_index = -1 while True: target = self.index + 1 - if target < len(self.files) and target != idx_to_process: - fname = self.files[target] - get_pdf_pages(fname) - idx_to_process = target + if target < len(self.files) and target != processed_index: + get_pdf_pages(self.files[target]) + processed_index = target time.sleep(0.05) - def load_current_image(self, use_prefetch=False): + def load_current_image(self) -> None: if self.index >= len(self.files): print("All files processed.") self.root.destroy() return - - filename = self.files[self.index] self.is_processing = False self.current_shift = 0 + self.trigger_processing(self.files[self.index], self.current_shift) - self.trigger_processing(filename, self.current_shift) - - def trigger_processing(self, filename, shift): - """Starts a thread to process image so GUI doesn't freeze.""" + def trigger_processing(self, pdf_path: Path, shift: int) -> None: self.is_processing = True - self.label_info.configure(text=f"Processing {filename} (Shift {shift})... Please wait.", fg="red") + self.label_info.configure( + text=f"Processing {pdf_path.name} (Shift {shift})... Please wait.", + fg="red", + ) - def worker(): - res = process_single_pdf(filename, shift, self.current_max_per_file) - self.manual_queue.put(res) - - Thread(target=worker, daemon=True).start() - self.check_manual_queue(filename) - - def check_manual_queue(self, filename): - """Polls the manual queue for result.""" - try: - result = self.manual_queue.get_nowait() - if result: - self.handle_processing_result(result, filename) - else: - print(f"Failed to process {filename}, skipping.") - self.index += 1 - self.load_current_image(use_prefetch=True) - self.is_processing = False - except Empty: - self.root.after(100, lambda: self.check_manual_queue(filename)) - - def handle_processing_result(self, result, filename): - """Unpacks result, saves files, and updates display.""" - preview, splits, schema = result - self.current_preview = preview - - Thread(target=save_results, args=(result, filename), daemon=True).start() - - self.update_display(filename, schema) - - def update_display(self, filename, schema=None): - if self.current_preview: - tk_image = ImageTk.PhotoImage(self.current_preview) - self.label_img.configure(image=tk_image) - self.label_img.image = tk_image - - schema_info = "" - if schema: - cols = str(schema['columns_per_file']) - schema_info = f"\nFiles: {schema['number_of_files']} | Cols: {cols}" - - self.label_info.configure( - text=f"[{self.index+1}/{len(self.files)}] {filename} | Shift: {self.current_shift}px" - f"{schema_info}\n" - f"Enter: Next | n: +50 | N: +100 | t: -50 | 1: use single column", - fg="black" + def worker() -> None: + self.manual_queue.put( + process_single_pdf(pdf_path, shift, self.current_max_per_file) ) - def on_shift(self, amount): + Thread(target=worker, daemon=True).start() + self.check_manual_queue(pdf_path) + + def check_manual_queue(self, pdf_path: Path) -> None: + try: + result = self.manual_queue.get_nowait() + if result is None: + print(f"Failed to process {pdf_path.name}, skipping.") + self.index += 1 + self.load_current_image() + else: + self.handle_processing_result(result, pdf_path) + self.is_processing = False + except Empty: + self.root.after(100, lambda: self.check_manual_queue(pdf_path)) + + def handle_processing_result( + self, + result: tuple[Image.Image, list[Image.Image], dict[str, object]], + pdf_path: Path, + ) -> None: + self.current_preview = result[0] + save_results(result, pdf_path, self.output_dir) + self.update_display(pdf_path.name, result[2]) + + def update_display(self, filename: str, schema: dict[str, object]) -> None: + if self.current_preview is None: + return + tk_image = ImageTk.PhotoImage(self.current_preview) + self.label_img.configure(image=tk_image) + self.label_img.image = tk_image + self.label_info.configure( + text=( + f"[{self.index + 1}/{len(self.files)}] {filename} | " + f"Shift: {self.current_shift}px\nFiles: {schema['number_of_files']} | " + f"Cols: {schema['columns_per_file']}\n" + "Enter: Next | n: +50 | N: +100 | t: -50 | " + "1: use single column" + ), + fg="black", + ) + + def on_shift(self, amount: int) -> None: if self.is_processing: return self.current_shift += amount print(f"Applying shift: {self.current_shift}") self.trigger_processing(self.files[self.index], self.current_shift) - def on_next(self, event): + def on_next(self, _event: object) -> None: if self.is_processing: return self.index += 1 self.current_shift = 0 self.current_max_per_file = self.default_max_per_file - self.load_current_image(use_prefetch=True) + self.load_current_image() -# --- Entry Point --- -if __name__ == "__main__": +def _selected_files( + workspace: EvaluationWorkspace, + target: Path, +) -> list[Path]: + workspace.require_directories("Copies") + if target.is_file(): + if target.suffix.casefold() != ".pdf": + raise CliError(f"Target is not a PDF: {target}", ExitCode.INVALID_ARGUMENTS) + return [target] + return sorted( + ( + path + for path in workspace.copies_dir.glob("*.pdf") + if "nonc" not in path.name.casefold() + ), + key=lambda path: path.name.casefold(), + ) + + +def run( + workspace: EvaluationWorkspace, + target: Path, + *, + fullpage: bool = False, +) -> ExitCode: + files = _selected_files(workspace, target) if not files: print("No PDF files found.") - else: - app = ImageReviewer(files, default_max_per_file=1 if fullpage_mode else 5) + return ExitCode.SUCCESS + workspace.cutleft_dir.mkdir(parents=True, exist_ok=True) + _get_pdf_pages_cached.cache_clear() + ImageReviewer( + files, + workspace.cutleft_dir, + default_max_per_file=1 if fullpage else 5, + ) + return ExitCode.SUCCESS + + +def build_parser() -> argparse.ArgumentParser: + parser = target_parser("Interactively crop the label margin from PDF copies") + parser.add_argument( + "--fullpage", + action="store_true", + help="Use each complete page instead of cropping the label margin", + ) + return parser + + +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, fullpage=args.fullpage) + + return execute(parser, argv, handle) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/splitting_int.py b/splitting_int.py index 99e0ae4..18bdf99 100644 --- a/splitting_int.py +++ b/splitting_int.py @@ -1,223 +1,266 @@ -import fitz # PyMuPDF -from pypdf import PdfWriter -from pypdf import PdfReader -import os -import sys -import json +from __future__ import annotations + +import argparse import shutil -from pathlib import Path +import tempfile from collections import defaultdict +from collections.abc import Sequence +from pathlib import Path -from utils import read_all_labels +import fitz +from pypdf import PdfReader, PdfWriter -carreau = 1000 // 38 +import utils +from copienator import ( + CliError, + EvaluationWorkspace, + ExitCode, + execute, + read_json, + target_parser, + workspace_from_target, +) +from copienator.filesystem import staged_directory + +SQUARE = 1000 // 38 +Coordinate = tuple[str, int, int, int, int, int] +ParsedCoordinate = tuple[str, str, int, int, int, int, int] -def decode_json(pdf_file): - file_path = Path(pdf_file) - with open(file_path.with_suffix(".json"), "r") as f: - json_result = json.load(f) - - nb_pages = len(PdfReader(file_path).pages) - - bb_list = json_result["list"] - name = json_result["name"] - column_width = 1000 // nb_pages - - def page_number(b): - return ((b[1] + b[3]) // 2) // column_width - - result = [] - for d in bb_list: - (b, label) = d["box_2d"], d["label"] - pn = page_number(b) - result.append((label, pn, b[0] - carreau, b[2]-carreau, b[1], b[3])) - result.sort(key=lambda x: (x[1], x[2])) - return (name, result) +def decode_json(pdf_file: str | Path) -> tuple[str, list[Coordinate]]: + """Read verified label coordinates associated with one copy PDF.""" + pdf_path = Path(pdf_file) + loaded = read_json(pdf_path.with_suffix(".json")) + if not isinstance(loaded, dict): + raise TypeError(f"Expected a JSON object for {pdf_path}") + boxes = loaded.get("list") + if not isinstance(boxes, list): + raise TypeError(f"Expected a list of labels for {pdf_path}") + page_count = len(PdfReader(pdf_path).pages) + if page_count == 0: + raise ValueError(f"PDF contains no pages: {pdf_path}") + column_width = 1000 // page_count + result: list[Coordinate] = [] + for entry in boxes: + if not isinstance(entry, dict): + raise TypeError(f"Malformed label entry for {pdf_path}: {entry!r}") + box = entry["box_2d"] + label = str(entry["label"]) + page_number = ((box[1] + box[3]) // 2) // column_width + result.append( + (label, page_number, box[0] - SQUARE, box[2] - SQUARE, box[1], box[3]) + ) + result.sort(key=lambda item: (item[1], item[2])) + return str(loaded.get("name", "")), result -def split_an_interro(base_dir, input_pdf, coords_list): - doc = fitz.open(input_pdf) - - output_dir = base_dir / "Copies" / input_pdf.stem - generated_files = set() - parts_by_label = defaultdict(list) - - # 1. Parse labels to strip '|' and determine type: L (Left), R (Right), N (Normal) - parsed_coords = [] - for item in coords_list: - label, pn, y0, y1, x0, x1 = item +def _parse_coordinates(coords_list: list[Coordinate]) -> list[ParsedCoordinate]: + parsed: list[ParsedCoordinate] = [] + for label, page, y0, y1, x0, x1 in coords_list: if label.startswith("|"): - c_type, clean_label = "L", label[1:] + kind, clean_label = "L", label[1:] elif label.endswith("|"): - c_type, clean_label = "R", label[:-1] + kind, clean_label = "R", label[:-1] else: - c_type, clean_label = "N", label - parsed_coords.append((clean_label, c_type, pn, y0, y1, x0, x1)) + kind, clean_label = "N", label + parsed.append((clean_label, kind, page, y0, y1, x0, x1)) + filtered: list[ParsedCoordinate] = [] + for item in parsed: + if not filtered or item[0] != filtered[-1][0]: + filtered.append(item) + return filtered - # 2. Filter consecutive duplicate labels based on the cleaned name - filtered_coords = [] - if parsed_coords: - filtered_coords.append(parsed_coords[0]) - for item in parsed_coords[1:]: - if item[0] != filtered_coords[-1][0]: - filtered_coords.append(item) - coords_list = filtered_coords - def scale_coord(y, page): - """Scale y from 0–1000 range to PDF points.""" - page_height = page.rect.height - return (y / 1000) * page_height - - def save_cropped_page(doc, page_num, x0, y0, x1, y1, out_path): - """Saves a cropped portion of a page as a new PDF.""" - page = doc[page_num] - rotated_rect = page.rect * page.transformation_matrix - visual_crop_rect = fitz.Rect(rotated_rect.x0 + x0, y0, rotated_rect.x0 + x1, y1) - unrotated_clip_rect = visual_crop_rect * page.derotation_matrix - - temp_doc = fitz.open() - temp_page = temp_doc.new_page( - width=visual_crop_rect.width, - height=visual_crop_rect.height - ) - temp_page.show_pdf_page( - temp_page.rect, - doc, - page_num, +def _save_cropped_page( + document: fitz.Document, + page_number: int, + x0: float, + y0: float, + x1: float, + y1: float, + output_path: Path, +) -> None: + page = document[page_number] + rotated_rectangle = page.rect * page.transformation_matrix + visual_crop = fitz.Rect( + rotated_rectangle.x0 + x0, + y0, + rotated_rectangle.x0 + x1, + y1, + ) + unrotated_clip = visual_crop * page.derotation_matrix + cropped = fitz.open() + try: + target_page = cropped.new_page(width=visual_crop.width, height=visual_crop.height) + target_page.show_pdf_page( + target_page.rect, + document, + page_number, rotate=-page.rotation, - clip=unrotated_clip_rect + clip=unrotated_clip, ) - temp_doc.save(out_path) - temp_doc.close() + cropped.save(output_path) + finally: + cropped.close() - # Iterate through all labels - for idx, (clean_label, c_type, start_page, y_start_raw, y_end_box, x0_raw, x1_raw) in enumerate(coords_list): - if clean_label == "_": + +def _render_split_outputs( + input_pdf: Path, + coords_list: list[Coordinate], + staging: Path, +) -> set[str]: + """Render every current answer into an otherwise empty staging directory.""" + document = fitz.open(input_pdf) + try: + parsed = _parse_coordinates(coords_list) + parts_by_label: defaultdict[str, list[Path]] = defaultdict(list) + with tempfile.TemporaryDirectory(prefix="copienator-split-") as temp_directory: + temporary = Path(temp_directory) + for index, item in enumerate(parsed): + clean_label, kind, start_page, y_start, _y_end, x0_raw, _x1_raw = item + if clean_label == "_": + continue + if not 0 <= start_page < document.page_count: + raise ValueError( + f"Invalid page {start_page} for {input_pdf.name}" + ) + end_page = document.page_count - 1 + end_y = 1000 + for next_item in parsed[index + 1 :]: + _next_label, next_kind, next_page, next_y, *_rest = next_item + if ( + (kind == "L" and next_kind in {"L", "N"}) + or (kind == "R" and next_kind in {"R", "N"}) + or kind == "N" + ): + end_page = next_page + end_y = min(next_y + int(1.5 * SQUARE), 1000) + break + + column_width = 1000 / document.page_count + if kind == "L": + fraction_x0 = (x0_raw % column_width) / column_width + fraction_x1 = 1.0 + end_y = min(1000, end_y + 40) + elif kind == "R": + fraction_x0 = 0.0 + left_labels = [entry for entry in parsed if entry[1] == "L"] + if left_labels: + closest = min(left_labels, key=lambda entry: abs(entry[3] - y_start)) + center = (closest[5] + closest[6]) / 2.0 + fraction_x1 = (center % column_width) / column_width + if fraction_x1 <= fraction_x0: + fraction_x1 = 1.0 + else: + fraction_x1 = 1.0 + else: + fraction_x0, fraction_x1 = 0.0, 1.0 + + for page_number in range(start_page, end_page + 1): + 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: + 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, + part_path, + ) + parts_by_label[clean_label].append(part_path) + + generated: set[str] = set() + for label, parts in parts_by_label.items(): + filename = f"{label}.pdf" + merger = PdfWriter() + try: + for part in parts: + merger.append(part) + merger.write(staging / filename) + finally: + merger.close() + generated.add(filename) + return generated + finally: + document.close() + + +def _preserve_previous_outputs( + output_dir: Path, + staging: Path, + generated_files: set[str], +) -> None: + if not output_dir.is_dir(): + return + for directory in (path for path in output_dir.iterdir() if path.is_dir()): + shutil.copytree(directory, staging / directory.name, dirs_exist_ok=True) + missing_dir = staging / "Missing" + for item in (path for path in output_dir.iterdir() if path.is_file()): + if item.name in generated_files: continue + print(f"ALERT: File '{item.name}' not generated. Moving to {missing_dir}") + missing_dir.mkdir(exist_ok=True) + shutil.copy2(item, missing_dir / item.name) - temp_parts = [] - end_page = doc.page_count - 1 - end_y_target_raw = 1000 - # RULE 2: Determine stopping label - for next_item in coords_list[idx + 1:]: - n_clean, n_type, n_pn, n_y_start, n_y_end, _, _ = next_item +def split_an_interro( + workspace: EvaluationWorkspace, + input_pdf: Path, + coords_list: list[Coordinate], +) -> None: + """Regenerate one copy's answers and preserve obsolete ones under Missing.""" + output_dir = workspace.copies_dir / input_pdf.stem + with staged_directory(output_dir) as staging: + generated = _render_split_outputs(input_pdf, coords_list, staging) + _preserve_previous_outputs(output_dir, staging, generated) - if c_type == "L": - is_stop = (n_type in ("L", "N")) - elif c_type == "R": - is_stop = (n_type in ("R", "N")) - else: - is_stop = True # Normal labels stop at anything - if is_stop: - end_page = n_pn - # end_y_target_raw = n_y_start - # On avait retiré un carreau précédemment inutilement, on le rajoute, plus un demi carreau - end_y_target_raw = min(n_y_start + int(1.5 * carreau), 1000) - break +def _selected_pdfs(workspace: EvaluationWorkspace, target: Path) -> list[Path]: + workspace.require_directories("Copies") + if target.is_file(): + if target.suffix.casefold() != ".pdf": + raise CliError(f"Target is not a PDF: {target}", ExitCode.INVALID_ARGUMENTS) + return [target] + return sorted(workspace.copies_dir.glob("*.pdf"), key=lambda path: path.name.casefold()) - # RULES 3 & 4: Calculate horizontal boundaries (0.0 to 1.0 fraction of local page width) - col_w = 1000 / doc.page_count - if c_type == "L": # |name - fraction_x0 = (x0_raw % col_w) / col_w - fraction_x1 = 1.0 - end_y_target_raw = min(1000, end_y_target_raw + 40) - elif c_type == "R": # name| - fraction_x0 = 0.0 - # Find the closest 'L' label in y-distance - L_labels = [it for it in parsed_coords if it[1] == "L"] - if L_labels: - closest_L = min(L_labels, key=lambda it: abs(it[3] - y_start_raw)) - closest_L_x_center = (closest_L[5] + closest_L[6]) / 2.0 - fraction_x1 = (closest_L_x_center % col_w) / col_w - if fraction_x1 <= fraction_x0: fraction_x1 = 1.0 # Fallback - else: - fraction_x1 = 1.0 - else: # Normal - fraction_x0 = 0.0 - fraction_x1 = 1.0 - current_p = start_page - while current_p <= end_page: - page = doc[current_p] +def run(workspace: EvaluationWorkspace, target: Path) -> ExitCode: + workspace.require_files("labels") + utils.read_all_labels(workspace.root) + pdf_files = _selected_pdfs(workspace, target) + status = ExitCode.SUCCESS + for pdf_path in pdf_files: + json_path = pdf_path.with_suffix(".json") + if not json_path.is_file(): + print(f"Warning: No JSON found for {pdf_path.name}") + status = ExitCode.PARTIAL + continue + name, coordinates = decode_json(pdf_path) + print(f"Decoded name: {name}") + split_an_interro(workspace, pdf_path, coordinates) + if not pdf_files: + print("No PDF copies found.") + return status - y0 = scale_coord(y_start_raw, page) if current_p == start_page else 0 - y1 = scale_coord(end_y_target_raw, page) if current_p == end_page else page.rect.height - if y1 > y0 + 1: - # Convert fractions to absolute PDF points - x0_pdf = fraction_x0 * page.rect.width - x1_pdf = fraction_x1 * page.rect.width +def build_parser() -> argparse.ArgumentParser: + return target_parser("Split verified PDF copies into answers by label") - temp_path = f"_part_{idx}_{current_p}.pdf" - save_cropped_page(doc, current_p, x0_pdf, y0, x1_pdf, y1, temp_path) - temp_parts.append(temp_path) - current_p += 1 +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() - parts_by_label[clean_label].extend(temp_parts) + def handle(args: argparse.Namespace) -> ExitCode: + workspace, target = workspace_from_target(args) + return run(workspace, target) - output_dir.mkdir(parents=True, exist_ok=True) - - # Process aggregated parts by label - for title, parts in parts_by_label.items(): - merger = PdfWriter() - for part in parts: - if os.path.exists(part): - merger.append(part) - - filename = f"{title}.pdf" - merger.write(output_dir / filename) - merger.close() - generated_files.add(filename) - - # Cleanup - for part in parts: - if os.path.exists(part): - os.remove(part) - - doc.close() - - # Move files not generated in this run to 'Missing' folder - if output_dir.exists(): - missing_dir = output_dir / "Missing" - for item in output_dir.iterdir(): - if item.is_file() and item.name not in generated_files: - print(f"ALERT: File '{item.name}' not generated. Moving to {missing_dir}") - missing_dir.mkdir(exist_ok=True) - item.rename(missing_dir / item.name) + return execute(parser, argv, handle) if __name__ == "__main__": - if len(sys.argv) < 2: - print("Usage: python script.py ") - sys.exit(1) - - input_arg = Path(sys.argv[1]) - - if input_arg.is_file(): - base_dir = input_arg.parent - if base_dir.name == "Copies": - base_dir = base_dir.parent - pdf_files = [input_arg] - elif input_arg.is_dir(): - base_dir = input_arg - copies_dir = base_dir / "Copies" - pdf_files = sorted(copies_dir.glob("*.pdf")) - else: - print(f"Error: {input_arg} is not a valid file or directory.") - sys.exit(1) - - read_all_labels(base_dir) - - for pdf_path in pdf_files: - json_path = pdf_path.with_suffix(".json") - # print("Debug :", json_path) - if json_path.exists(): - (name, coords) = decode_json(pdf_path) - print("Decoded name : ", name) - split_an_interro(base_dir, pdf_path, coords) - else: - print(f"Warning: No JSON found for {pdf_path.name}") + raise SystemExit(main()) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 84943c9..faf7252 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -302,6 +302,10 @@ class StandardCliTests(unittest.TestCase): "reading_grouped_annotations": load_script_module( "reading_grouped_annotations.py", "reading_grouped_annotations" ), + "cutleft": load_script_module("cutleft.py", "cutleft"), + "splitting_int": load_script_module( + "splitting_int.py", "splitting_int" + ), "copies_tools": load_script_module( "copies_tools.py", "copienator_copies_tools_test" ), @@ -339,6 +343,8 @@ class StandardCliTests(unittest.TestCase): "annotating_by_label": [missing], "reading_annotations": [missing], "reading_grouped_annotations": [missing], + "cutleft": [missing], + "splitting_int": [missing], } for name, arguments in invocations.items(): with self.subTest(script=name), redirect_stderr(io.StringIO()): @@ -431,6 +437,16 @@ class StandardCliTests(unittest.TestCase): "grouped", {"target": evaluation, "update_score": True, "refaire": True}, ), + "cutleft": ( + "cutleft", + "default", + {"target": evaluation, "fullpage": True}, + ), + "splitting_int": ( + "splitting", + "default", + {"target": evaluation}, + ), } for module_name, (step_id, variant_id, values) in cases.items(): step = steps[step_id] @@ -513,6 +529,131 @@ class StandardCliTests(unittest.TestCase): self.assertEqual(module.main([str(evaluation)]), 0) self.assertTrue((evaluation / "Par label").is_dir()) + def test_cutleft_can_target_one_copy_without_rendering_at_import(self) -> None: + module = self.modules["cutleft"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_pdf = evaluation / "Copies" / "Copie01.pdf" + copy_pdf.parent.mkdir(parents=True) + copy_pdf.write_bytes(b"pdf") + with patch.object(module, "ImageReviewer") as reviewer: + self.assertEqual(module.main([str(copy_pdf), "--fullpage"]), 0) + files, output_dir = reviewer.call_args.args[:2] + self.assertEqual(files, [copy_pdf]) + self.assertEqual(output_dir, evaluation / "Cutleft") + self.assertEqual(reviewer.call_args.kwargs["default_max_per_file"], 1) + + def test_cutleft_atomic_save_removes_only_obsolete_copy_outputs(self) -> None: + module = self.modules["cutleft"] + with tempfile.TemporaryDirectory() as directory: + output_dir = Path(directory) / "Cutleft" + output_dir.mkdir() + (output_dir / "Copie01_01.jpg").write_bytes(b"old-one") + (output_dir / "Copie01_02.jpg").write_bytes(b"old-two") + (output_dir / "Copie02_01.jpg").write_bytes(b"other-copy") + (output_dir / "Copie01_schema.json").write_text( + "{}", encoding="utf-8" + ) + image = Image.new("RGB", (5, 5), "white") + result = ( + image, + [image], + { + "original_filename": "Copie01.pdf", + "total_pages": 1, + "number_of_files": 1, + "columns_per_file": [1], + }, + ) + module.save_results(result, Path("Copie01.pdf"), output_dir) + self.assertTrue((output_dir / "Copie01_01.jpg").is_file()) + self.assertFalse((output_dir / "Copie01_02.jpg").exists()) + self.assertEqual( + (output_dir / "Copie02_01.jpg").read_bytes(), b"other-copy" + ) + self.assertEqual( + read_json(output_dir / "Copie01_schema.json")["total_pages"], 1 + ) + + def test_splitting_missing_json_is_partial(self) -> None: + module = self.modules["splitting_int"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_pdf = evaluation / "Copies" / "Copie01.pdf" + copy_pdf.parent.mkdir(parents=True) + copy_pdf.write_bytes(b"pdf") + (evaluation / "labels").write_text("Ex 1\n", encoding="utf-8") + self.assertEqual(module.main([str(evaluation)]), 4) + + def test_splitting_failure_preserves_previous_copy_outputs(self) -> None: + module = self.modules["splitting_int"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_pdf = evaluation / "Copies" / "Copie01.pdf" + output_dir = evaluation / "Copies" / "Copie01" + output_dir.mkdir(parents=True) + copy_pdf.write_bytes(b"pdf") + (output_dir / "sentinel.pdf").write_bytes(b"old") + workspace = EvaluationWorkspace(evaluation) + with patch.object( + module, + "_render_split_outputs", + side_effect=RuntimeError("render failed"), + ), self.assertRaises(RuntimeError): + module.split_an_interro(workspace, copy_pdf, []) + self.assertEqual((output_dir / "sentinel.pdf").read_bytes(), b"old") + + def test_splitting_moves_obsolete_outputs_to_missing_on_commit(self) -> None: + module = self.modules["splitting_int"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_pdf = evaluation / "Copies" / "Copie01.pdf" + output_dir = evaluation / "Copies" / "Copie01" + missing_dir = output_dir / "Missing" + missing_dir.mkdir(parents=True) + copy_pdf.write_bytes(b"pdf") + (output_dir / "Old.pdf").write_bytes(b"obsolete") + (missing_dir / "Earlier.pdf").write_bytes(b"earlier") + + def render(_pdf, _coordinates, staging): + (staging / "Ex 1.pdf").write_bytes(b"new") + return {"Ex 1.pdf"} + + with patch.object(module, "_render_split_outputs", side_effect=render): + module.split_an_interro(EvaluationWorkspace(evaluation), copy_pdf, []) + self.assertEqual((output_dir / "Ex 1.pdf").read_bytes(), b"new") + self.assertEqual( + (output_dir / "Missing" / "Old.pdf").read_bytes(), b"obsolete" + ) + self.assertEqual( + (output_dir / "Missing" / "Earlier.pdf").read_bytes(), b"earlier" + ) + + def test_splitting_renders_a_real_one_page_answer(self) -> None: + module = self.modules["splitting_int"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_pdf = evaluation / "Copies" / "Copie01.pdf" + copy_pdf.parent.mkdir(parents=True) + document = module.fitz.open() + page = document.new_page(width=600, height=800) + page.insert_text((100, 200), "Student answer") + document.save(copy_pdf) + document.close() + (evaluation / "labels").write_text("Ex 1\n", encoding="utf-8") + atomic_write_json( + copy_pdf.with_suffix(".json"), + { + "name": "Copie01", + "list": [{"label": "Ex 1", "box_2d": [100, 100, 300, 300]}], + }, + ) + + self.assertEqual(module.main([str(evaluation)]), 0) + answer = evaluation / "Copies" / "Copie01" / "Ex 1.pdf" + self.assertTrue(answer.is_file()) + self.assertEqual(len(PdfReader(answer).pages), 1) + def test_post_correction_main_cleans_json_atomically(self) -> None: module = self.modules["post_correction"] with tempfile.TemporaryDirectory() as directory: