From fd35675e4ccbf86a07ca1a0b0a8cd9cf92f89854 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Thu, 9 Jul 2026 15:54:36 +0200 Subject: [PATCH] enonce_info.py --- Readme.org | 30 +- enonce_info.py | 62 +--- gemini_for_enonce.py | 561 +++++++++++++++++++++++++++------ prompting.py | 25 +- reading_grouped_annotations.py | 11 +- utils.py | 110 +++++++ 6 files changed, 619 insertions(+), 180 deletions(-) diff --git a/Readme.org b/Readme.org index 48b84fd..8d2acc8 100644 --- a/Readme.org +++ b/Readme.org @@ -1,7 +1,7 @@ #+title: Script #+author: Sébastien Miquel #+date: 14-03-2026 -# Time-stamp: <29-06-26 13:05> +# Time-stamp: <09-07-26 15:54> #+OPTIONS: * Méta @@ -72,17 +72,19 @@ export GEMINI_API_KEY=… 1. Créer un fichier =names= dans le dossier courant, avec les noms/prénoms des élèves, un par ligne 2. Créer un dossier correspondant à l'évaluation, comme =Interro= - 3. Mettre les fichiers pdfs scannés dans =Interro=. - 4. Dans le dossier =Interro= créer un fichier =labels= avec les labels + 3. Mettre l'énoncé, au format pdf, et l'énoncé et le corrigé au + format .tex dans le dossier. + 4. Mettre les fichiers pdfs scannés dans =Interro=. + 5. Dans le dossier =Interro= créer un fichier =labels= avec les labels des questions, comme =Ex 1 : 1)a)=, un par ligne. - 5. Il faudra créer des dossiers =Text=, =Sol= et =Persp=, et dans ces + 6. Il faudra créer des dossiers =Text=, =Sol= et =Persp=, et dans ces dossiers créer, pour chaque label (ou groupe de labels : par exemple un seul fichier =Ex 1= peut être utilisé pour toutes les questions de l'exercice 1) un fichier texte qui contient respectivement l'énoncé, un corrigé, et des indications de comment corriger (Gemini met une note sur 4, on peut demander 2 points pour tel truc, etc) - 6. Suivre les étapes plus bas. + 7. Suivre les étapes plus bas. * Étapes et Script ** Prétraitement de l'énoncé @@ -92,8 +94,16 @@ export GEMINI_API_KEY=… + `enonce.tex` + `correction.tex`. - `python gemini_for_enonce.py Interro` - Se charge de créer des dossiers `Text` et `Sol` avec - + Le fichier `Text` contient + Se charge de créer + + un fichier `labels` avec les labels des questions + + Un dossier `Text` avec le contenu textuel des questions, + regroupées. + + Un dossier `Sol` avec le contenu textuel du corrigé, question par + question + + Un dossier `Text2`, qui compile un fichier `.tex` pour chaque + question + + Un dossier `Sol2`, qui compile un fichier `.tex` pour chaque + correction de chaque question. ** Prétraitement des copies @@ -123,8 +133,10 @@ export GEMINI_API_KEY=… 1. =python enonce_info.py Interro= (gestion perso) OU - 2. =python gemini_for_enonce.py Interro= - + Nécessite =enonce.tex/org= et `correction.tex/org` + 1. =python gemini_for_enonce.py Interro= + + Nécessite `enonce.pdf`, =enonce.tex/org= et `correction.tex/org` + + Génère : `Text` and `Sol` folders. Todo : `Text2`. + `python gemini_for_enonce.py Interro --reread` will regenerate the ** Labelisation et regroupement diff --git a/enonce_info.py b/enonce_info.py index e6ea5bb..5cf4e19 100644 --- a/enonce_info.py +++ b/enonce_info.py @@ -8,57 +8,7 @@ import subprocess import tempfile import shutil -def compile_to_pdf(text, output_pdf_path): # 21 cm + 3.8 (dimension de la marge de gauche) - """Wraps text in a standalone template and compiles it to PDF.""" - latex_template = f"""\\documentclass[varwidth=24.8cm,margin=0.4cm]{{standalone}} -\\usepackage[utf8]{{inputenc}} -\\usepackage[T1]{{fontenc}} -\\usepackage{{lmodern}} -\\usepackage{{amsmath, amssymb}} -\\usepackage{{commands}} -\\usepackage{{minted}} -\\usepackage{{graphicx}} -\\usepackage{{enumitem}} -\\begin{{document}} - \\begin{{minipage}}{{24.8cm}} -{text} - \\end{{minipage}} -\\end{{document}} -""" - with tempfile.TemporaryDirectory() as temp_dir: - tex_filename = 'text.tex' - pdf_filename = 'text.pdf' - tex_path = os.path.join(temp_dir, tex_filename) - - with open(tex_path, 'w', encoding='utf-8') as f: - f.write(latex_template) - - # Set TEXINPUTS so pdflatex can find commands.sty if it's in the current dir - # env = os.environ.copy() - # current_dir = os.getcwd() - # env['TEXINPUTS'] = f".:{current_dir}:" - - try: - subprocess.run( - ['pdflatex', '-interaction=nonstopmode', tex_filename], - cwd=temp_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False - ) - if "minted" in text: - subprocess.run( - ['pdflatex', '-interaction=nonstopmode', tex_filename], - cwd=temp_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False) - - generated_pdf = os.path.join(temp_dir, pdf_filename) - if os.path.exists(generated_pdf): - shutil.move(generated_pdf, output_pdf_path) - except Exception as e: - print(f"Compilation error for {output_pdf_path}: {e}") +from utils import compile_to_pdf def fetch_and_save_sub_text(ex_id, indices, label, text_path): """Fetches text for a specific sub-question and saves it to Text/{label}.tex""" @@ -194,7 +144,9 @@ def process_directory(directory): # Prepare output directories paths = { 'Text': os.path.join(directory, "Text"), + 'Text2': os.path.join(directory, "Text2"), 'Sol': os.path.join(directory, "Sol"), + 'Sol2': os.path.join(directory, "Sol2"), 'Persp': os.path.join(directory, "Persp") } for p in paths.values(): @@ -240,15 +192,15 @@ def process_directory(directory): if not indexes: label = f"Ex {current_ex_num}" f_labels.write(f"{label}\n") - fetch_and_save_sub_text(ids, [], label, paths['Text']) - fetch_and_save_sub_sol(ids, [], label, paths['Sol']) + fetch_and_save_sub_text(ids, [], label, paths['Text2']) + fetch_and_save_sub_sol(ids, [], label, paths['Sol2']) else: for item in indexes: suffix = format_indices(item['indices'], problem) label = f"Ex {current_ex_num}" + (f" : {suffix}" if suffix else "") f_labels.write(f"{label}\n") - fetch_and_save_sub_text(ids, item['indices'], label, paths['Text']) - fetch_and_save_sub_sol(ids, item['indices'], label, paths['Sol']) + fetch_and_save_sub_text(ids, item['indices'], label, paths['Text2']) + fetch_and_save_sub_sol(ids, item['indices'], label, paths['Sol2']) # Construct URL (append pb=true if \Roman matched) diff --git a/gemini_for_enonce.py b/gemini_for_enonce.py index 4064b21..ecb71a9 100644 --- a/gemini_for_enonce.py +++ b/gemini_for_enonce.py @@ -1,31 +1,70 @@ import shlex +import re import os import subprocess import sys import argparse from pathlib import Path from pydantic import BaseModel, Field -from typing import List +from typing import List, Union from google import genai from google.genai import types +from concurrent.futures import ThreadPoolExecutor +from 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 + return s1[:i] + -# Bug : l'output est limité à 8k token… # MODEL_ID = "gemini-3-flash-preview" MODEL_ID = "gemini-3.1-flash-lite" api_key = os.environ.get("GEMINI_API_KEY") -class QuestionItem(BaseModel): +# --- Modèles pour la Requête 1 --- +class QuestionOnlyItem(BaseModel): label: str = Field(description="The unique label of the question (e.g., '1.a', 'Exercice 1')") question_content: str = Field(description="The source text of the question, strictly extracted from the enonce file, EXCLUDING the label itself.") + +class ExamQuestions(BaseModel): + questions: List[QuestionOnlyItem] + +# --- Modèles pour la Requête 2 --- +class SolutionOnlyItem(BaseModel): + label: str = Field(description="The exact unique label of the question provided in the input.") solution_content: str = Field(description="The source text of the solution, strictly extracted from the correction file.") -class ExamExtraction(BaseModel): - questions: List[QuestionItem] +class ExamSolutions(BaseModel): + solutions: List[SolutionOnlyItem] -PROMPT = """I am providing: +# --- Modèles pour la Requête 3 --- +class ExtractedContext(BaseModel): + target_question_label: str = Field(description="The exact label of the FIRST question that comes immediately AFTER this information in the exam.") + context_content: str = Field(description="The source text of the definitions, notations, or hypotheses, extracted from the enonce.") + +class ExamContext(BaseModel): + contexts: List[ExtractedContext] + +# --- Modèle fusionné (pour le reste du script) --- +class QuestionItem(BaseModel): + label: str + question_content: str + solution_content: str + +class ContextItem(BaseModel): + content: str # Juste une string encapsulée pour le différencier facilement + +class ExamExtraction(BaseModel): + items: List[Union[QuestionItem, ContextItem]] # Liste mixte + +class GroupedExamExtraction(BaseModel): + groups: List[List[Union[QuestionItem, ContextItem]]] + +PROMPT_1 = """I am providing: 1. A PDF of an exam (`enonce.pdf`) 2. The source code of the exam questions (`enonce` file) -3. The source code of the exam solutions (`correction` file) Your task: 1. Identify all distinct question labels using the PDF document. @@ -34,9 +73,26 @@ Your task: from the `enonce` source file. Do not include the label itself in this extracted text (nor LaTeX like `item` nor org-mode list labelling like `2.`). - 3. For each label, extract its exact corresponding solution textual - content from the `correction` source file. Return the result as - a JSON list in the exact reading order of the document. + Return the result as a JSON list in the exact reading order of the document. +""" + +PROMPT_2 = """I am providing: +1. A JSON list of question labels and their texts extracted from an exam. +2. The source code of the exam solutions (`correction` file). + +Your task: + For each question label provided in the JSON, extract its exact corresponding solution textual + content from the `correction` source file. Return the result as a JSON list in the exact same order. +""" + +PROMPT_3 = """I am providing: +1. A JSON list of question labels and their texts extracted from an exam. +2. The source code of the exam questions (`enonce` file). + +Your task: + Extract important information necessary to understand the questions (e.g., definitions of objects, global notations, hypotheses, context) that are NOT part of the question texts themselves. + For each extracted piece of information, identify the label of the FIRST question that comes immediately AFTER this information in the exam. + Return the result as a JSON list. """ def find_file(folder: Path, base_name: str) -> Path: @@ -70,126 +126,445 @@ def process_exam(folder_path: str): client = genai.Client(api_key=api_key) - contents = [ + # ========================================== + # REQUÊTE 1 : Extraction des Énoncés + # ========================================== + contents_1 = [ types.Content( role="user", parts=[ - types.Part.from_text(text=PROMPT), + 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(), + ) + + cache_q_file = folder / "gemini_questions.json" + + if cache_q_file.is_file(): + print("Loading cached questions from 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...") + cache_q_file.write_text(response_q_text, encoding="utf-8") + + 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"--- EXTRACTED QUESTIONS ---\n{extracted_questions_json}"), types.Part.from_text(text=f"--- CORRECTION SOURCE ({correction_path.name}) ---\n{correction_text}"), ], ) ] - config = types.GenerateContentConfig( + config_2 = types.GenerateContentConfig( temperature=0.1, response_mime_type="application/json", - response_json_schema=ExamExtraction.model_json_schema(), + response_json_schema=ExamSolutions.model_json_schema(), ) - cache_file = folder / "gemini_response.json" + cache_s_file = folder / "gemini_solutions.json" - if cache_file.is_file(): - print("Loading cached response from gemini_response.json...") - response_text = cache_file.read_text(encoding="utf-8") + if cache_s_file.is_file(): + print("Loading cached solutions from gemini_solutions.json...") + response_s_text = cache_s_file.read_text(encoding="utf-8") else: - print("Sending request to Gemini...") - response = client.models.generate_content( + print("Sending request 2 (Solutions) to Gemini...") + response_s = client.models.generate_content( model=MODEL_ID, - contents=contents, - config=config + contents=contents_2, + config=config_2 ) - response_text = response.text + response_s_text = response_s.text + print("Saving solutions to cache...") + cache_s_file.write_text(response_s_text, encoding="utf-8") - print("Saving response to cache...") - cache_file.write_text(response_text, encoding="utf-8") + solutions_data = ExamSolutions.model_validate_json(response_s_text) - # Validate from the text variable (cached or fresh) - extracted_data = ExamExtraction.model_validate_json(response_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"--- EXTRACTED QUESTIONS ---\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(), + ) + + cache_c_file = folder / "gemini_context.json" + + if cache_c_file.is_file(): + print("Loading cached context from 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...") + cache_c_file.write_text(response_c_text, encoding="utf-8") + + 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} + + # 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] += "\n\n" + c.context_content + else: + ctx_map[c.target_question_label] = c.context_content + + merged_items = [] + for q in questions_data.questions: + # 1. S'il y a un contexte pour cette question, on l'insère d'abord dans la liste + if q.label in ctx_map: + merged_items.append(ContextItem(content=ctx_map[q.label])) + + # 2. Puis on ajoute la question + 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 = folder / "exam_items.txt" + full_items_file = folder / "exam_items_full.txt" + trunc_map = {} + + # --- INITIAL GROUPING COMPUTATION --- + questions_only = [item for item in extracted_data.items if isinstance(item, QuestionItem)] + + q_group_indices = [] + if questions_only: + current_g = [0] + for i in range(1, len(questions_only)): + p = get_lcp(questions_only[current_g[0]].label, questions_only[i].label) + proposed = current_g + [i] + valid = True + for k in range(len(proposed) - 1): + if get_lcp(questions_only[proposed[k]].label, questions_only[proposed[k+1]].label) != p: + valid = False + break + if valid: + current_g.append(i) + else: + q_group_indices.append(current_g) + current_g = [i] + q_group_indices.append(current_g) + + group_starter_labels = {questions_only[g[0]].label for g in q_group_indices[1:]} if q_group_indices else set() + + initial_groups = [] + current_group = [] + + for i, item in enumerate(extracted_data.items): + is_new_group = False + if isinstance(item, QuestionItem): + if item.label in group_starter_labels: + if not (len(current_group) > 0 and isinstance(current_group[-1], ContextItem)): + is_new_group = True + elif isinstance(item, ContextItem): + if i + 1 < len(extracted_data.items): + next_item = extracted_data.items[i+1] + if isinstance(next_item, QuestionItem) and next_item.label in group_starter_labels: + is_new_group = True + + if is_new_group and current_group: + initial_groups.append(current_group) + current_group = [] + + current_group.append(item) + + if current_group: + initial_groups.append(current_group) + + # ---- Transform labels, and check uniqueness + + seen_labels = set() + for group in initial_groups: + for item in group: + if isinstance(item, QuestionItem): + # 1. Transform label + item.label = item.label.replace("Exercice", "Ex") + item.label = item.label.replace(".", ")") + + # 2. Ensure uniqueness (prefix with XX) + while item.label in seen_labels: + item.label = f"XX{item.label}" + seen_labels.add(item.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...") + editor = os.environ.get("EDITOR") + try: + if editor: + subprocess.run(shlex.split(editor) + [str(items_file)]) + else: + if sys.platform.startswith("linux"): + subprocess.run(["xdg-open", str(items_file)]) + elif sys.platform == "darwin": + subprocess.run(["open", str(items_file)]) + else: + os.startfile(str(items_file)) + input("Press ENTER here once you have saved and closed the text file...") + except Exception as e: + print(f"Error running editor: {e}") + + 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 + + # 2. Actual Parsing + grouped_items = [] + current_group = [] + labels_list = [] + orig_idx = 0 + + 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 new_label != "CONTEXT": + labels_list.append(new_label) + + 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 orig_idx < len(extracted_data.items): + orig_item = extracted_data.items[orig_idx] + if isinstance(orig_item, QuestionItem): + current_group.append(QuestionItem( + label=new_label, + question_content=edited_content, + solution_content=orig_item.solution_content + )) + elif isinstance(orig_item, ContextItem): + current_group.append(ContextItem(content=edited_content)) + orig_idx += 1 + + if current_group: + grouped_items.append(current_group) + + # If we reached here without 'continue', the data is valid + break + + # Save labels and proceed + with open(folder / "labels", 'w', encoding='utf-8') as f_labels: + for label in labels_list: + f_labels.write(f"{label}\n") + + grouped_extraction = GroupedExamExtraction(groups=grouped_items) # 2. Setup output directories text_dir = folder / "Text" sol_dir = folder / "Sol" - text_dir.mkdir(exist_ok=True) - sol_dir.mkdir(exist_ok=True) + text2_dir = folder / "Text2" + sol2_dir = folder / "Sol2" + dirs = [text_dir, sol_dir, text2_dir, sol2_dir] - labels_file = folder / "labels" + 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() - # Step 1: Write initial labels - print("Writing initial labels file...") - with open(labels_file, "w", encoding="utf-8") as flabels: - for q in extracted_data.questions: - flabels.write(f"{q.label}\n") + if answer not in ("y", "yes"): + print("Aborted.") + sys.exit(1) + # 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) - # Step 2: Open labels file for user editing - print("Opening labels file for editing...") - editor = os.environ.get("EDITOR") - try: - if editor: - subprocess.run(shlex.split(editor) + [str(labels_file)]) - else: - # Fallbacks if $EDITOR is not set - if sys.platform.startswith("linux"): - subprocess.Popen(["xdg-open", str(labels_file)]) - elif sys.platform == "darwin": - subprocess.Popen(["open", str(labels_file)]) - else: - os.startfile(str(labels_file)) + text_dir.mkdir(exist_ok=True) + sol_dir.mkdir(exist_ok=True) - # xdg-open/open usually do not block, so we wait for user confirmation - input("Press ENTER here once you have saved and closed the labels file...") - except Exception: - print("Error running editor, using labels as given.") - # Step 3 & 4: Read the edited file back and create a mapping - with open(labels_file, "r", encoding="utf-8") as flabels: - edited_lines = [line.strip() for line in flabels if line.strip()] + print("Writing grouped question and solution files...") - mapping = [] - final_labels = [] - orig_idx = 0 + 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] - for line in edited_lines: - if line.startswith("+"): - new_label = line[1:].lstrip() - final_labels.append(new_label) - # New label, no source content - mapping.append((new_label, None)) - else: - new_label = line - final_labels.append(new_label) - # Map to initial order, advancing index only for non-'+' items - q_item = extracted_data.questions[orig_idx] if orig_idx < len(extracted_data.questions) else None - mapping.append((new_label, q_item)) - orig_idx += 1 + if not labels: + continue # Skip if a group has no questions (only contexts) - # Rewrite the labels file cleanly (removing '+' prefixes) - with open(labels_file, "w", encoding="utf-8") as flabels: - for lbl in final_labels: - flabels.write(f"{lbl}\n") + # 1. Compute the common prefix for the group + prefix = labels[0] + for lbl in labels[1:]: + prefix = get_lcp(prefix, lbl) - # Step 5: Write the final question and solution files - print("Writing question and solution files...") - for new_label, q_item in mapping: - safe_label = new_label.replace("/", "_") + # 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("/", "_") - if q_item: - q_content = q_item.question_content.replace("\\n", "\n") - s_content = q_item.solution_content.replace("\\n", "\n") - else: - q_content = "" - s_content = "" + text_content_lines = [] - # Write Text/label - with open(text_dir / safe_label, "w", encoding="utf-8") as f: - f.write(f"{new_label}\n{q_content}") + # 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) - # Write Sol/label - with open(sol_dir / safe_label, "w", encoding="utf-8") as f: - f.write(f"{new_label}\n{s_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}") + 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(f"CONTEXT :") + text_content_lines.append(tabulated_ctx) + + # 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): + """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: + print(f"Error compiling {tex_file.name}: {e}") + + print(f"Compiling {len(all_tex_files)} files to PDF using 4 threads...") + with ThreadPoolExecutor(max_workers=4) as executor: + executor.map(compile_worker, all_tex_files) - print(f"Success! Processed {len(mapping)} labels.") if __name__ == "__main__": if not api_key: diff --git a/prompting.py b/prompting.py index 1c6beb4..a842cbe 100644 --- a/prompting.py +++ b/prompting.py @@ -1,5 +1,6 @@ from pathlib import Path import io +import utils main_prompt = """I'm giving you an image of several written answers to an exam. @@ -85,26 +86,14 @@ Here is a possible correct answer : You are asked to score the question or exercice labeled `<