import sys import os import json import shutil import concurrent.futures import threading import img2pdf from reportlab.pdfgen import canvas # Fix for Matplotlib in threads: Set backend to non-interactive 'Agg' import matplotlib matplotlib.use('Agg') from PIL import Image, ImageDraw, ImageFont import annotating from annotating import MARGIN_LEFT, ANNOT_WIDTH # Global lock for Matplotlib/Latex rendering to prevent race conditions LATEX_LOCK = threading.Lock() DPI = 100 BOX_SIZE = 30 SCORE_BOX_SIZE = 40 SCORES = [x * 0.5 for x in range(9)] # 0.0 to 4.0 try: CHECKBOX_FONT = ImageFont.truetype("DejaVuSans.ttf", 20) except IOError: try: CHECKBOX_FONT = ImageFont.truetype("arial.ttf", 20) except IOError: 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] def safe_render_latex(*args, **kwargs): """Thread-safe wrapper for latex rendering.""" # with LATEX_LOCK: # return annotating.render_latex_text(*args, **kwargs) return annotating.render_real_latex_text(*args, **kwargs) class CheckboxRenderer: def __init__(self, label_name): self.label = label_name self.checkboxes = [] # List of {type, box, etc.} def callback(self, kind, draw, pos, meta): """ Called by compose_label_image during rendering. pos contains {x, y, w, h} or {box}. meta contains {data, index, etc.} """ if kind == "header_item": # meta['data'] is either result object (for score) or feedback object if meta.get("type") == "score": # Draw score boxes start_x = pos['w'] + 20 for val in SCORES: box = draw_checkbox(draw, start_x, pos['y'] + 25, SCORE_BOX_SIZE, str(val)) self.checkboxes.append({ "type": "score", "label": self.label, "value": val, "rel_box": box # Will be adjusted for global Y later }) start_x += SCORE_BOX_SIZE + 60 elif meta.get("type") == "global_fb": # Draw delete box for global feedback bx = pos['w'] - BOX_SIZE - 5 by = pos['y'] + 5 box = draw_checkbox(draw, bx, by, 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": # Delete rect checkbox b = pos['box'] # [xmin, ymin, xmax, ymax] box = draw_checkbox(draw, b[2] - BOX_SIZE, b[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": # Delete whole local feedback checkbox bx = pos['x'] + pos['w'] - BOX_SIZE by = pos['y'] box = draw_checkbox(draw, bx, by, BOX_SIZE) self.checkboxes.append({ "type": "del_local", "label": self.label, "index": meta["index"], "final_box": box, "text_preview": meta["data"]["text"][:20] }) from utils import natural_key def process_student(args): """Thread worker: Processes one student.""" root_dir, student_id, labels, overwrite = args output_dir = os.path.join(root_dir, "Bnot", f"Copie{student_id}") if os.path.exists(output_dir): if not overwrite: print(f"Skipping {student_id}: Output already exists.") return shutil.rmtree(output_dir) print(f"Generating Checkable PDF for: {student_id}") os.makedirs(output_dir) label_images = [] # ... (rest of the function remains exactly the same) all_checkboxes = [] bnote_entries = [] # For bnote.json sorted_labels = sorted(labels.items(), key=lambda x: natural_key(x[0])) for label, content in sorted_labels: pdf_path = content['pdf_path'] if not os.path.exists(pdf_path): continue base_img, _, _ = annotating.make_base_image(pdf_path) # Initialize the hook cb_renderer = CheckboxRenderer(label) # Render using the shared engine final_img, header_h = annotating.compose_label_image( base_img, label, content['result'], content['coordinates'][0], draw_callback=cb_renderer.callback ) if final_img == None: continue label_images.append(final_img) all_checkboxes.append(cb_renderer.checkboxes) bnote_entries.append({ "id": student_id, "label": label, "header_height": header_h, # hmin/hmax will be filled during concatenation "img_h": final_img.height }) if not label_images: return # Concatenate max_w = max(i.width for i in label_images) total_h = sum(i.height for i in label_images) concat_img = Image.new("RGB", (max_w, total_h), "white") final_json_map = [] current_y = 0 for idx, (img, boxes) in enumerate(zip(label_images, all_checkboxes)): concat_img.paste(img, (0, current_y)) bnote_entries[idx]["hmin"] = current_y bnote_entries[idx]["hmax"] = current_y + img.height del bnote_entries[idx]["img_h"] # Clean up temp data # Adjust coordinates for concatenated image for item in boxes: # item might have 'rel_box' (header) or 'final_box' (local) # Both were relative to the label image. We just add current_y. b = item.get('final_box') or item.get('rel_box') item['global_box'] = [b[0], b[1] + current_y, b[2], b[3] + current_y] final_json_map.append(item) current_y += img.height bnote_data = { "width": max_w, "height": total_h, "images": bnote_entries } with open(os.path.join(output_dir, "bnote.json"), "w") as f: json.dump(bnote_data, f, indent=2) with open(os.path.join(output_dir, "checkboxes.json"), "w") as f: json.dump(final_json_map, f, indent=2) temp_img_path = os.path.join(output_dir, "Reference.jpg") # Can't use png here concat_img.save(temp_img_path, quality=90) pdf_path = os.path.join(output_dir, "Concat.pdf") w, h = concat_img.size c = canvas.Canvas(pdf_path, pagesize=(w, h)) c.drawImage(temp_img_path, 0, 0, width=w, height=h) c.save() import argparse # Added if __name__ == "__main__": parser = argparse.ArgumentParser(description="Generate annotated PDFs.") parser.add_argument("input_path", help="Directory or specific file path") parser.add_argument("--overwrite", action="store_true", help="Overwrite existing output files") args = parser.parse_args() input_path = args.input_path overwrite = args.overwrite # Capture flag target_id = None # Detect if input is a specific file if os.path.isfile(input_path): root_dir = os.path.dirname(input_path) or "." # Extract ID from filename (e.g., Copie40.pdf -> 40) match = re.search(r'Copie(\d+)', os.path.basename(input_path)) if match: target_id = match.group(1) else: print("Error: Could not extract student ID from filename.") sys.exit(1) else: root_dir = input_path results = annotating.make_dictionary(root_dir) # Filter results if a specific target ID was requested if target_id: if target_id in results: results = {target_id: results[target_id]} else: print(f"Student ID {target_id} not found in directory scan.") results = {} tasks = sorted([(root_dir, sid, lbls, overwrite) for sid, lbls in results.items()]) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: results = executor.map(process_student, tasks) try: for _ in results: pass except Exception: import traceback traceback.print_exc()