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 import annotating import utils from copienator import ( CliError, EvaluationWorkspace, ExitCode, atomic_write_json, execute, read_json, target_parser, workspace_from_target, ) from copienator.annotation_data import load_annotation_data from copienator.filesystem import staged_directory from 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((output_dir / name).is_file() for name in EXPECTED_OUTPUTS) def _render_student( workspace: EvaluationWorkspace, student_id: str, labels: dict[str, dict[str, Any]], *, overwrite: bool, output_mode: str, ) -> str: output_dir = 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, ) if final_image is None: continue label_images.append(final_image) checkbox_groups.append(checkbox_renderer.checkboxes) bnote_entries.append( { "id": student_id, "label": label, "header_height": header_height, "img_h": final_image.height, } ) if not label_images: print(f"Warning: no annotations could be rendered for Copie{student_id}") return "partial" max_width = max(image.width for image in label_images) total_height = sum(image.height for image in label_images) 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 del bnote_entries[index]["img_h"] 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 with staged_directory(output_dir) as staging: atomic_write_json( staging / "bnote.json", {"width": max_width, "height": total_height, "images": bnote_entries}, ) atomic_write_json(staging / "checkboxes.json", checkbox_map) reference = staging / "Reference.jpg" concatenated.save(reference, quality=90) pdf_path = staging / "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() return "partial" if problems else "success" 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, ) -> 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])) 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 return ExitCode.SUCCESS def build_parser() -> argparse.ArgumentParser: parser = target_parser("Generate annotated PDFs with checkboxes.") 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, ) return execute(parser, argv, handler) if __name__ == "__main__": raise SystemExit(main())