853 lines
33 KiB
Python
853 lines
33 KiB
Python
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.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]
|
|
|
|
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 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(
|
|
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(
|
|
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(
|
|
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)
|
|
|
|
contents_4 = [
|
|
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_4 = types.GenerateContentConfig(
|
|
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(),
|
|
)
|
|
|
|
print(f"Generating rubric (Persp) for group: {', '.join(labels)}...")
|
|
try:
|
|
response_r = client.models.generate_content(
|
|
model=MODEL_ID,
|
|
contents=contents_4,
|
|
config=config_4
|
|
)
|
|
rubrics_data = GroupRubrics.model_validate_json(response_r.text)
|
|
rubrics_map = {r.label: r.rubric_content for r in rubrics_data.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)
|
|
|
|
# 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),
|
|
)
|
|
|
|
return ExitCode.PARTIAL if processing_errors else ExitCode.SUCCESS
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = evaluation_parser("Extract exam and solution code via Gemini")
|
|
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: process_exam(
|
|
workspace_from_args(args),
|
|
restart=args.restart,
|
|
),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|