from __future__ import annotations import argparse import re from collections.abc import Sequence from concurrent.futures import ThreadPoolExecutor from pathlib import Path from google import genai from google.genai import types from pydantic import BaseModel, Field from copienator import configuration as config from copienator import utils from copienator import ( CliError, EvaluationWorkspace, ExitCode, atomic_write_text, evaluation_parser, execute, workspace_from_args, ) from copienator.platform import validate_windows_labels from copienator.filesystem import staged_directory from copienator.utils import compile_to_pdf def get_lcp(s1: str, s2: str) -> str: i = 0 while i < len(s1) and i < len(s2) and s1[i] == s2[i]: i += 1 lcp = s1[:i] if ')' in s1 or ')' in s2: last_paren = lcp.rfind(')') if last_paren != -1: return lcp[:last_paren + 1] return lcp MODEL_ID = config.MODEL_LITE_ID api_key = config.API_KEY # --- Modèles pour la Requête 1 --- class QuestionOnlyItem(BaseModel): label: str = Field(description="Label unique de la question (par exemple '1.a' ou 'Exercice 1').") question_content: str = Field(description="Texte source de la question, extrait exactement du fichier d’énoncé, SANS le label lui-même.") class ExamQuestions(BaseModel): questions: list[QuestionOnlyItem] # --- Modèles pour la Requête 2 --- class SolutionOnlyItem(BaseModel): label: str = Field(description="Label exact de la question fourni en entrée, à conserver sans traduction.") solution_content: str = Field(description="Texte source de la solution, extrait exactement du fichier de correction.") class ExamSolutions(BaseModel): solutions: list[SolutionOnlyItem] # --- Modèles pour la Requête 3 --- class ExtractedContext(BaseModel): target_question_label: str = Field(description="Label exact de la PREMIÈRE question située immédiatement APRÈS cette information dans l’énoncé.") last_question_label: str = Field(description="Label exact de la DERNIÈRE question qui utilise cette information.") context_content: str = Field(description="Texte source des définitions, notations ou hypothèses, extrait de l’énoncé.") class ExamContext(BaseModel): contexts: list[ExtractedContext] # --- Modèles pour la Requête 4 (Barèmes) --- class RubricItem(BaseModel): label: str = Field(description="Label exact de la question, à conserver sans traduction.") rubric_content: str = Field(description="Barème détaillé sur 4 points : toutes les consignes, explications et justifications doivent être rédigées en français.") class GroupRubrics(BaseModel): rubrics: list[RubricItem] class LabelGroups(BaseModel): groups: list[list[str]] PROMPT_4 = """Je te fournis les questions, le contexte éventuel, et les corrections pour un groupe de questions d'un examen. Ta tâche : Établir un barème de correction détaillé pour CHAQUE question. Rédige intégralement en français le contenu de chaque champ `rubric_content`, y compris les consignes de notation, les explications et les justifications, même si certains textes fournis sont dans une autre langue. Conserve les formules mathématiques, les labels exacts des questions et les clés JSON `rubrics`, `label` et `rubric_content` sans les traduire. Chaque question DOIT être notée sur exactement 4 points. Propose une répartition logique de ces points. Par exemple : - Au moins 2 points si le résultat est correct. - Mettre la moitié des points si le raisonnement est correct mais pas le résultat. - Retirer 1,5 point si les hypothèses d'un théorème ou d'une question précédente ne sont pas vérifiées. Renvoie uniquement un objet JSON contenant une liste `rubrics`. Pour chaque question fournie, cette liste contient un objet avec son `label` exact et son barème en français dans `rubric_content`. """ # --- Modèle fusionné (pour le reste du script) --- class QuestionItem(BaseModel): label: str question_content: str solution_content: str class ContextItem(BaseModel): target_question_label: str last_question_label: str content: str # Juste une string encapsulée pour le différencier facilement class ExamExtraction(BaseModel): items: list[QuestionItem | ContextItem] # Liste mixte class GroupedExamExtraction(BaseModel): groups: list[list[QuestionItem | ContextItem]] PROMPT_1 = """Je te fournis : 1. Le PDF d'un examen (`enonce.pdf`). 2. Le code source de ses questions (fichier `enonce`). Ta tâche : 1. Identifie tous les labels distincts des questions à l'aide du PDF. Ils doivent être uniques : utilise par exemple `Ex 1 : 1)a)` ou `I)1)b)`. 2. Pour chaque label, extrais exactement le texte de la question correspondante dans le fichier source `enonce`. N'inclus ni le label lui-même, ni les commandes de liste LaTeX comme `item`, ni les marques de liste org-mode comme `2.`. Ne reformule pas et ne traduis pas le texte extrait ; conserve le LaTeX. Renvoie les questions dans l'ordre exact de lecture du document, dans la liste `questions` de l'objet JSON attendu. Conserve les clés `label` et `question_content`. """ PROMPT_2 = """Je te fournis : 1. Une liste JSON des labels des questions d'un examen et de leurs textes. 2. Le code source du corrigé de l'examen (fichier `correction`). Pour chaque label fourni, extrais exactement le texte de la solution correspondante dans le fichier source `correction`. Ne reformule pas et ne traduis pas le texte extrait ; conserve le LaTeX. Renvoie les solutions dans le même ordre que les questions, dans la liste `solutions` de l'objet JSON attendu. Conserve les clés `label` et `solution_content` ainsi que les labels exacts des questions. """ PROMPT_3 = """Je te fournis : 1. Une liste JSON des labels des questions d'un examen et de leurs textes. 2. Le code source des questions de l'examen (fichier `enonce`). Extrais les informations importantes nécessaires à la compréhension des questions, mais qui ne font PAS partie des textes des questions : définitions des objets, notations générales, hypothèses ou contexte. Ces informations figurent souvent dans un \\item précédent qui ne constitue pas lui-même une question, mais contient une liste de questions. Par exemple, dans ce code LaTeX : \\item Soient N et M deux matrices qui commutent. \\begin{itemize} \\item Montrer que N et M ont un vecteur propre commun. \\item Montrer que N et M sont simultanément trigonalisables. \\end{itemize} La phrase « Soient N et M deux matrices qui commutent » n'est pas une question ; elle est nécessaire pour comprendre les deux questions suivantes. Pour chaque information extraite, identifie : 1. Le label de la PREMIÈRE question située immédiatement APRÈS cette information dans l'énoncé (`target_question_label`). 2. Le label de la DERNIÈRE question qui utilise cette information (`last_question_label`). Conserve le texte source dans `context_content`, sans le reformuler ni le traduire, et conserve le LaTeX ainsi que les labels exacts. Renvoie le résultat dans la liste `contexts` de l'objet JSON attendu. """ def find_file(folder: Path, base_name: str) -> Path | None: for ext in [".org", ".tex"]: path = folder / f"{base_name}{ext}" if path.is_file(): return path return None def generate_rubrics(client, group_context_text: str) -> dict[str, str]: """Use the same rubric request for full and selective statement generation.""" response = client.models.generate_content( model=MODEL_ID, contents=[types.Content(role="user", parts=[ types.Part.from_text(text=PROMPT_4), types.Part.from_text(text=f"--- CONTENU DU GROUPE ---\n{group_context_text}"), ])], config=types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), system_instruction="Rédige tous les barèmes et consignes de notation en français. Conserve les clés JSON, les labels et les formules mathématiques.", temperature=0.2, response_mime_type="application/json", response_json_schema=GroupRubrics.model_json_schema(), ), ) rubrics = GroupRubrics.model_validate_json(response.text).rubrics if len({item.label for item in rubrics}) != len(rubrics): raise ValueError("Gemini returned duplicate rubric labels") return {item.label: item.rubric_content for item in rubrics} def validate_groups(groups: list[list[str]], labels: list[str]) -> None: flattened = [label for group in groups for label in group] if (not groups or any(not group for group in groups) or len(flattened) != len(set(flattened)) or set(flattened) != set(labels)): raise CliError("Groups must contain every existing label exactly once") def refine_existing(workspace: EvaluationWorkspace, mode: str, *, api_client=None) -> ExitCode: """Regroup or replace rubrics without regenerating statements or solutions.""" labels = workspace.read_labels() if not labels or len(labels) != len(set(labels)): raise CliError("Generate unique question labels before refining the statement") validate_windows_labels(labels) questions = {} for label in labels: safe_label = label.replace("/", "_") parts = [] for directory, title in (("Text2", "Question"), ("Sol2", "Correction")): path = workspace.root / directory / f"{safe_label}.tex" if not path.is_file(): raise CliError(f"Missing {path}; generate statements and solutions first") parts.append(f"{title} [{label}]:\n{path.read_text(encoding='utf-8')}") questions[label] = "\n".join(parts) context = utils.enonce_total(workspace.root) if api_client is None: if not api_key: raise CliError("GEMINI_API_KEY is not configured") api_client = genai.Client(api_key=api_key) if mode == "groups": response = api_client.models.generate_content( model=MODEL_ID, contents=[types.Content(role="user", parts=[types.Part.from_text(text=( "Regroupe ces questions d’examen en groupes cohérents pour la correction " "et l’annotation, selon leurs dépendances et leur contexte commun. " "Ne mélange pas des exercices différents. Conserve l’ordre des questions. " "Chaque label doit apparaître exactement une fois, sans modification. " "Renvoie uniquement un objet JSON groups contenant des listes de labels.\n\n" + context + "\n\n" + "\n\n".join(questions.values()) ))])], config=types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), temperature=0.1, response_mime_type="application/json", response_json_schema=LabelGroups.model_json_schema(), ), ) groups = LabelGroups.model_validate_json(response.text).groups validate_groups(groups, labels) atomic_write_text(workspace.label_groups_file, "".join(", ".join(group) + "\n" for group in groups)) print(f"Updated label_groups: {len(groups)} Gemini groups.") elif mode == "persp": if not workspace.label_groups_file.is_file(): raise CliError("Generate label_groups before generating rubrics") groups = [[label.strip() for label in line.split(",") if label.strip()] for line in workspace.label_groups_file.read_text(encoding="utf-8").splitlines() if line.strip()] validate_groups(groups, labels) with staged_directory(workspace.root / "Persp") as staging: for group in groups: print(f"Generating rubric (Persp) for group: {', '.join(group)}...") group_content = ( "Contexte général de l’examen, fourni uniquement pour comprendre les questions :\n" + context + "\n\nProduis des barèmes UNIQUEMENT pour les labels suivants : " + ", ".join(group) + "\n\n" + "\n\n".join(questions[label] for label in group) ) rubrics = generate_rubrics(api_client, group_content) if set(rubrics) != set(group) or any(not value.strip() for value in rubrics.values()): raise CliError("Incomplete or unexpected Gemini rubrics; previous Persp preserved") for label, rubric in rubrics.items(): (staging / label.replace("/", "_")).write_text( f"{label}\n{rubric}", encoding="utf-8") print("Replaced Persp with Gemini rubrics.") else: raise ValueError(f"Unknown statement refinement: {mode}") return ExitCode.SUCCESS def process_exam( workspace: EvaluationWorkspace, restart: bool = False, *, api_client=None, ) -> ExitCode: folder = workspace.root cache_dir = folder / "Cache" tmp_dir = folder / "Tmp" cache_dir.mkdir(exist_ok=True) tmp_dir.mkdir(exist_ok=True) cache_q_file = cache_dir / "gemini_questions.json" cache_s_file = cache_dir / "gemini_solutions.json" cache_c_file = cache_dir / "gemini_context.json" # 1. Resolve files pdf_path = folder / "enonce.pdf" enonce_path = find_file(folder, "enonce") correction_path = find_file(folder, "correction") missing = [] if not pdf_path.is_file(): missing.append("enonce.pdf") if not enonce_path: missing.append("enonce.org or enonce.tex") if not correction_path: missing.append("correction.org or correction.tex") if missing: raise CliError( f"Missing files in {folder}: {', '.join(missing)}", ExitCode.INVALID_WORKSPACE, ) print("Reading files...") pdf_bytes = pdf_path.read_bytes() enonce_text = enonce_path.read_text(encoding="utf-8") correction_text = correction_path.read_text(encoding="utf-8") if api_client is None: if not api_key: raise CliError("GEMINI_API_KEY is not configured") api_client = genai.Client(api_key=api_key) client = api_client # ========================================== # REQUÊTE 1 : Extraction des Énoncés # ========================================== contents_1 = [ types.Content( role="user", parts=[ types.Part.from_text(text=PROMPT_1), types.Part.from_bytes(data=pdf_bytes, mime_type="application/pdf"), types.Part.from_text(text=f"--- ENONCE SOURCE ({enonce_path.name}) ---\n{enonce_text}"), ], ) ] config_1 = types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), temperature=0.1, response_mime_type="application/json", response_json_schema=ExamQuestions.model_json_schema(), ) if cache_q_file.is_file() and not restart: print("Loading cached questions from Cache/gemini_questions.json...") response_q_text = cache_q_file.read_text(encoding="utf-8") else: print("Sending request 1 (Questions) to Gemini...") response_q = client.models.generate_content( model=MODEL_ID, contents=contents_1, config=config_1 ) response_q_text = response_q.text print("Saving questions to cache...") atomic_write_text(cache_q_file, response_q_text) questions_data = ExamQuestions.model_validate_json(response_q_text) # ========================================== # REQUÊTE 2 : Extraction des Corrections # ========================================== extracted_questions_json = questions_data.model_dump_json(indent=2) contents_2 = [ types.Content( role="user", parts=[ types.Part.from_text(text=PROMPT_2), types.Part.from_text(text=f"--- QUESTIONS EXTRAITES ---\n{extracted_questions_json}"), types.Part.from_text(text=f"--- CORRECTION SOURCE ({correction_path.name}) ---\n{correction_text}"), ], ) ] config_2 = types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), temperature=0.1, response_mime_type="application/json", response_json_schema=ExamSolutions.model_json_schema(), ) if cache_s_file.is_file() and not restart: print("Loading cached solutions from Cache/gemini_solutions.json...") response_s_text = cache_s_file.read_text(encoding="utf-8") else: print("Sending request 2 (Solutions) to Gemini...") response_s = client.models.generate_content( model=MODEL_ID, contents=contents_2, config=config_2 ) response_s_text = response_s.text print("Saving solutions to cache...") atomic_write_text(cache_s_file, response_s_text) solutions_data = ExamSolutions.model_validate_json(response_s_text) # ========================================== # REQUÊTE 3 : Extraction du Contexte (Notations, etc.) # ========================================== contents_3 = [ types.Content( role="user", parts=[ types.Part.from_text(text=PROMPT_3), types.Part.from_text(text=f"--- QUESTIONS EXTRAITES ---\n{extracted_questions_json}"), types.Part.from_text(text=f"--- ENONCE SOURCE ({enonce_path.name}) ---\n{enonce_text}"), ], ) ] config_3 = types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), temperature=0.1, response_mime_type="application/json", response_json_schema=ExamContext.model_json_schema(), ) if cache_c_file.is_file() and not restart: print("Loading cached context from Cache/gemini_context.json...") response_c_text = cache_c_file.read_text(encoding="utf-8") else: print("Sending request 3 (Context) to Gemini...") response_c = client.models.generate_content( model=MODEL_ID, contents=contents_3, config=config_3 ) response_c_text = response_c.text print("Saving context to cache...") atomic_write_text(cache_c_file, response_c_text) context_data = ExamContext.model_validate_json(response_c_text) # ========================================== # FUSION des trois résultats # ========================================== sol_map = {s.label: s.solution_content for s in solutions_data.solutions} # Map labels to their index to validate ordering label_to_idx = {q.label: i for i, q in enumerate(questions_data.questions)} for c in context_data.contexts: first_idx = label_to_idx.get(c.target_question_label, -1) last_idx = label_to_idx.get(c.last_question_label, -1) # Enforce LAST is after (or equal to) FIRST if first_idx != -1 and last_idx != -1 and last_idx < first_idx: print(f"Warning: LAST question ({c.last_question_label}) is before FIRST ({c.target_question_label}). Fixing.") c.last_question_label = c.target_question_label elif last_idx == -1: # Fallback if invalid c.last_question_label = c.target_question_label # Grouper les contextes par label cible ctx_map = {} for c in context_data.contexts: if c.target_question_label in ctx_map: ctx_map[c.target_question_label]['content'] += "\n\n" + c.context_content # Keep the furthest LAST question label curr_last = ctx_map[c.target_question_label]['last'] if label_to_idx.get(c.last_question_label, -1) > label_to_idx.get(curr_last, -1): ctx_map[c.target_question_label]['last'] = c.last_question_label else: ctx_map[c.target_question_label] = { 'content': c.context_content, 'last': c.last_question_label } merged_items = [] for q in questions_data.questions: if q.label in ctx_map: merged_items.append(ContextItem( target_question_label=q.label, last_question_label=ctx_map[q.label]['last'], content=ctx_map[q.label]['content'] )) sol_content = sol_map.get(q.label, "") merged_items.append(QuestionItem( label=q.label, question_content=q.question_content, solution_content=sol_content )) extracted_data = ExamExtraction(items=merged_items) # ========================================== # INITIAL GROUPING COMPUTATION # ========================================== items_file = tmp_dir / "exam_items.txt" full_items_file = tmp_dir / "exam_items_full.txt" trunc_map = {} # --- INITIAL GROUPING COMPUTATION --- # 1. Normalize labels first for item in extracted_data.items: if isinstance(item, QuestionItem): item.label = item.label.replace("Exercice", "Ex").replace(".", ")") # 2. Extract questions and compute grouping indices questions_only = [item for item in extracted_data.items if isinstance(item, QuestionItem)] # questions_only = [item for item in extracted_data.items if isinstance(item, QuestionItem)] q_group_indices = [] if questions_only: n = len(questions_only) if n == 1: q_group_indices = [[0]] else: adj_lcp = [get_lcp(questions_only[i].label, questions_only[i+1].label) for i in range(n - 1)] current_g = [0] for i in range(n - 1): p = adj_lcp[i] prev_p = adj_lcp[i - 1] if i > 0 else "" next_p = adj_lcp[i + 1] if i < n - 2 else "" # Group i and i+1 together if p is non-empty and at least as specific as adjacent LCPs if p and len(p) >= len(prev_p) and len(p) >= len(next_p): current_g.append(i + 1) else: q_group_indices.append(current_g) current_g = [i + 1] q_group_indices.append(current_g) # Build list of unique ContextItems from extracted data all_contexts = [item for item in extracted_data.items if isinstance(item, ContextItem)] initial_groups = [] for g_indices in q_group_indices: group_items = [] first_q_idx = g_indices[0] for q_idx in g_indices: q_item = questions_only[q_idx] # 1. Collect contexts targeting this specific question # 2. Or contexts carried over from an earlier group (only added at the start of the group) for ctx in all_contexts: target_idx = label_to_idx.get(ctx.target_question_label, -1) last_idx = label_to_idx.get(ctx.last_question_label, -1) if target_idx != -1 and last_idx != -1: is_exact_target = (target_idx == q_idx) is_carried_over = (q_idx == first_q_idx and target_idx < first_q_idx and last_idx >= first_q_idx) if is_exact_target or is_carried_over: group_items.append(ContextItem( target_question_label=ctx.target_question_label, last_question_label=ctx.last_question_label, content=ctx.content )) group_items.append(q_item) initial_groups.append(group_items) # ---- Transform labels, and check uniqueness seen_labels = set() label_updates = {} for group in initial_groups: for item in group: if isinstance(item, QuestionItem): orig_label = item.label # 2. Ensure uniqueness (prefix with XX) while item.label in seen_labels: item.label = f"XX{item.label}" seen_labels.add(item.label) label_updates[orig_label] = item.label # Sync the modified labels to ContextItem for group in initial_groups: for item in group: if isinstance(item, ContextItem): item.target_question_label = label_updates.get(item.target_question_label, item.target_question_label) item.last_question_label = label_updates.get(item.last_question_label, item.last_question_label) # --- WRITE TEXT FILES --- print(f"Writing items files to {items_file.name} and {full_items_file.name}...") with open(items_file, "w", encoding="utf-8") as f, \ open(full_items_file, "w", encoding="utf-8") as f_full: header = "# Edit labels. Modify groups (---). Ensure label uniqueness (XX). Duplicate CONTEXT.\n\n" f.write(header) f_full.write(header) for g_idx, group in enumerate(initial_groups): if g_idx > 0: f.write("\n---\n\n") f_full.write("\n---\n\n") for item in group: if isinstance(item, QuestionItem): safe_content = item.question_content.replace('\n', ' \\n ') f_full.write(f"{item.label} ### {safe_content}\n") if len(safe_content) > 65: trunc_content = safe_content[:64] + "…" trunc_map[trunc_content] = safe_content else: trunc_content = safe_content f.write(f"{item.label} ### {trunc_content}\n") elif isinstance(item, ContextItem): safe_content = item.content.replace('\n', ' \\n ') f.write(f"CONTEXT ### {safe_content}\n") f_full.write(f"CONTEXT ### {safe_content}\n") # --- OPEN EDITOR AND PARSE --- while True: print("Opening items file for editing...") utils.edit_file_and_enter(items_file) print(f"Parsing edited items from {items_file.name}...") with open(items_file, "r", encoding="utf-8") as f: edited_lines = [line.strip() for line in f if line.strip() and not line.startswith("#")] # 1. Validation for XX labels has_xx = False for line in edited_lines: if " ### " in line: lbl = line.split(" ### ", 1)[0].strip() if lbl.startswith("XX"): has_xx = True break if has_xx: print("\n!!! ERROR: Some labels still start with 'XX'. Please remove the 'XX' prefixes to ensure unique, valid labels.") input("Press ENTER to return to the editor...") continue # Map original contexts by normalized content orig_contexts = {c.context_content.strip(): c for c in context_data.contexts} # 2. Actual Parsing grouped_items = [] all_new_q_labels = [] # Pass 1: Read all edited lines and collect question labels in sequence for line in edited_lines: if line == "---" or " ### " not in line: continue lbl, _content_raw = line.split(" ### ", 1) lbl = lbl.strip() if lbl != "CONTEXT": all_new_q_labels.append(lbl) # Mapping from original question index to new label idx_to_new_label = {i: all_new_q_labels[i] for i in range(min(len(questions_only), len(all_new_q_labels)))} orig_q_idx = 0 current_group = [] for line in edited_lines: if line == "---": if current_group: grouped_items.append(current_group) current_group = [] continue if " ### " not in line: continue new_label, edited_content_raw = line.split(" ### ", 1) new_label = new_label.strip() if "…" in edited_content_raw and edited_content_raw in trunc_map: edited_content_raw = trunc_map[edited_content_raw] edited_content = edited_content_raw.replace(' \\n ', '\n') if new_label == "CONTEXT": current_group.append(('CONTEXT', edited_content)) else: sol_content = questions_only[orig_q_idx].solution_content if orig_q_idx < len(questions_only) else "" current_group.append(QuestionItem( label=new_label, question_content=edited_content, solution_content=sol_content )) orig_q_idx += 1 if current_group: grouped_items.append(current_group) # Pass 2: Resolve ContextItem target/last labels per group final_grouped_items = [] for group in grouped_items: final_group = [] q_in_group = [item for item in group if isinstance(item, QuestionItem)] g_first_label = q_in_group[0].label if q_in_group else "" g_last_label = q_in_group[-1].label if q_in_group else "" for i, item in enumerate(group): if isinstance(item, tuple) and item[0] == 'CONTEXT': c_text = item[1] norm_text = c_text.strip() # Find next question label in group following this context next_q_label = g_first_label for successor in group[i+1:]: if isinstance(successor, QuestionItem): next_q_label = successor.label break if norm_text in orig_contexts: orig_c = orig_contexts[norm_text] orig_target_idx = label_to_idx.get(orig_c.target_question_label, -1) orig_last_idx = label_to_idx.get(orig_c.last_question_label, -1) mapped_target = idx_to_new_label.get(orig_target_idx, next_q_label) mapped_last = idx_to_new_label.get(orig_last_idx, g_last_label) # Check if context's last question is BEFORE the first question of this group first_q_idx_in_exam = all_new_q_labels.index(g_first_label) if g_first_label in all_new_q_labels else -1 last_q_idx_in_exam = all_new_q_labels.index(mapped_last) if mapped_last in all_new_q_labels else -1 if last_q_idx_in_exam != -1 and first_q_idx_in_exam != -1 and last_q_idx_in_exam < first_q_idx_in_exam: mapped_last = g_last_label final_group.append(ContextItem( target_question_label=mapped_target, last_question_label=mapped_last, content=c_text )) else: # New context created by user final_group.append(ContextItem( target_question_label=next_q_label, last_question_label=g_last_label, content=c_text )) else: final_group.append(item) final_grouped_items.append(final_group) grouped_items = final_grouped_items break labels_list = [item.label for group in grouped_items for item in group if isinstance(item, QuestionItem)] validate_windows_labels(labels_list) # Save labels and proceed grouped_extraction = GroupedExamExtraction(groups=grouped_items) # 2. Setup output directories text_dir = folder / "Text" sol_dir = folder / "Sol" text2_dir = folder / "Text2" sol2_dir = folder / "Sol2" persp_dir = folder / "Persp" dirs = [text_dir, sol_dir, text2_dir, sol2_dir, persp_dir] import shutil # Ask only if any directory already exists if any(d.exists() for d in dirs): answer = input( "Output directories already exist. Delete their contents? [y/N] " ).strip().lower() if answer not in ("y", "yes"): raise CliError("Output replacement aborted", ExitCode.INVALID_ARGUMENTS) # Empty each directory for d in dirs: if d.exists(): shutil.rmtree(d) d.mkdir(parents=True) else: # Create them if they don't exist for d in dirs: d.mkdir(parents=True) text_dir.mkdir(exist_ok=True) sol_dir.mkdir(exist_ok=True) print("Writing grouped question and solution files...") processing_errors = [] for group in grouped_extraction.groups: q_items = [item for item in group if isinstance(item, QuestionItem)] labels = [q.label for q in q_items] if not labels: continue # Skip if a group has no questions (only contexts) # ========================================== # REQUÊTE 4 : Génération du Barème pour le groupe # ========================================== group_text_parts = [] for item in group: if isinstance(item, QuestionItem): group_text_parts.append(f"Question [{item.label}]:\n{item.question_content}\nCorrection [{item.label}]:\n{item.solution_content}") elif isinstance(item, ContextItem): group_text_parts.append(f"Contexte (Cible: {item.target_question_label}):\n{item.content}") group_context_text = "\n\n---\n\n".join(group_text_parts) print(f"Generating rubric (Persp) for group: {', '.join(labels)}...") try: rubrics_map = generate_rubrics(client, group_context_text) except Exception as e: # noqa: BLE001 - remote API boundary print(f"Error generating rubric for group {labels[0]}: {e}") processing_errors.append(str(e)) rubrics_map = {} # 1. Compute the common prefix for the group prefix = labels[0] for lbl in labels[1:]: prefix = get_lcp(prefix, lbl) # 2. Format the Text filename: prefix [label1, label2] labels_str = ",".join([label[len(prefix):] for label in labels]) group_filename = f"{prefix}[{labels_str}]" safe_group_filename = group_filename.replace("/", "_") text_content_lines = [] # 3. Process each item in the group for item in group: if isinstance(item, QuestionItem): # 1. Prepare tabulated content: # Start with a tab, then replace every newline+whitespace with newline+tab raw_content = item.question_content.strip() tabulated = "\t" + re.sub(r'\n\s*', '\n\t', raw_content) # 2. Build Text entry text_content_lines.append(f"{item.label} :") text_content_lines.append(tabulated) # Write individual Sol file (remains unchanged) safe_label = item.label.replace("/", "_") with open(sol_dir / safe_label, "w", encoding="utf-8") as f_sol: f_sol.write(f"{item.label}\n{item.solution_content}") with open(text2_dir / f"{safe_label}.tex", "w", encoding="utf-8") as f_t2: f_t2.write(f"\\textbf{{{item.label}}} {item.question_content}") with open(sol2_dir / f"{safe_label}.tex", "w", encoding="utf-8") as f_s2: f_s2.write(f"\\textbf{{{item.label}}} {item.solution_content}") # --- Écriture du Barème (Persp) --- rubric_text = rubrics_map.get(item.label, "") with open(persp_dir / safe_label, "w", encoding="utf-8") as f_persp: f_persp.write(f"{item.label}\n{rubric_text}") elif isinstance(item, ContextItem): raw_ctx = item.content.strip() tabulated_ctx = "\t" + re.sub(r'\n\s*', '\n\t', raw_ctx) text_content_lines.append("CONTEXT :") text_content_lines.append(tabulated_ctx) # --- Save context to Text2 (Concatenating if exists) --- safe_first = item.target_question_label.replace("/", "_") safe_last = item.last_question_label.replace("/", "_") ctxt_filename = f"CTXT {safe_first} -> {safe_last}.tex" ctxt_path = text2_dir / ctxt_filename # If file exists, prepend some spacing before appending prefix = "\n\n" if ctxt_path.exists() else "" with open(ctxt_path, "a", encoding="utf-8") as f_c2: f_c2.write(prefix + item.content) # 4. Write the grouped Text file with open(text_dir / safe_group_filename, "w", encoding="utf-8") as f_text: f_text.write("\n".join(text_content_lines)) print(f"Success! Processed {len(grouped_extraction.groups)} groups.") # ========================================== # PDF COMPILATION (4 Threads) # ========================================== all_tex_files = list(text2_dir.glob("*.tex")) + list(sol2_dir.glob("*.tex")) def compile_worker(tex_file: Path) -> str | None: """Helper to read content and call the utility function.""" try: content = tex_file.read_text(encoding="utf-8") pdf_path = tex_file.with_suffix(".pdf") compile_to_pdf(content, pdf_path) except Exception as e: # noqa: BLE001 - compiler worker boundary return f"Error compiling {tex_file.name}: {e}" return None print(f"Compiling {len(all_tex_files)} files to PDF using 4 threads...") with ThreadPoolExecutor(max_workers=4) as executor: compile_errors = [ error for error in executor.map(compile_worker, all_tex_files) if error ] for error in compile_errors: print(error) processing_errors.extend(compile_errors) atomic_write_text( workspace.labels_file, "".join(f"{label}\n" for label in labels_list), ) atomic_write_text( workspace.label_groups_file, "".join(", ".join(item.label for item in group if isinstance(item, QuestionItem)) + "\n" for group in grouped_extraction.groups if any(isinstance(item, QuestionItem) for item in group)), ) return ExitCode.PARTIAL if processing_errors else ExitCode.SUCCESS def build_parser() -> argparse.ArgumentParser: parser = evaluation_parser("Extract exam and solution code via Gemini") actions = parser.add_mutually_exclusive_group() actions.add_argument("--groups-only", action="store_true", help="Regroup existing questions with Gemini; update only label_groups") actions.add_argument("--persp-only", action="store_true", help="Replace only Persp with Gemini rubrics for existing groups") parser.add_argument( "--restart", action="store_true", help="Ignore cached Gemini extraction responses", ) return parser def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() return execute( parser, argv, lambda args: refine_existing(workspace_from_args(args), "groups" if args.groups_only else "persp") if args.groups_only or args.persp_only else process_exam( workspace_from_args(args), restart=args.restart, ), ) if __name__ == "__main__": raise SystemExit(main())