from __future__ import annotations import argparse import io import json import re from collections.abc import Sequence from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime from pathlib import Path from uuid import uuid4 from google import genai from google.genai import types from pydantic import BaseModel, Field, field_validator from copienator import ( CliError, EvaluationWorkspace, ExitCode, atomic_write_json, atomic_write_text, evaluation_parser, execute, utils, workspace_from_args, ) from copienator import configuration as config from copienator.gemini_config import thinking_config from copienator.filesystem import staged_files from copienator.latex_json import normalize_literal_newlines, repair_latex_json from copienator.platform import validate_windows_labels from copienator.rubrics import present_rubric_labels from copienator.statement_checkpoints import extract_stage, report_error 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 StatementContent(BaseModel): @field_validator("question_content", "solution_content", "context_content", "rubric_content", "content", check_fields=False) @classmethod def normalize_newlines(cls, value: str) -> str: return normalize_literal_newlines(value) class QuestionOnlyItem(StatementContent): 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(StatementContent): 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(StatementContent): 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(StatementContent): label: str = Field(description="Label exact de la question, à conserver sans traduction.") rubric_content: str = Field(description="Barème détaillé sur 4 points, en français, avec formules LaTeX. Échapper chaque antislash LaTeX conformément à la syntaxe JSON.") 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. Il est inutile d'indiquer dans `rubric_content` que le barème totalise 4 points : ce total est toujours implicite. N'utilise pas de caractères mathématiques Unicode dans `rubric_content`. Écris les expressions mathématiques en LaTeX, par exemple `$\\lfloor \\sqrt{k} \\rfloor$` plutôt qu'avec des symboles Unicode. 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`. """ LATEX_JSON_INSTRUCTION = r""" IMPORTANT — Échappement JSON du LaTeX : Dans les chaînes JSON, chaque antislash LaTeX doit être doublé. Exemple d’échappement dans un barème (adapte les clés au schéma demandé) : {"rubrics":[{"label":"1)","rubric_content":"1 point : $\\frac{1}{2}$ et $f \\circ g$.\n1 point : $x \\rightarrow 0$ et $x \\not=0$."}]} Ne confonds pas les commandes LaTeX avec les échappements JSON : \\frac, \\rightarrow, \\not, \\theta et \\begin sont des commandes LaTeX ; \n est réservé à un véritable saut de ligne du texte. N'écris jamais \\n pour un saut de ligne : après décodage JSON, le texte doit contenir un véritable saut de ligne, pas une commande LaTeX \n. """ def validate_rubric_content(content: str) -> str: """Catch decoded control escapes; ordinary JSON newlines remain valid. A misescaped LaTeX command starting with n is indistinguishable from an intentional newline. Do not guess or rewrite those after decoding. """ if any((ord(char) < 32 and char != "\n") or ord(char) == 127 for char in content): positions = ", ".join( f"U+{ord(char):04X} at offset {index}" for index, char in enumerate(content) if (ord(char) < 32 and char != "\n") or ord(char) == 127 ) raise ValueError( "Rubric contains control characters, likely from unescaped LaTeX: " f"{positions}; decoded content: {content!r}" ) return content # --- Modèle fusionné (pour le reste du script) --- class QuestionItem(StatementContent): label: str question_content: str solution_content: str class ContextItem(StatementContent): 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. """ PROMPT_1 += LATEX_JSON_INSTRUCTION PROMPT_2 += LATEX_JSON_INSTRUCTION PROMPT_3 += LATEX_JSON_INSTRUCTION PROMPT_4 += LATEX_JSON_INSTRUCTION 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 _request_rubrics(client, group_context_text: str): 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.", thinking_config=thinking_config(config.THINKING_LEVEL_LITE), response_mime_type="application/json", response_json_schema=GroupRubrics.model_json_schema(), ), ) return response def generate_rubrics( client, group_context_text: str, *, workspace: EvaluationWorkspace | None = None, expected_labels: list[str] | None = None, ) -> dict[str, str]: """Save the raw response before decoding; report failures without hiding text.""" raw = None response_path = None try: response = _request_rubrics(client, group_context_text) raw = response.text if workspace is not None: stamp = datetime.now(UTC).strftime('%Y%m%dT%H%M%S%fZ') cache_path = workspace.root / 'Cache' / 'rubric_responses' / f'{stamp}-{uuid4().hex}.json' atomic_write_text(cache_path, raw or '') response_path = cache_path payload = json.loads(repair_latex_json(raw)) if not isinstance(payload, dict) or not isinstance(payload.get('rubrics'), list): raise TypeError('Gemini response must contain a rubrics list') result = {} item_errors = [] for entry in payload['rubrics']: try: item = RubricItem.model_validate(entry) if item.label in result: raise ValueError('Gemini returned duplicate rubric labels') if expected_labels is not None and item.label not in expected_labels: raise ValueError('Unexpected rubric label') result[item.label] = validate_rubric_content(item.rubric_content) except ValueError as error: label = entry.get('label', 'unknown') if isinstance(entry, dict) else 'unknown' item_errors.append(f"Question {label}: {error}; item (escaped): {entry!r}") if workspace is not None and expected_labels is not None: # Retain valid siblings even if another rubric in this response fails. for label, content in result.items(): if label in expected_labels and content.strip(): atomic_write_text(workspace.root / 'Persp' / label.replace('/', '_'), f'{label}\n{content}') if item_errors: raise ValueError('\n'.join(item_errors)) if expected_labels is not None and ( set(result) != set(expected_labels) or any(not value.strip() for value in result.values()) ): raise CliError("Incomplete or unexpected Gemini rubrics; failed group preserved") return result except Exception as error: diagnostic = ( f"Rubric generation failed for {expected_labels or 'unspecified group'}: {error}\n" f"Raw response (escaped): {raw!r}\n" ) if response_path is not None: diagnostic += f"Raw response saved to: {response_path}\n" print(diagnostic, end='', flush=True) if workspace is not None: log_path = workspace.root / 'Cache' / 'rubric_errors.log' try: log_path.parent.mkdir(parents=True, exist_ok=True) with log_path.open('a', encoding='utf-8') as log: log.write(f"[{datetime.now(UTC).isoformat()}]\n{diagnostic}\n") print(f"Error log: {log_path}", flush=True) except OSError as log_error: print(f"Could not save rubric error log: {log_error}", flush=True) raise 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, delete_persp: bool = False) -> ExitCode: """Regroup questions or generate missing rubrics, retaining completed groups.""" 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) present = set(present_rubric_labels(workspace.root, labels)) if mode == 'persp' else set() if mode == 'persp': print(f"Persp: {len(present)}/{len(labels)} present, {len(labels) - len(present)} missing.") if len(present) == len(labels) and not delete_persp: print('All Persp rubrics are already present; nothing to generate.') return ExitCode.SUCCESS 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), thinking_config=thinking_config(config.THINKING_LEVEL_LITE), 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) if delete_persp: persp_dir = workspace.root / 'Persp' remove = [path.name for path in persp_dir.iterdir() if path.is_file()] if persp_dir.exists() else [] with staged_files(persp_dir, remove=remove): pass present.clear() print('Deleted current Persp; generating all rubrics.') generated = 0 for group in groups: missing = [label for label in group if label not in present] if not missing: print(f"Skipping existing Persp for group: {', '.join(group)}.") continue print(f"Generating missing rubric (Persp): {', '.join(missing)}...") 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(missing) + "\n\n" + "\n\n".join(questions[label] for label in group) ) rubrics = generate_rubrics( api_client, group_content, workspace=workspace, expected_labels=missing, ) with staged_files(workspace.root / "Persp") as staging: for label, rubric in rubrics.items(): (staging / label.replace("/", "_")).write_text( f"{label}\n{rubric}", encoding="utf-8") generated += len(rubrics) print(f"Generated {generated} missing Persp rubrics; all {len(labels)} are now present.") 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) if restart: archive = cache_dir / 'statement_history' / uuid4().hex for path in [*cache_dir.glob('statement_*.json'), *cache_dir.glob('gemini_*.json'), tmp_dir / 'exam_items.txt', tmp_dir / 'exam_items_full.txt']: if path.is_file(): archive.mkdir(parents=True, exist_ok=True) path.replace(archive / path.name) print('Starting a new extraction; previous checkpoints and edits archived.') atomic_write_json(cache_dir / 'statement_reset_outputs.json', {'pending': True}) # 1. Resolve files pdf_path = workspace.statement_pdf_file enonce_path = find_file(folder, "enonce") if enonce_path is None: enonce_path = workspace.statement_source_file correction_path = find_file(folder, "correction") if correction_path is None: correction_path = workspace.correction_source_file missing = [] if not pdf_path.is_file(): missing.append("enonce.pdf") if not enonce_path.is_file(): missing.append("enonce.tex") if not correction_path.is_file(): missing.append("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), thinking_config=thinking_config(config.THINKING_LEVEL_LITE), response_mime_type="application/json", response_json_schema=ExamQuestions.model_json_schema(), ) questions_data = ExamQuestions(questions=extract_stage( folder, 'questions', QuestionOnlyItem, 'question_content', lambda _pending: client.models.generate_content(model=MODEL_ID, contents=contents_1, config=config_1), )) # ========================================== # REQUÊTE 2 : Extraction des Corrections # ========================================== extracted_questions_json = questions_data.model_dump_json(indent=2) config_2 = types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), thinking_config=thinking_config(config.THINKING_LEVEL_LITE), response_mime_type="application/json", response_json_schema=ExamSolutions.model_json_schema(), ) question_labels = [q.label for q in questions_data.questions] def request_solutions(pending): requested = pending if pending is not None else question_labels selected = ExamQuestions(questions=[q for q in questions_data.questions if q.label in requested]) contents = [types.Content(role='user', parts=[ types.Part.from_text(text=PROMPT_2), types.Part.from_text(text='--- QUESTIONS EXTRAITES ---\n' + selected.model_dump_json(indent=2)), types.Part.from_text(text=f'--- CORRECTION SOURCE ({correction_path.name}) ---\n{correction_text}'), ])] return client.models.generate_content(model=MODEL_ID, contents=contents, config=config_2) solutions_data = ExamSolutions(solutions=extract_stage( folder, 'solutions', SolutionOnlyItem, 'solution_content', request_solutions, labels=question_labels, )) # ========================================== # 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), thinking_config=thinking_config(config.THINKING_LEVEL_LITE), response_mime_type="application/json", response_json_schema=ExamContext.model_json_schema(), ) def request_contexts(pending): contents = contents_3 if pending: contents = [types.Content(role='user', parts=[*contents_3[0].parts, types.Part.from_text(text='Produis des contextes UNIQUEMENT pour les labels cibles suivants : ' + ', '.join(pending))])] return client.models.generate_content(model=MODEL_ID, contents=contents, config=config_3) context_data = ExamContext(contexts=extract_stage( folder, 'contexts', ExtractedContext, 'context_content', request_contexts, labels=question_labels, )) grouping_checkpoint = cache_dir / 'statement_grouping.json' if grouping_checkpoint.is_file(): print('Resuming approved question groups from checkpoint.') grouped_extraction = GroupedExamExtraction.model_validate_json(grouping_checkpoint.read_text()) return write_exam_outputs(workspace, grouped_extraction, client) # ========================================== # 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 io.StringIO() as f, io.StringIO() 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") if not items_file.exists(): atomic_write_text(items_file, f.getvalue()) if not full_items_file.exists(): atomic_write_text(full_items_file, f_full.getvalue()) # --- OPEN EDITOR AND PARSE --- while True: print('Check the question list against enonce.pdf before approving: ' 'verify completeness, labels, order and context coverage.') 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 if len(all_new_q_labels) != len(questions_only): raise CliError('The edited groups must retain every extracted question. ' 'Grouping edits were preserved; correct the list and rerun.') 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) if not labels_list or len(labels_list) != len(set(labels_list)): raise CliError('Approved groups must contain unique question labels') atomic_write_json(grouping_checkpoint, grouped_extraction.model_dump()) return write_exam_outputs(workspace, grouped_extraction, client, restart=restart) def write_exam_outputs(workspace: EvaluationWorkspace, grouped_extraction: GroupedExamExtraction, client, *, restart: bool = False) -> ExitCode: folder = workspace.root labels_list = [item.label for group in grouped_extraction.groups for item in group if isinstance(item, QuestionItem)] text_dir, sol_dir, text2_dir, sol2_dir, persp_dir = [ folder / name for name in ('Text', 'Sol', 'Text2', 'Sol2', 'Persp')] reset_marker = folder / 'Cache' / 'statement_reset_outputs.json' if restart or reset_marker.exists(): archive = folder / 'Cache' / 'statement_history' / uuid4().hex for directory in (text_dir, sol_dir, text2_dir, sol2_dir, persp_dir): if directory.exists(): archive.mkdir(parents=True, exist_ok=True) directory.replace(archive / directory.name) reset_marker.unlink(missing_ok=True) for directory in (text_dir, sol_dir, text2_dir, sol2_dir, persp_dir): directory.mkdir(parents=True, exist_ok=True) atomic_write_text(workspace.labels_file, ''.join(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))) def save_missing(path: Path, text: str): if not path.is_file() or not path.read_text(encoding='utf-8').strip(): atomic_write_text(path, text) present = set(present_rubric_labels(folder, labels_list)) context_sources = {} 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) missing_rubrics = [label for label in labels if label not in present] try: rubrics_map = {} if missing_rubrics: print(f"Generating missing rubric (Persp): {', '.join(missing_rubrics)}...") rubrics_map = generate_rubrics( client, group_context_text + '\nProduis des barèmes UNIQUEMENT pour : ' + ', '.join(missing_rubrics), workspace=workspace, expected_labels=missing_rubrics, ) 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) safe_label = item.label.replace('/', '_') save_missing(sol_dir / safe_label, f'{item.label}\n{item.solution_content}') save_missing(text2_dir / f'{safe_label}.tex', f'\\textbf{{{item.label}}} {item.question_content}') save_missing(sol2_dir / f'{safe_label}.tex', f'\\textbf{{{item.label}}} {item.solution_content}') # --- Écriture du Barème (Persp) --- if item.label in rubrics_map: atomic_write_text(persp_dir / safe_label, f"{item.label}\n{rubrics_map[item.label]}") 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 context_sources.setdefault(ctxt_path, []) if item.content not in context_sources[ctxt_path]: context_sources[ctxt_path].append(item.content) save_missing(text_dir / safe_group_filename, '\n'.join(text_content_lines)) for path, contents in context_sources.items(): save_missing(path, '\n\n'.join(contents)) print(f"Wrote {len(grouped_extraction.groups)} groups; checking PDF compilation...") # ========================================== # 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: pdf_path = tex_file.with_suffix('.pdf') status_path = folder / 'Cache' / 'statement_pdf_status' / tex_file.parent.name / (tex_file.stem + '.json') failed_before = status_path.exists() and not json.loads(status_path.read_text()).get('complete') if (pdf_path.is_file() and pdf_path.stat().st_size > 0 and pdf_path.stat().st_mtime_ns >= tex_file.stat().st_mtime_ns and not failed_before): return None atomic_write_json(status_path, {'complete': False}) content = tex_file.read_text(encoding='utf-8') compile_to_pdf(content, pdf_path, strict=True) atomic_write_json(status_path, {'complete': True}) 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: report_error(folder, 'PDF compilation', error) processing_errors.extend(compile_errors) 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="Generate only missing Persp rubrics for existing groups") parser.add_argument('--delete-persp', action='store_true', help='Delete current Persp before generation (requires --persp-only)') parser.add_argument( "--restart", action="store_true", help="Start over; archive previous extraction checkpoints, edits and generated outputs", ) return parser def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() def run(args): if args.delete_persp and not args.persp_only: raise CliError('--delete-persp requires --persp-only', ExitCode.INVALID_ARGUMENTS) if args.groups_only or args.persp_only: kwargs = {'delete_persp': True} if args.delete_persp else {} return refine_existing(workspace_from_args(args), 'groups' if args.groups_only else 'persp', **kwargs) return process_exam(workspace_from_args(args), restart=args.restart) return execute( parser, argv, run, ) if __name__ == "__main__": raise SystemExit(main())