from __future__ import annotations import argparse import concurrent.futures from collections import defaultdict from collections.abc import Sequence from pathlib import Path from typing import Any from PIL import Image, ImageDraw import annotating import utils from copienator import ( EvaluationWorkspace, ExitCode, atomic_write_json, evaluation_parser, execute, read_json, workspace_from_args, ) from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data from copienator.filesystem import staged_files from reading_annotations import ( concatenate, detect_checks_and_notes, has_significant_notes, ) LabelNotes = dict[str, dict[str, Any]] ScanResult = tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes]] def get_extra_pdfs_as_images( root_dir: str | Path, label: str, annotating_module: Any, all_labels: list[str], ) -> list[Image.Image]: """Convert the context, question and solution PDFs associated with a label.""" paths = [ *utils.pdf_images_of_contexts(root_dir, label, all_labels), utils.pdf_image_of_enonce(root_dir, label), utils.pdf_image_of_solution(root_dir, label), ] images = [] for path in paths: if path: image, _, _ = annotating_module.make_base_image(path) if image is not None: images.append(image) return images def save_paginated_pdf(image_groups: list[list[Image.Image]], output_path: Path) -> None: """Paginate vertically concatenated image groups and save them as a PDF.""" non_empty = [group for group in image_groups if group] if not non_empty: return max_width = max(image.width for group in non_empty for image in group) max_page_height = int(max_width * 1.414 * 1.25) border = int((0.2 / 2.54) * 100) left_margin = int((0.3 / 2.54) * 100) vertical_margin = int((0.2 / 2.54) * 100) max_content_height = max_page_height - 2 * vertical_margin pages: list[Image.Image] = [] page_images: list[Image.Image] = [] page_height = 0 def finish_page() -> None: nonlocal page_images, page_height if not page_images: return page = Image.new( "RGB", (max_width + left_margin, page_height + 2 * vertical_margin), "white", ) current_y = vertical_margin for image in page_images: page.paste(image, (left_margin, current_y)) current_y += image.height pages.append(page) page_images = [] page_height = 0 for group in non_empty: processed: list[Image.Image] = [] for index, image in enumerate(group): if index in (0, 1): image = image.copy() color = "black" if index == 0 else "blue" ImageDraw.Draw(image).rectangle( [0, 0, image.width - 1, image.height - 1], outline=color, width=border, ) processed.append(image) group_height = sum(image.height for image in processed) if page_images and page_height + group_height > max_content_height: finish_page() page_images.extend(processed) page_height += group_height finish_page() pages[0].save( output_path, "PDF", resolution=100.0, save_all=True, append_images=pages[1:], ) def _scan_annotation_directory( directory: Path, only_ids: set[str] | None = None, default_student_id: str | None = None, ) -> ScanResult: bnote_path = directory / "bnote.json" if not bnote_path.is_file(): raise FileNotFoundError(f"Missing {bnote_path}") bnote = read_json(bnote_path) if not isinstance(bnote, dict): raise TypeError(f"Expected a JSON object in {bnote_path}") images = [item for item in bnote.get("images", []) if isinstance(item, dict)] if only_ids and not any( str(item.get("id", default_student_id)) in only_ids for item in images ): return {}, {} actions, notes_image = detect_checks_and_notes(directory) if notes_image is None: return {}, {} actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list) notes_by_student: dict[str, LabelNotes] = defaultdict(dict) for action in actions: raw_student_id = action.get("student_id", default_student_id) if raw_student_id is not None: actions_by_student[str(raw_student_id)].append(action) for image_info in images: student_id = str(image_info.get("id", default_student_id or "")) label = str(image_info.get("label", "")) hmin = int(image_info.get("hmin", 0)) hmax = int(image_info.get("hmax", 0)) if student_id and label and hmax > hmin: crop = notes_image.crop((0, hmin, notes_image.width, hmax)) if has_significant_notes(crop): notes_by_student[student_id][label] = { "img": crop, "old_header_h": int(image_info.get("header_height", 0)), } return dict(actions_by_student), dict(notes_by_student) def _merge_scan_result( target_actions: dict[str, list[dict[str, Any]]], target_notes: dict[str, LabelNotes], result: ScanResult, ) -> None: actions, notes = result for student_id, student_actions in actions.items(): target_actions[student_id].extend(student_actions) for student_id, student_notes in notes.items(): target_notes[student_id].update(student_notes) def apply_actions_and_regenerate_grouped( workspace: EvaluationWorkspace, data: AnnotationData, student_id: str, actions: list[dict[str, Any]], label_notes: LabelNotes, all_labels: list[str], *, update_score: bool = False, ) -> tuple[ExitCode, str]: """Apply grouped annotations and atomically merge regenerated student files.""" logs = [f"\nProcessing compilation for: Copie{student_id}"] output_dir = workspace.annotation_dir("grouped") / f"Copie{student_id}" labels_data = data.get(student_id, {}) dirty_labels = apply_checkbox_actions(labels_data, actions, logs.append) if update_score: dirty_labels |= apply_score_overrides( labels_data, output_dir / "score.json", logs.append ) scores = dict.fromkeys(all_labels, "") dirty_images: dict[str, Image.Image] = {} concat_images: list[Image.Image] = [] filtered_groups: list[list[Image.Image]] = [] incomplete = False for label, content in sorted(labels_data.items(), key=lambda item: utils.natural_key(item[0])): result = content["result"] scores[label] = str(result.get("score", 0)) pdf_path = Path(content["pdf_path"]) if not pdf_path.is_file(): logs.append(f" Missing answer PDF: {pdf_path}") incomplete = True continue base_image, _, _ = annotating.make_base_image(pdf_path) final_image, new_header_height = annotating.compose_label_image( base_image, label, result, content["coordinates"][0], with_error=False, ) if final_image is None: incomplete = True continue has_notes = False if label in label_notes: sub_note = label_notes[label]["img"] old_header_height = int(label_notes[label]["old_header_h"]) has_notes = has_significant_notes(sub_note) if has_notes: width, height = sub_note.size if old_header_height > 0: header = sub_note.crop((0, 0, width, min(height, old_header_height))) final_image.paste(header, (0, 0), mask=header) if height > old_header_height: body = sub_note.crop((0, old_header_height, width, height)) final_image.paste(body, (0, new_header_height), mask=body) if label in dirty_labels or has_notes: dirty_images[label] = final_image logs.append(f" Saved dirty image: {label}.jpg") concat_images.append(final_image) feedbacks = result.get("feedback", []) perfect = float(scores[label]) >= 4.0 and all( feedback.get("to_delete", False) for feedback in feedbacks ) if not perfect or has_notes: extras = get_extra_pdfs_as_images( workspace.root, label, annotating, all_labels ) filtered_groups.append([*extras, final_image]) concat_image = concatenate(concat_images) with staged_files(output_dir) as staging: for label, image in dirty_images.items(): image.save(staging / f"{label}.jpg") atomic_write_json(staging / "score.json", scores) if concat_image is not None: concat_image.save(staging / "Concat.jpg") if filtered_groups: save_paginated_pdf(filtered_groups, staging / "Concat_F.pdf") logs.append(f" Saved regenerated files in {output_dir}") status = ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS return status, "\n".join(logs) def _read_refaire(workspace: EvaluationWorkspace) -> tuple[RefaireList, dict[str, list[str]]]: loaded = read_json(workspace.refaire_file) if not isinstance(loaded, list): raise TypeError("refaire.json must contain a JSON array") entries: RefaireList = [] by_student: dict[str, list[str]] = {} for entry in loaded: if not isinstance(entry, list) or len(entry) != 2 or not isinstance(entry[1], list): raise TypeError(f"Malformed refaire entry: {entry!r}") copy_name, labels = entry student_id = str(copy_name).removeprefix("Copie") normalized_labels = [str(label) for label in labels] entries.append([str(copy_name), normalized_labels]) by_student[student_id] = normalized_labels return entries, by_student def run( workspace: EvaluationWorkspace, *, refaire: bool = False, update_score: bool = False, ) -> ExitCode: workspace.require_files("labels", "correction.json") workspace.require_directories("Copies", "Par label", "BGnot") refaire_list: RefaireList | None = None refaire_by_student: dict[str, list[str]] = {} if refaire: workspace.require_files("refaire.json") workspace.require_directories("BRnot") refaire_list, refaire_by_student = _read_refaire(workspace) all_labels = utils.read_all_labels(workspace.root) loaded = load_annotation_data(workspace, refaire_list=refaire_list) for warning in loaded.warnings: print(f"Warning: {warning}") if not loaded.data: print("No annotation data found.") return ExitCode.PARTIAL actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list) notes_by_student: dict[str, LabelNotes] = defaultdict(dict) only_ids = set(refaire_by_student) or None group_dirs = [ path for path in workspace.annotation_dir("grouped").iterdir() if path.is_dir() and not path.name.startswith("Copie") ] with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: futures = [ executor.submit(_scan_annotation_directory, path, only_ids) for path in group_dirs ] for future in concurrent.futures.as_completed(futures): _merge_scan_result(actions_by_student, notes_by_student, future.result()) refaire_incomplete = False if refaire: for student_id, requested_labels in refaire_by_student.items(): selected = requested_labels or list(loaded.data.get(student_id, {})) selected_set = set(selected) directory = workspace.annotation_dir("refaire") / f"Copie{student_id}" if not directory.is_dir(): print(f"Warning: missing refaire annotation directory {directory}") refaire_incomplete = True continue actions_by_student[student_id] = [ action for action in actions_by_student[student_id] if str(action.get("label")) not in selected_set ] for label in selected: notes_by_student[student_id].pop(label, None) refaire_actions, refaire_notes = _scan_annotation_directory( directory, default_student_id=student_id ) for action in refaire_actions.get(student_id, []): if str(action.get("label")) in selected_set: actions_by_student[student_id].append(action) for label, note in refaire_notes.get(student_id, {}).items(): if label in selected_set: notes_by_student[student_id][label] = note status = ( ExitCode.PARTIAL if loaded.warnings or refaire_incomplete else ExitCode.SUCCESS ) student_ids = list(refaire_by_student) if refaire else sorted(loaded.data, key=utils.natural_key) with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: futures = { executor.submit( apply_actions_and_regenerate_grouped, workspace, loaded.data, student_id, actions_by_student[student_id], notes_by_student[student_id], all_labels, update_score=update_score, ): student_id for student_id in student_ids if student_id in loaded.data } for future in concurrent.futures.as_completed(futures): result, output = future.result() print(output) if result != ExitCode.SUCCESS: status = ExitCode.PARTIAL return status def build_parser() -> argparse.ArgumentParser: parser = evaluation_parser("Read grouped annotations and regenerate copies") parser.add_argument( "--refaire", action="store_true", help="Use refaire.json and merge annotations from BRnot", ) parser.add_argument( "--update-score", action="store_true", help="Override generated scores with values from existing score.json files", ) return parser def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() def handle(args: argparse.Namespace) -> ExitCode: return run( workspace_from_args(args), refaire=args.refaire, update_score=args.update_score, ) return execute(parser, argv, handle) if __name__ == "__main__": raise SystemExit(main())