from __future__ import annotations import argparse import concurrent.futures import re from collections.abc import Sequence from pathlib import Path from typing import Any import matplotlib matplotlib.use("Agg") from PIL import Image, ImageFont from reportlab.pdfgen import canvas from copienator import ( CliError, EvaluationWorkspace, ExitCode, atomic_write_json, execute, read_json, target_parser, utils, workspace_from_target, ) from copienator.annotation_data import load_annotation_data from copienator.annotation_documents import MAX_HEIGHT_PX, MAX_PIXELS, copy_documents, document_fits from copienator.commands import annotating from copienator.commands.import_annotations import clear_import_directory from copienator.filesystem import staged_directory from copienator.utils import natural_key BOX_SIZE = 30 SCORE_BOX_SIZE = 40 SCORES = [value * 0.5 for value in range(10)] EXPECTED_OUTPUTS = ("bnote.json", "checkboxes.json", "Reference.jpg", "Concat.pdf") try: CHECKBOX_FONT = ImageFont.truetype("DejaVuSans.ttf", 20) except OSError: try: CHECKBOX_FONT = ImageFont.truetype("arial.ttf", 20) except OSError: CHECKBOX_FONT = ImageFont.load_default() def draw_checkbox(draw, x, y, size=BOX_SIZE, label=None, fill="white"): if label: draw.text( (x - BOX_SIZE - 5, y + 2), str(label), fill="black", font=CHECKBOX_FONT ) draw.rectangle([x, y, x + size, y + size], fill=fill, outline="black", width=2) return [x, y, x + size, y + size] class CheckboxRenderer: def __init__(self, label_name): self.label = label_name self.checkboxes = [] def callback(self, kind, draw, pos, meta): if kind == "header_item": if meta.get("type") == "score": start_x = pos["w"] + 20 for value in SCORES: box = draw_checkbox( draw, start_x, pos["y"] + 25, SCORE_BOX_SIZE, str(value), ) self.checkboxes.append( { "type": "score", "label": self.label, "value": value, "rel_box": box, } ) start_x += SCORE_BOX_SIZE + 45 start_x += SCORE_BOX_SIZE + 60 box = draw_checkbox( draw, start_x, pos["y"] + 25, SCORE_BOX_SIZE, "clr", ) self.checkboxes.append( {"type": "clear_all", "label": self.label, "rel_box": box} ) elif meta.get("type") == "global_fb": box = draw_checkbox( draw, pos["w"] - BOX_SIZE - 5, pos["y"] + 5, BOX_SIZE, ) self.checkboxes.append( { "type": "del_global", "label": self.label, "index": meta["index"], "rel_box": box, "text_preview": meta["data"]["text"][:20], } ) elif kind == "local_rect": rectangle = pos["box"] box = draw_checkbox( draw, rectangle[2] - BOX_SIZE, rectangle[1], BOX_SIZE, ) self.checkboxes.append( { "type": "del_local_rect", "label": self.label, "index": meta["index"], "final_box": box, "text_preview": meta["data"]["text"][:20], } ) elif kind == "local_text": box = draw_checkbox( draw, pos["x"] + pos["w"] - BOX_SIZE, pos["y"], BOX_SIZE, ) self.checkboxes.append( { "type": "del_local", "label": self.label, "index": meta["index"], "final_box": box, "text_preview": meta["data"]["text"][:20], } ) def _output_complete(output_dir: Path) -> bool: return all( (document / name).is_file() for document in copy_documents(output_dir) for name in EXPECTED_OUTPUTS ) def _render_student( workspace: EvaluationWorkspace, student_id: str, labels: dict[str, dict[str, Any]], *, overwrite: bool, output_mode: str, output_root: Path | None = None, ) -> str: output_dir = ( output_root if output_root is not None else workspace.annotation_dir(output_mode) ) / f"Copie{student_id}" if _output_complete(output_dir) and not overwrite: print(f"Skipping {student_id}: output is complete.") return "skipped" print(f"Generating checkable PDF for: {student_id}") label_images: list[Image.Image] = [] checkbox_groups: list[list[dict[str, Any]]] = [] bnote_entries: list[dict[str, Any]] = [] problems = False for label, content in sorted(labels.items(), key=lambda item: natural_key(item[0])): pdf_path = Path(content["pdf_path"]) if not pdf_path.exists(): print(f"Warning: answer PDF not found: {pdf_path}") problems = True continue base_image, _, _ = annotating.make_base_image(pdf_path) checkbox_renderer = CheckboxRenderer(label) final_image, header_height = annotating.compose_label_image( base_image, label, content["result"], content["coordinates"][0], draw_callback=checkbox_renderer.callback, source_pdf=pdf_path, ) if final_image is None: problems = True continue label_images.append(final_image) checkbox_groups.append(checkbox_renderer.checkboxes) bnote_entries.append( { "id": student_id, "label": label, "header_height": header_height, "layout": final_image.info.get('annotation_layout', annotating.layout_for(None)), } ) if not label_images: print(f"Warning: no annotations could be rendered for Copie{student_id}") return "partial" batches = [] start = 0 height = 0 width = 0 for index, image in enumerate(label_images): if not document_fits(image.width, image.height, MAX_HEIGHT_PX, MAX_PIXELS): raise CliError( f"Copie{student_id} ({bnote_entries[index]['label']}) is " f"{image.width} × {image.height} px ({image.width * image.height} pixels); " f"a single answer cannot be split. Document limits are " f"{MAX_HEIGHT_PX} px high and {MAX_PIXELS} pixels." ) if not document_fits(max(width, image.width), height + image.height, MAX_HEIGHT_PX, MAX_PIXELS): batches.append((start, index)) start = index height = 0 width = 0 height += image.height width = max(width, image.width) batches.append((start, len(label_images))) with staged_directory(output_dir) as staging: parts = [] for index, (start, end) in enumerate(batches, start=1): destination = staging if len(batches) > 1: name = f"Part{index}" parts.append(name) destination /= name _save_document(destination, label_images[start:end], checkbox_groups[start:end], bnote_entries[start:end]) if parts: atomic_write_json(staging / "bnote.json", { "parts": parts, "images": bnote_entries, }) return "partial" if problems else "success" def _save_document( output_dir: Path, label_images: list[Image.Image], checkbox_groups: list[list[dict[str, Any]]], bnote_entries: list[dict[str, Any]], ) -> None: max_width = max(image.width for image in label_images) total_height = sum(image.height for image in label_images) if not document_fits(max_width, total_height, MAX_HEIGHT_PX, MAX_PIXELS): raise CliError( f"Document would be {max_width} × {total_height} px " f"({max_width * total_height} pixels); document limits are " f"{MAX_HEIGHT_PX} px high and {MAX_PIXELS} pixels." ) concatenated = Image.new("RGB", (max_width, total_height), "white") checkbox_map: list[dict[str, Any]] = [] current_y = 0 for index, (image, checkboxes) in enumerate( zip(label_images, checkbox_groups, strict=True) ): concatenated.paste(image, (0, current_y)) bnote_entries[index]["hmin"] = current_y bnote_entries[index]["hmax"] = current_y + image.height for item in checkboxes: box = item.get("final_box") or item.get("rel_box") item["global_box"] = [ box[0], box[1] + current_y, box[2], box[3] + current_y, ] checkbox_map.append(item) current_y += image.height output_dir.mkdir(parents=True, exist_ok=True) atomic_write_json( output_dir / "bnote.json", {"width": max_width, "height": total_height, "images": bnote_entries}, ) atomic_write_json(output_dir / "checkboxes.json", checkbox_map) reference = output_dir / "Reference.jpg" concatenated.save(reference, quality=90) pdf_path = output_dir / "Concat.pdf" pdf_canvas = canvas.Canvas(str(pdf_path), pagesize=(max_width, total_height)) pdf_canvas.drawImage( str(reference), 0, 0, width=max_width, height=total_height, ) pdf_canvas.save() def _copy_id_from_target(workspace: EvaluationWorkspace, target: Path) -> str | None: if target == workspace.root: return None match = re.search(r"Copie(\d+)", target.name) if match is None: raise CliError(f"Could not extract a copy id from target: {target}") return match.group(1) def _load_refaire(workspace: EvaluationWorkspace): workspace.require_files("refaire.json") loaded = read_json(workspace.refaire_file) if not isinstance(loaded, list): raise CliError("refaire.json must contain a JSON array") return loaded def run( workspace: EvaluationWorkspace, target: Path, *, overwrite: bool = False, refaire: bool = False, clear_import: bool = False, ) -> ExitCode: workspace.require_files("labels", "correction.json") workspace.require_directories("Copies", "Par label") utils.read_all_labels(workspace.root) copy_id = _copy_id_from_target(workspace, target) refaire_list = _load_refaire(workspace) if refaire else None loaded = load_annotation_data( workspace, refaire_list=refaire_list, copy_id=None if refaire else copy_id, ) for warning in loaded.warnings: print(f"Warning: {warning}") if not loaded.data: print("Warning: no annotation data was found.") return ExitCode.PARTIAL output_mode = "refaire" if refaire else "checks" tasks = sorted(loaded.data.items(), key=lambda item: natural_key(item[0])) if refaire: output_root = workspace.annotation_dir("refaire") if output_root.exists() and not overwrite: raise CliError( "BRnot already exists; use --overwrite to replace the previous redo." ) class IncompleteRedo(Exception): pass try: with staged_directory(output_root) as staging: with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: futures = [ executor.submit( _render_student, workspace, student_id, labels, overwrite=True, output_mode="refaire", output_root=staging, ) for student_id, labels in tasks ] statuses = [future.result() for future in futures] if loaded.warnings or any(status != "success" for status in statuses): raise IncompleteRedo except IncompleteRedo: print("Warning: incomplete redo generation; previous BRnot was preserved.") return ExitCode.PARTIAL if clear_import: clear_import_directory(workspace) return ExitCode.SUCCESS statuses: list[str] = [] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: futures = [ executor.submit( _render_student, workspace, student_id, labels, overwrite=overwrite, output_mode=output_mode, ) for student_id, labels in tasks ] for future in futures: statuses.append(future.result()) if loaded.warnings or "partial" in statuses: return ExitCode.PARTIAL if clear_import: clear_import_directory(workspace) return ExitCode.SUCCESS def build_parser() -> argparse.ArgumentParser: parser = target_parser("Generate annotated PDFs with checkboxes.") parser.add_argument("--clear-import", action="store_true", help="Empty IMPORT_DIR after successful annotation generation") parser.add_argument( "--overwrite", action="store_true", help="Replace existing outputs" ) parser.add_argument( "--refaire", action="store_true", help="Process only entries from refaire.json", ) return parser def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() def handler(args: argparse.Namespace) -> ExitCode: workspace, target = workspace_from_target(args) return run( workspace, target, overwrite=args.overwrite, refaire=args.refaire, clear_import=args.clear_import, ) return execute(parser, argv, handler) if __name__ == "__main__": raise SystemExit(main())