diff --git a/copienator/commands/gemini_for_enonce.py b/copienator/commands/gemini_for_enonce.py index c4aa68a..04d63ec 100644 --- a/copienator/commands/gemini_for_enonce.py +++ b/copienator/commands/gemini_for_enonce.py @@ -1,29 +1,36 @@ 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 configuration as config -from copienator import utils 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.filesystem import staged_files +from copienator.latex_json import normalize_literal_newlines, repair_latex_json from copienator.platform import validate_windows_labels -from copienator.latex_json import repair_latex_json, normalize_literal_newlines -from copienator.filesystem import staged_directory +from copienator.rubrics import present_rubric_labels +from copienator.statement_checkpoints import extract_stage, report_error from copienator.utils import compile_to_pdf @@ -130,7 +137,15 @@ def validate_rubric_content(content: str) -> str: 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): - raise ValueError("Rubric contains control characters, likely from unescaped LaTeX") + 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) --- @@ -224,8 +239,7 @@ def find_file(folder: Path, base_name: str) -> Path | None: return None -def generate_rubrics(client, group_context_text: str) -> dict[str, str]: - """Use the same rubric request for full and selective statement generation.""" +def _request_rubrics(client, group_context_text: str): response = client.models.generate_content( model=MODEL_ID, contents=[types.Content(role="user", parts=[ @@ -240,10 +254,71 @@ def generate_rubrics(client, group_context_text: str) -> dict[str, str]: response_json_schema=GroupRubrics.model_json_schema(), ), ) - rubrics = GroupRubrics.model_validate_json(repair_latex_json(response.text)).rubrics - if len({item.label for item in rubrics}) != len(rubrics): - raise ValueError("Gemini returned duplicate rubric labels") - return {item.label: validate_rubric_content(item.rubric_content) for item in rubrics} + 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: @@ -253,12 +328,19 @@ def validate_groups(groups: list[list[str]], labels: list[str]) -> None: raise CliError("Groups must contain every existing label exactly once") -def refine_existing(workspace: EvaluationWorkspace, mode: str, *, api_client=None) -> ExitCode: - """Regroup or replace rubrics without regenerating statements or solutions.""" +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("/", "_") @@ -304,21 +386,34 @@ def refine_existing(workspace: EvaluationWorkspace, mode: str, *, api_client=Non for line in workspace.label_groups_file.read_text(encoding="utf-8").splitlines() if line.strip()] validate_groups(groups, labels) - with staged_directory(workspace.root / "Persp") as staging: - for group in groups: - print(f"Generating rubric (Persp) for group: {', '.join(group)}...") - group_content = ( - "Contexte général de l’examen, fourni uniquement pour comprendre les questions :\n" - + context + "\n\nProduis des barèmes UNIQUEMENT pour les labels suivants : " - + ", ".join(group) + "\n\n" + "\n\n".join(questions[label] for label in group) - ) - rubrics = generate_rubrics(api_client, group_content) - if set(rubrics) != set(group) or any(not value.strip() for value in rubrics.values()): - raise CliError("Incomplete or unexpected Gemini rubrics; previous Persp preserved") + 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") - print("Replaced Persp with Gemini rubrics.") + 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 @@ -336,9 +431,15 @@ def process_exam( 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" + 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 @@ -392,38 +493,16 @@ def process_exam( 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(repair_latex_json(response_q_text)) + 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) - contents_2 = [ - types.Content( - role="user", - parts=[ - types.Part.from_text(text=PROMPT_2), - types.Part.from_text(text=f"--- QUESTIONS EXTRAITES ---\n{extracted_questions_json}"), - types.Part.from_text(text=f"--- CORRECTION SOURCE ({correction_path.name}) ---\n{correction_text}"), - ], - ) - ] - config_2 = types.GenerateContentConfig( automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), temperature=0.1, @@ -431,21 +510,19 @@ def process_exam( 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(repair_latex_json(response_s_text)) + 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.) @@ -468,21 +545,22 @@ def process_exam( 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) + 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, + )) - context_data = ExamContext.model_validate_json(repair_latex_json(response_c_text)) + 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 @@ -634,8 +712,7 @@ def process_exam( # --- 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: + 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) @@ -664,8 +741,15 @@ def process_exam( 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) @@ -704,6 +788,9 @@ def process_exam( 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 @@ -802,37 +889,40 @@ def process_exam( # 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) + 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 = [] @@ -855,9 +945,15 @@ def process_exam( group_context_text = "\n\n---\n\n".join(group_text_parts) - print(f"Generating rubric (Persp) for group: {', '.join(labels)}...") + missing_rubrics = [label for label in labels if label not in present] try: - rubrics_map = generate_rubrics(client, group_context_text) + 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)) @@ -887,21 +983,15 @@ def process_exam( 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}") + 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) --- - 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}") + 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() @@ -915,15 +1005,14 @@ def process_exam( 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 "" + context_sources.setdefault(ctxt_path, []) + if item.content not in context_sources[ctxt_path]: + context_sources[ctxt_path].append(item.content) - with open(ctxt_path, "a", encoding="utf-8") as f_c2: - f_c2.write(prefix + item.content) + save_missing(text_dir / safe_group_filename, '\n'.join(text_content_lines)) - # 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)) + 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...") # ========================================== @@ -934,9 +1023,16 @@ def process_exam( 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") + 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 @@ -947,19 +1043,8 @@ def process_exam( error for error in executor.map(compile_worker, all_tex_files) if error ] for error in compile_errors: - print(error) + report_error(folder, 'PDF compilation', error) processing_errors.extend(compile_errors) - atomic_write_text( - workspace.labels_file, - "".join(f"{label}\n" for label in labels_list), - ) - atomic_write_text( - workspace.label_groups_file, - "".join(", ".join(item.label for item in group if isinstance(item, QuestionItem)) + "\n" - for group in grouped_extraction.groups - if any(isinstance(item, QuestionItem) for item in group)), - ) - return ExitCode.PARTIAL if processing_errors else ExitCode.SUCCESS @@ -969,26 +1054,31 @@ def build_parser() -> argparse.ArgumentParser: actions.add_argument("--groups-only", action="store_true", help="Regroup existing questions with Gemini; update only label_groups") actions.add_argument("--persp-only", action="store_true", - help="Replace only Persp with Gemini rubrics for existing groups") + 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="Ignore cached Gemini extraction responses", + 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, - lambda args: refine_existing(workspace_from_args(args), - "groups" if args.groups_only else "persp") - if args.groups_only or args.persp_only else process_exam( - workspace_from_args(args), - restart=args.restart, - ), + run, ) diff --git a/copienator/rubrics.py b/copienator/rubrics.py new file mode 100644 index 0000000..379f18e --- /dev/null +++ b/copienator/rubrics.py @@ -0,0 +1,21 @@ +"""Shared inventory of usable, individual Persp files.""" +from pathlib import Path + + +def present_rubric_labels(root: Path, labels: list[str]) -> list[str]: + present = [] + for label in labels: + path = root / 'Persp' / label.replace('/', '_') + if not path.is_file(): + continue + try: + content = path.read_text(encoding='utf-8').strip() + except UnicodeError: + continue + lines = content.split('\n', 1) + body = lines[1].strip() if lines[0].strip() == label and len(lines) > 1 else content + if body and body != label and not any( + (ord(char) < 32 and char != '\n') or ord(char) == 127 for char in content + ): + present.append(label) + return present diff --git a/copienator/statement_checkpoints.py b/copienator/statement_checkpoints.py new file mode 100644 index 0000000..8d2d099 --- /dev/null +++ b/copienator/statement_checkpoints.py @@ -0,0 +1,154 @@ +"""Extraction checkpoints assume unchanged exam sources until an explicit restart.""" +from __future__ import annotations + +import json +import re +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +from copienator import CliError, atomic_write_json, atomic_write_text +from copienator.latex_json import repair_latex_json + + +def archive_response(root: Path, stage: str, raw: str | None) -> Path: + stamp = datetime.now(UTC).strftime('%Y%m%dT%H%M%S%fZ') + path = root / 'Cache' / 'statement_responses' / f'{stamp}-{stage}-{uuid4().hex}.json' + atomic_write_text(path, raw or '') + return path + + +def report_error(root: Path, stage: str, error, raw=None, response_path=None) -> None: + diagnostic = f'{stage}: {error}\n' + if raw is not None: + diagnostic += f'Raw response (escaped): {raw!r}\n' + if response_path: + diagnostic += f'Raw response saved to: {response_path}\n' + print(diagnostic, end='', flush=True) + try: + path = root / 'Cache' / 'statement_errors.log' + path.parent.mkdir(parents=True, exist_ok=True) + with path.open('a', encoding='utf-8') as log: + log.write(f'[{datetime.now(UTC).isoformat()}]\n{diagnostic}\n') + except OSError as log_error: + print(f'Could not save extraction error log: {log_error}', flush=True) + + +def validate_text(text: str) -> None: + if not text.strip(): + raise ValueError('Empty extracted content') + bad = [(i, ord(c)) for i, c in enumerate(text) + if (ord(c) < 32 and c not in '\n\t') or ord(c) == 127] + # Tabs can be legitimate indentation in code, but in math they often + # indicate an unknown misescaped TeX command beginning with t. + for match in re.finditer(r'(\$\$.*?\$\$|\$.*?\$|\\\(.*?\\\)|\\\[.*?\\\])', text, re.DOTALL): + bad.extend((match.start() + i, ord(c)) for i, c in enumerate(match.group()) if c == '\t') + if bad: + positions = ', '.join(f'U+{code:04X} at offset {i}' for i, code in bad) + raise ValueError(f'Control characters: {positions}; decoded content: {text!r}') + + +def extract_stage(root: Path, stage: str, item_model, content_field: str, request, + *, labels: list[str] | None = None) -> list: + """Keep valid items; request only pending known labels on the next run. + + Questions are re-requested as a complete ordered inventory because there + is no independent list of question labels before extraction. Contexts may + legitimately be empty; an empty questions/solutions response may not. + """ + checkpoint = root / 'Cache' / f'statement_{stage}.json' + state = json.loads(checkpoint.read_text()) if checkpoint.exists() else {} + accepted = state.get(stage, []) + inventory = state.get('inventory', [item['label'] for item in accepted] if stage == 'questions' else []) + if state.get('complete'): + return [item_model.model_validate(item) for item in accepted] + pending = state.get('pending') + if stage == 'solutions': + known = {item['label'] for item in accepted} + pending = [label for label in labels if label not in known] + legacy = root / 'Cache' / f'gemini_{"context" if stage == "contexts" else stage}.json' + raw = None + response_path = None + try: + if not checkpoint.exists() and legacy.exists(): + raw = legacy.read_text(encoding='utf-8') + print(f'Importing legacy {stage} response from {legacy}') + else: + print(f'Extracting {stage}' + (f' for {pending}' if pending else '') + '...') + raw = request(pending).text + response_path = archive_response(root, stage, raw) + payload = json.loads(repair_latex_json(raw)) + if not isinstance(payload, dict) or not isinstance(payload.get(stage), list): + raise TypeError(f'Response must contain a {stage} list') + except Exception as error: + # Mark legacy data as attempted too, so an invalid cache is never + # replayed indefinitely. Previously accepted items remain untouched. + atomic_write_json(checkpoint, {stage: accepted, 'complete': False, 'pending': pending, + 'inventory': inventory}) + report_error(root, stage, error, raw, response_path) + raise CliError(f'{stage} extraction failed; saved progress can be resumed') from error + + errors = [] + failed_targets = [] + items = [] + seen = set() + for index, entry in enumerate(payload[stage]): + if stage == 'questions' and isinstance(entry, dict): + label = entry.get('label') + if isinstance(label, str) and label.strip() and label not in inventory: + inventory.append(label) + try: + item = item_model.model_validate(entry) + validate_text(getattr(item, content_field)) + if stage == 'contexts': + first, last = item.target_question_label, item.last_question_label + if first not in labels or last not in labels or labels.index(first) > labels.index(last): + raise ValueError('Invalid context target or range') + if pending and first not in pending: + raise ValueError('Unexpected context target') + key = (first, last, item.context_content) + else: + if not item.label.strip(): + raise ValueError('Empty question label') + if labels is not None and item.label not in (pending or labels): + raise ValueError('Unexpected solution label') + key = item.label + if key in seen: + raise ValueError('Duplicate extracted item') + seen.add(key) + items.append(item.model_dump()) + except (ValueError, TypeError) as error: + errors.append(f'Item {index}: {error}; item (escaped): {entry!r}') + if isinstance(entry, dict) and entry.get('target_question_label') in (labels or []): + failed_targets.append(entry['target_question_label']) + + if stage == 'contexts': + # Preserve good contexts even when another context for the same target fails. + combined = accepted + [item for item in items if item not in accepted] + next_pending = list(dict.fromkeys(failed_targets)) if len(failed_targets) == len(errors) else None + absent = [label for label in (pending or []) if not any( + item['target_question_label'] == label for item in items)] + if absent: + errors.append(f'Missing context targets: {absent}') + next_pending = list(dict.fromkeys((next_pending or []) + absent)) + else: + previous = {item['label']: item for item in accepted} + previous.update({item['label']: item for item in items if item['label'] not in previous}) + order = [item['label'] for item in items] if stage == 'questions' else labels + combined = [previous[label] for label in order if label in previous] + combined += [item for label, item in previous.items() if label not in order] + missing = [label for label in (labels or []) if label not in previous] + if stage == 'questions': + missing = [label for label in inventory if label not in seen] + if not items: + errors.append('No valid questions returned') + if missing: + errors.append(f'Missing labels: {missing}') + next_pending = [label for label in (labels or []) if label not in previous] or None + complete = not errors + atomic_write_json(checkpoint, {stage: combined, 'complete': complete, 'pending': next_pending, + 'inventory': inventory}) + if errors: + report_error(root, stage, '\n'.join(errors), raw, response_path) + raise CliError(f'Incomplete {stage}; {len(combined)} valid items saved. Rerun to resume.') + return [item_model.model_validate(item) for item in combined] diff --git a/copienator_gui/app.py b/copienator_gui/app.py index 664a511..8c26bf8 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -21,6 +21,7 @@ from copienator.platform import ( validate_windows_labels, ) from copienator.utils import LATEX_WARNING_MARKER +from copienator.rubrics import present_rubric_labels from .diagnostics import collect_diagnostics from .batch_monitor import BatchMonitor @@ -821,6 +822,7 @@ class CopienatorApp(tk.Tk): "plotting", "manual_resolution", "giving_names", + "statement_persp", } self.console.configure(height=8 if expanded_form else 14) self.rowconfigure(1, weight=4 if expanded_form else 3) @@ -877,6 +879,10 @@ class CopienatorApp(tk.Tk): if spec.variants and variant.id not in spec.variants: continue value = values.get(spec.name, value_for_default(spec.default, evaluation_arg)) + if step.id == 'statement_persp' and spec.name == 'delete_persp': + value = False # Deletion is a one-run choice, never a saved default. + if step.id == 'statement' and spec.name == 'restart': + value = False choices = spec.choices if spec.name == "annotation_dir" and step.id in {"export", "import"}: choices, value = self._annotation_directory_choices(step.id) @@ -910,6 +916,16 @@ class CopienatorApp(tk.Tk): attach_tooltip(widget, spec.help) row += 1 + if step.id == 'statement_persp': + self.persp_inventory_label = ttk.Label(self.form, wraplength=680, justify='left') + self.persp_inventory_label.grid(row=row, column=0, columnspan=2, sticky='ew', pady=8) + row += 1 + ttk.Button(self.form, text='Actualiser les barèmes présents', + command=self._refresh_persp_inventory).grid(row=row, column=1, sticky='w') + row += 1 + self.arg_vars['target'].trace_add('write', lambda *_args: self._refresh_persp_inventory()) + self._refresh_persp_inventory() + if step.id == "manual_resolution": def manual_evaluation(): target = self.arg_vars.get("target") @@ -947,6 +963,23 @@ class CopienatorApp(tk.Tk): self._update_command_preview() self._update_controls() + def _refresh_persp_inventory(self) -> None: + if not self.current_step or self.current_step.id != 'statement_persp': + return + target = self.arg_vars.get('target') + value = str(target.get()).strip() if target else '' + root = (self.repository / value).resolve() if value else self.evaluation + try: + labels = EvaluationWorkspace(root).read_labels() if root else [] + present = present_rubric_labels(root, labels) if root else [] + missing = [label for label in labels if label not in set(present)] + text = (f'Barèmes présents : {len(present)}/{len(labels)} — Manquants : {len(missing)}\n' + + 'Présents : ' + (', '.join(present) or 'aucun') + '\n' + + 'Manquants : ' + (', '.join(missing) or 'aucun')) + except (OSError, ValueError) as error: + text = f'Inventaire Persp indisponible : {error}' + self.persp_inventory_label.configure(text=text) + def _render_context_controls(self, step: StepDefinition, row: int) -> int: if step.id == "annotation" and self.variant_var.get() == "grouped": controls = ttk.Frame(self.form) @@ -1589,6 +1622,12 @@ class CopienatorApp(tk.Tk): self._append_console(f"\n$ {command_display(command)}\n") try: self.runner.start(command, self.repository, environment, log_path) + if step.id == 'statement_persp': + self.arg_vars['delete_persp'].set(False) + self._save_current_form() + if step.id == 'statement' and 'restart' in self.arg_vars: + self.arg_vars['restart'].set(False) + self._save_current_form() if step.id == "correction" and variant.id in {"batch", "hybrid"}: self.state_store.update_step("correction", batch_started=True) except (OSError, RuntimeError) as exc: @@ -1922,6 +1961,8 @@ class CopienatorApp(tk.Tk): self.active_step_id = None self._populate_tree() self._update_controls() + if step_id == 'statement_persp': + self._refresh_persp_inventory() if step_id == "statement" and self._latex_warning_paths: self.after_idle(self._show_current_latex_warnings) if step_id == "personal_read": diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index 7ef2170..bd7296c 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -152,10 +152,10 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: arg_target("le dossier de l’évaluation"), ArgumentSpec( "restart", - "Ignorer le cache (--restart)", + "Recommencer l’analyse (--restart)", kind="bool", flag="--restart", - help="Ignore les résultats Gemini mis en cache et recommence entièrement l’analyse de l’énoncé.", + help="Recommence entièrement l’analyse. Archive les checkpoints, les modifications des groupes et les sorties précédentes. Sans cette option, reprend le travail incomplet en conservant les résultats déjà enregistrés.", variants=("gemini",), ), ), @@ -172,10 +172,16 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: optional=True, personal=True, requires=("labels", "Text2", "Sol2"), ), StepDefinition( - "statement_persp", "Prétraitement de l’énoncé", "Générer les barèmes par Gemini", - "Facultatif. Inutile après l'analyse de l'énoncé par Gemini, puisque les barèmes ont déjà été générés. ", + "statement_persp", "Prétraitement de l’énoncé", "Générer les barèmes manquants avec Gemini", + "Génère uniquement les barèmes manquants dans Persp et conserve ceux déjà présents. " + "Les fichiers vides ou contenant seulement le label sont considérés comme manquants. " + "Après un échec, relancez cette étape pour compléter les barèmes. " + "Cochez Delete current Persp pour supprimer les barèmes actuels au lancement et tous les régénérer.", (python("default", "Barèmes Gemini", "statement", fixed_args=("--persp-only",)),), - arguments=(arg_target("le dossier de l’évaluation"),), + arguments=(arg_target("le dossier de l’évaluation"), + ArgumentSpec('delete_persp', 'Delete current Persp', kind='bool', + flag='--delete-persp', default=False, + help='Supprime tous les fichiers actuels de Persp au lancement, y compris les barèmes modifiés manuellement. Puis régénère tous les barèmes.')), optional=True, personal=True, requires=("labels", "label_groups", "Text2", "Sol2"), ), StepDefinition( diff --git a/docs/guide_utilisateur_gui.md b/docs/guide_utilisateur_gui.md new file mode 100644 index 0000000..a1f7067 --- /dev/null +++ b/docs/guide_utilisateur_gui.md @@ -0,0 +1,421 @@ +# Copienator — guide de l’interface graphique + +Ce guide suit les intitulés et l’ordre des étapes de l’interface. Un **label** est l’identifiant d’une question ; une **copie** correspond au travail d’un élève. + +## Prendre en main l’interface + +- Choisissez le dossier **Évaluation** avec **Parcourir…**, puis **Charger**. +- Suivez la liste de gauche. Avant **Exécuter**, vérifiez le **Mode**, la **Cible** et les prérequis signalés. La cible est normalement toute l’évaluation ; choisir un fichier permet, lorsque proposé, de ne traiter qu’une copie. Les aides **ⓘ** précisent les options. +- N’activez une option d’écrasement que pour remplacer volontairement les résultats précédents. + +## Préparation et énoncé + +### Vérifier les fichiers d’entrée + +**Effet :** vérifie la présence des documents nécessaires à l’évaluation. + +Placez les scans des copies, `enonce.pdf`, `enonce.tex`, `correction.tex` et la liste d’élèves `names` dans le dossier d’évaluation (`names` peut aussi être à la racine du projet). + +Pour un dossier nommé `A`, les noms `A.pdf`, `A.tex`, `Ac.tex` sont également acceptés. Vérifiez que qu’aucun PDF étranger ne sera pris pour une copie. + +Après un ajout, cliquez sur **Recharger — vérifier à nouveau**. + +### Analyser l’énoncé + +**Effet :** extrait les labels des questions, les contenus des questions, contextes et solutions de vos fichiers, puis prépare des regroupement initiaux de questions, ainsi que des barèmes. + +Choisissez **Analyse avec Gemini**. + +Éventuellement traitez les alertes **Problèmes LaTeX…** avant de poursuivre. + +**Ignorer le cache** recommence l’analyse Gemini : à réserver à une +analyse devenue incorrecte ou obsolète. + +### Relire les barèmes — facultatif + +**Effet :** permet de vérifier et d’ajuster les instructions qui guideront la correction. + +Ouvrez les barèmes dans `Persp`. Vérifiez les réponses acceptables, les raisonnements alternatifs et les exigences de justification ; corrigez les consignes trop sévères, ambiguës ou erronées. Enregistrez vos modifications avant de marquer l’étape terminée. + +## Prétraitement des copies + +### Retourner toutes les copies — facultatif + +**Effet :** applique une rotation de 180° à tous les scans de copies du dossier. + +Exécutez seulement si toutes les copies sont à l’envers. + +### Renommer les PDF en CopieXX.pdf + +**Effet :** attribue un numéro aux copies selon l’ordre de leurs noms de fichiers. + +Vérifiez qu’un PDF correspond à une copie et que le dossier ne contient pas de documents parasites. Exécutez puis contrôlez le nombre de copies. + +### Séparer et réordonner les pages + +**Effet :** Découpe les scans A3 en pages A4 correctement ordonnées. + +Examinez chaque page : + - déplacer la coupure si jamais l'élève a dépassé d'une moitié de sa + copie A3 sur l'autre. + - Utiliser les raccourcis donnés pour retirer les pages vides. + +Suivez les raccourcis affichés dans la fenêtre, qui peuvent varier +selon votre profil. Par défaut : + +| Action | Touche | +|---------------------------------------------------------------------|-----------------| +| Déplacer la séparation | ← / → | +| Garder les deux moitiés et avancer | `s` | +| Garder seulement la gauche / la droite | `q` / `d` | +| Garder la page entière, pour une page déjà A4 | `f` | +| Écarter une page réellement vide | `z` | +| Retourner la page / toutes les pages du fichier / tous les fichiers | `r` / `R` / `,` | +| Envoyer la page en fin / inverser l’ordre des pages | `a` / `i` | +| Recommencer le fichier / revenir au fichier précédent | `T` / `P` | +| Ouvrir l’outil de réorganisation, s’il est disponible | `A` | + +**Copies signalées uniquement** limite le traitement aux copies +précédemment signalées (par une étape suivante). + +### Rogner les zones vides — facultatif + +**Effet :** retire les marges vides en haut et en bas des pages, avec +sauvegarde des versions précédentes. + +Ce n'est pas particulièrement utile, et il pourrait y avoir des faux positifs. + +### Découper une partie à gauche pour détection des labels + +**Effet :** Prépare des images qui seront utilisées pour repérés les numéros de questions. + +Éventuellement, cocher `Toujours utiliser la page entière`. Pour rendre +cette étape non interactive, cocher aussi `et ne pas vérifier visuellement`. +Ce mode est plus cher : sans avoir vérifié, Astra estime que le coût +passerait de 0,50€ à 2€. + +Sans quoi, pour chaque copie, il faut cadrer la bande pour qu'elle +contienne bien tous les labels : + +**←/→** la déplacent, **↑/↓** l’élargissent ou la réduisent ; **Entrée** valide. + +Si les labels sont dispersés, utilisez **1** pour utiliser les pages entières pour cette copie. + +Cette vérification visuelle permet toutefois de repérer des copies à +l'envers, ou dont les pages sont dans le mauvais ordre. Utiliser **!** +pour signaler une copie problématique. + +Vous pourrez alors relancer l'étape de réordonnement des copies avec +les copies signalées. + +## Labels et regroupement + +### Détecter les labels avec Gemini + +**Effet :** reconnaît les identifiants de questions et leur position dans les copies. + +### Vérifier visuellement les labels + +**Effet :** valide ou corrige les repères qui serviront à découper les réponses. + +Vérifiez visuellement les labels. + - Quand un label est manquant, le suivant est affiché en jaune. + - Quand un label n'est pas dans le bon ordre, il est affiché en rouge. + +Pour faire des modifications : **e** ouvre le fichier .json à modifier à la main + - Vous pouvez retirer un label, ou le modifier. + - Pour ajouter un label manqué : + - cliquer sur l'image là où est le label, ce qui va copie les coordonnées correspondantes + - puis les coller dans le fichier .json, et renseignez le label. + +**Entrée** pour valider, **i** pour ouvrir l'énoncer, **o** pour +ouvrir la copie d'origine, **p** pour revenir en arrière. + +### Découper les réponses par question + +**Effet :** crée un document de réponse pour chaque question repérée +dans chaque copie. + +### Rogner le bas des réponses — facultatif + +**Effet :** enlève les grands espaces vides sous les réponses découpées, avec sauvegarde des originaux. + +### Regrouper les réponses + +**Effet :** regroupe les réponses par label pour préparer la correction. + +## Correction + +### Lancer ou intégrer la correction + +**Effet :** obtient les corrections automatiques ou intègre des résultats préparés en différé. + +Il faut choisir un des modes suivants. + +Initialement, vous voulez + - soit le mode **Correction immédiate**, qui + va faire les requêtes de corrections en live, ce qui peut prendre de + l'ordre de 30min ou 1h. (Si elle est interrompue, il suffit de la + relancer pour qu'elle reprenne d'où elle en était). + - soit le mode **Préparer toutes les requêtes batch** : au lieu de + faire des requêtes en live, les requêtes sont batchées et toutes + envoyée à Gemini, qui les traitera en théorie au pire sous 24h, en + pratique dans l'heure. L'avantage est que le coût est divisé par + deux. + +Le mode **Batch à partir d'un label** est un mode hybride inutile, où +certaines requêtes sont faites en live, et d'autres sont batch. + +| Mode | À faire et à vérifier | +|----------------------------------------|---------------------------------------------------------------------------------------------------------------------------| +| **Correction immédiate** | Lancez et surveillez le nombre de groupes traités. | +| **Préparer toutes les requêtes batch** | Préparez les demandes, puis passez à **Envoyer les batchs** : la préparation seule ne les envoie pas. | +| **Batch à partir d’un label** | Indiquez le premier label différé ; les précédents sont corrigés immédiatement. Suivez ensuite le circuit batch. | +| **Recorrection depuis refaire.json** | Relancez les questions sélectionnées, notamment après résolution des conflits. Vérifiez la sélection avant de lancer. | +| **Intégrer les résultats batch** | Utilisez après leur récupération ; attendez la fin de l’intégration avant de générer les annotations. | +| **Réinitialiser les corrections** | Réservez ce mode à un recommencement volontaire ; lisez la confirmation et sauvegardez les corrections utiles auparavant. | + +Laissez **Écraser les corrections existantes** décoché pour reprendre +sans tout recalculer. Une **Limite d’appels Pro** peut limiter le +travail de cette exécution : vérifiez ce qui reste à traiter. +**Interrompre** bloque les nouveaux appels, attend ceux en cours et +sauvegarde leurs résultats ; relancez pour terminer les validations +restantes. Examinez les erreurs et les conflits proposés, puis relisez +les résultats sur les copies annotées. + +### Envoyer les batchs — si correction différée + +**Effet :** transmet à Gemini les demandes de correction préparées. + +Exécutez après la préparation batch et vérifiez que les envois +aboutissent. En cas d’envoi partiel, reprenez cette étape et consultez +l’état des jobs avant de poursuivre. + +### Consulter l’état des batchs — si correction différée + +**Effet :** vérifie si tous les traitements différés ont réussi et sont récupérables. + +Cliquez sur **Exécuter**, ou sur **Vérifier toutes les 5 minutes** en +gardant Copienator ouvert. Attendez que tous les résultats soient +prêts ; une notification l’indique. Un job en attente ou en échec ne +vaut pas une correction terminée : lisez son état au lieu de forcer le +passage à la suite. + +### Récupérer les résultats batch — si correction différée + +**Effet :** télécharge et rassemble les réponses des traitements terminés. + +Exécutez lorsque tous les résultats sont disponibles. Vérifiez +l’absence de récupération manquante ; l’interface revient ensuite à +**Lancer ou intégrer la correction**, en mode **Intégrer les résultats +batch**. Exécutez cette intégration. + +### Résoudre les conflits manuels — si des conflits sont signalés + +**Effet :** corrige les attributions ambiguës de réponses avant leur recorrection. + +Lors de la correction, certains labels seront inévitablement érronés +(l'étudiant a mis le mauvais label, ou n'a pas mis de label, ou à +traiter une question à droite d'une autre). + +L'étape de correction détecte ces problèmes (enfin, certains…), et +essaie de les corriger automatiquement. Les cas non automatiques +(«conflits») sont ceux où elle détecte un contenu correspondant à un +label `B`, alors qu'il y a déjà une image pour le label `B` + +Cette étape permet de résoudre manuellement les conflits : on vous +donne une image initiale, avec un label d'origine `A` dont le contenu +est bizarre et correspond, éventuellement partiellement, à un label +`B` cible, qui a déjà une image. + +On vous demande de choisir ce qu'il doit advenir de l'image d'origine. + +Ouvrez les PDF source et cible proposés et comparez-les à l’énoncé. + +Cliquez sur **Ouvrir dans un éditeur de texte** pour ajuster les +instructions. Leur forme est `Copie01 Label source OP Label cible|`. +Choisissez l’action d’après le contenu réel : + +| Opérateur | Résultat | +|-------------|---------------------------------------------------------------------------------| +| `->` / `x>` | Ajouter la source à la cible ; conserver / archiver la source pour le label `A` | +| `-x` / `xx` | Remplacer la cible par la source ; conserver / archiver la source. | +| `ss` | Garder les deux réponses séparées (= ne rien faire) | +| `sx` / `xs` | Garder la source / la cible et archiver l’autre. | + +Pour une fusion, `Cible|` met la cible avant la source, `|Cible` fait l’inverse. + +Une autre opération est **Cut** : dans les cas où vous voulez découper +l'image d'origine en deux. Utiliser le bouton **Cut** pour choisir la +coupure, et quelle moitié doit être gardée. L'interface graphique vous +fournira alors une instruction texte à copier et à mettre dans le +fichier texte. + +Enregistrez le fichier texte, éventuellement le **Recharger**, puis +**Exécuter**. + +Retirer une instruction signifie accepter les PDF sans changement ; +exécutez même si vous avez retiré toutes les lignes. + +Après succès, l’interface revient à l'étape de correction (avec le +mode **recorrection à partir de refaire.json** ) : lancez-la. + +### Nettoyer la correction + +**Effet :** Essaye de réparer des problèmes d'encodages courant dans les réponses de Gemini. + +## Annotations et tablette + +### Générer les copies annotées + +**Effet :** Génère des versions des copies, annotées avec la correction Gemini. + +Choisir le mode : + - **Annotations simples (Anot)** Annote chaque copie en des documents qu + - **Annotations avec cases (Bnot)** Annote chaque copie, avec des + cases qui vous permettrons, en les cochant, de supprimer certains + commentaires, ou de modifier la note + - **Annotations groupées (BGnot)** Génère des documents pdf, où les + réponses annotées sont regroupées par label (ou plutôt par groupes + de label). + + Les groupes de label sont ceux qui ont été générés dans l'analyse + de l'énoncé. Il est possible de les modifier à la main ici. + +**Écraser les sorties** remplace les documents existants. **Mode +refaire** ne concerne qu’une reprise sélectionnée ; utilisez de +préférence le parcours dédié décrit plus bas. + +### Exporter — facultatif + +**Effet :** place les documents générés dans le dossier de transfert configuré. + +Choisissez le même **Dossier d’annotations** qu’à la génération. + +### Annoter les PDF sur la tablette + +**Effet :** recueille votre validation pédagogique, vos changements de scores et vos annotations manuscrites. + +Relisez chaque réponse avec sa correction. Dans les modes avec cases, cochez une seule note par question pour la remplacer ; laissez les notes correctes inchangées. + +Une case près d’un commentaire le supprime, celle sur un rectangle +supprime ce rectangle ; **clr** supprime tous les commentaires et +rectangles de la question. Ajoutez vos propres remarques sans écrire +accidentellement dans les cases. Gardez les dimensions et l’ordre des +pages pour permettre la lecture du retour. + +Avec le circuit d’export/import, renvoyez les fichiers sous **leur nom +exporté** (groupe ou `CopieXX.pdf`), dans le dossier d’import prévu. + +### Importer les annotations manuscrites + +**Effet :** rattache les fichiers modifiés manuellement aux documents de l’évaluation. + +Vérifiez l’évaluation, le dossier d’import et le **Dossier +d’annotations**, puis exécutez. Écartez les anciens retours et ceux +d’autres évaluations pour éviter une mauvaise association. + +## Finalisation + +### Lire les annotations manuscrites + +**Effet :** applique les cases cochées et les ajouts manuscrits pour reconstruire les copies et leurs scores. + +Choisissez **Lecture Bnot** ou **Lecture BGnot** conformément à la +génération. Pour le parcours `Anot`, passez directement à +l’attribution des noms. Après lecture, ouvrez les résultats et +vérifiez notes modifiées, suppressions de commentaires et remarques +manuscrites. + +Si les options `answers` sont présentes, choisissez les compléments +souhaités dans les réponses individuelles : contexte, énoncé, +correction. + +En mode groupé, le PDF final peut omettre une réponse parfaite sans +commentaire conservé ni annotation manuscrite significative ; son +score reste enregistré. Le PDF peut même être absent si aucune réponse +ne reste à y inclure. Contrôlez donc aussi les scores et l’image +complète, sans assimiler automatiquement une omission à une perte de +réponse. + +Le mode **Régénérer avec les nouveaux énoncés/corrigés et les +score.json** sert à mettre à jour les scores sur les images finales à +partir d'une modification manuelle des fichiers `score.json`. + + +### Attribuer les noms et préparer A Rendre + +**Effet :** rassemble les résultats de chaque élève dans le dossier `A Rendre`. + +Choisissez le dossier d’annotations utilisé et exécutez. Dans +**Vérification des noms**, traitez tous les noms `Unknown` et les +doublons : **Ouvrir l’originale**, saisir le nom, puis **Renommer**. + +**Mettre à jour uniquement les images de answers** ne rafraîchit que +les réponses individuelles des élèves existants. + +## Refaire une partie de l’évaluation + +Utilisez ce parcours après avoir terminé le passage principal. Conservez ses annotations : elles servent à préserver les questions non sélectionnées. Ne démarrez **Nouvelle reprise** qu’après avoir importé et fusionné la reprise courante ; **Refaire la même sélection** permet de repartir avec les mêmes copies et questions. + +### Choisir les copies et les questions + +**Effet :** enregistre les réponses à reprendre et leur mode de présentation. + +Choisissez une copie et une question, puis **+ Ajouter** ; répétez selon les besoins. **Toute la copie** sélectionne toutes ses questions ; **Toutes les copies** applique le choix à la classe. Vérifiez le tableau, retirez les erreurs, choisissez le **Passage principal** correct et la présentation **PDF à vérifier**, puis **Enregistrer la sélection**. + +### Reprendre le découpage — facultatif + +**Effet :** permet de corriger les repères des copies sélectionnées. + +Reprenez les contrôles de la vérification visuelle des labels. Validez les images modifiées ; chaque copie sélectionnée s’ouvre à son tour, et fermer sa fenêtre passe à la suivante. Ignorez si les découpages sont déjà corrects. + +### Redécouper les réponses — facultatif + +**Effet :** reconstruit les réponses des copies sélectionnées à partir des repères corrigés. + +Exécutez après toute modification du découpage. Ouvrez les réponses concernées et vérifiez leurs limites ; en présence de résolutions manuelles, contrôlez particulièrement les versions `_new` et `_old`. Ignorez si le découpage n’a pas changé. + +### Refaire la correction — facultatif + +**Effet :** relance la correction automatique des seules questions sélectionnées. + +Vérifiez la sélection puis exécutez. Contrôlez les erreurs et les résultats proposés. Ignorez si vous souhaitez uniquement modifier les notes ou commentaires à la main lors de la relecture. + +### Préparer les copies à vérifier + +**Effet :** génère dans `BRnot` les documents avec cases du passage de reprise. + +Exécutez puis vérifiez que seules les réponses voulues sont présentes, avec la présentation choisie. Cette génération remplace les documents de reprise concernés : ne la relancez pas sur une relecture manuscrite à conserver. + +### Exporter vers la tablette — facultatif + +**Effet :** transfère les documents du passage de reprise vers le dossier d’export. + +Écartez les anciens exports, exécutez et contrôlez les documents transférés. Conservez leur nom complet, y compris le préfixe de reprise s’il est présent : il permet d’associer les retours au bon passage. Ignorez si vous relisez directement les fichiers locaux. + +### Vérifier sur la tablette + +**Effet :** recueille la validation et les modifications des réponses reprises. + +Appliquez les mêmes règles de cases et d’écriture que pour le premier passage. Retournez tous les documents, même sans changement, sous leur nom exporté dans le dossier d’import ; retirez les anciens retours. En travail local, placez les `Concat_annotated.pdf` dans les sous-dossiers de cette reprise. + +### Importer les copies vérifiées — facultatif + +**Effet :** récupère les annotations du passage de reprise courant. + +Exécutez et vérifiez que tous les retours attendus sont acceptés. Les fichiers portant le préfixe d’une autre reprise sont ignorés. Ignorez cette étape seulement si les `Concat_annotated.pdf` sont déjà au bon endroit. + +### Mettre à jour les copies finales + +**Effet :** fusionne les réponses refaites avec les autres réponses conservées du passage principal. + +Vérifiez le **Passage principal**, puis exécutez. Contrôlez une question refaite et une question conservée pour chaque copie concernée ; résolvez tout retour manquant avant de poursuivre. Relancez ensuite **Attribuer les noms et préparer A Rendre** et vérifiez les livrables mis à jour avant de les remettre aux élèves. + +## Archivage + +### Nettoyer les fichiers intermédiaires — facultatif + +**Effet :** supprime définitivement les fichiers de travail et la progression de l’interface en conservant les livrables prévus. + +Attendez la fin des relectures et reprises. Vérifiez les résultats de tous les élèves dans `A Rendre` et sauvegardez l’évaluation si vous voulez pouvoir reprendre le travail. Choisissez d’abord **Prévisualiser sans supprimer**, puis examinez la liste. Ne lancez **Nettoyage définitif** qu’une fois certain de ne plus avoir besoin des intermédiaires : les PDF traités, textes de l’énoncé, corrections, journaux et livrables de `A Rendre` sont conservés, mais une reprise exigera de régénérer les données supprimées. diff --git a/docs/statement_recovery.md b/docs/statement_recovery.md new file mode 100644 index 0000000..cbce02b --- /dev/null +++ b/docs/statement_recovery.md @@ -0,0 +1,55 @@ +# Reprendre la génération Gemini de l’énoncé + +Relancez la même commande après une erreur ou une interruption : + +```sh +python -m copienator statement Interro05 +``` + +Sans `--restart`, les sources sont supposées inchangées. Les questions, +solutions et contextes validés sont conservés dans `Cache/statement_*.json`. +Une réponse lisible contenant un élément invalide conserve ses autres éléments +valides. Les solutions et contextes incomplets sont redemandés pour leurs labels +manquants. Une réponse JSON illisible reste archivée, mais ses éléments ne sont +pas récupérés automatiquement. + +Les questions sont redemandées sous forme d’une liste complète et ordonnée : +avant leur extraction, aucun inventaire indépendant de labels n’est disponible. +Les labels déjà observés doivent rester présents. Vérifiez la liste contre +`enonce.pdf` dans l’éditeur de groupes avant de l’approuver, notamment pour +repérer une question que Gemini aurait entièrement omise. + +Les modifications de `Tmp/exam_items.txt` survivent à une interruption de +l’éditeur. Après approbation, les groupes sont enregistrés dans +`Cache/statement_grouping.json` ; une reprise utilise ce checkpoint sans rouvrir +l’éditeur. Les fichiers générés déjà présents sont conservés, les fichiers +manquants sont complétés, et seuls les PDF manquants, échoués ou plus anciens +que leur source générée sont recompilés. Les contextes ne sont pas ajoutés une +seconde fois lors d’une reprise. + +Les barèmes `Persp` valides sont conservés, y compris ceux des autres questions +d’une réponse dont un barème est invalide. Les labels et groupes sont écrits +avant la génération des barèmes et la compilation, ce qui permet aussi une +reprise par `statement Interro05 --persp-only`. + +Toutes les réponses d’extraction sont archivées avant décodage dans +`Cache/statement_responses/`. Les erreurs, avec le texte échappé et le chemin +de la réponse, sont affichées et ajoutées à `Cache/statement_errors.log`. +Les barèmes utilisent `Cache/rubric_responses/` et `Cache/rubric_errors.log`. +Les anciens caches `gemini_questions.json`, `gemini_solutions.json` et +`gemini_context.json` sont importés une fois : une réponse invalide n’est pas +rechargée indéfiniment. + +Pour recommencer explicitement depuis les sources, notamment après les avoir +modifiées : + +```sh +python -m copienator statement Interro05 --restart +``` + +Les checkpoints et modifications des groupes précédents sont déplacés dans +`Cache/statement_history/`. Les anciennes sorties y sont également déplacées +quand la nouvelle extraction et les groupes sont prêts, avant leur remplacement. +Une erreur d’extraction conserve donc les anciennes sorties. Les réponses brutes +restent disponibles. Si ce nouveau départ échoue, relancez sans `--restart` +pour reprendre sa progression. diff --git a/tests/test_statement_personal_flow.py b/tests/test_statement_personal_flow.py index a17836a..31ce5ad 100644 --- a/tests/test_statement_personal_flow.py +++ b/tests/test_statement_personal_flow.py @@ -74,7 +74,8 @@ class SelectiveGeminiTests(unittest.TestCase): (self.root / folder).mkdir() for label in self.labels: suffix = ".tex" if folder.endswith("2") else "" - (self.root / folder / (label + suffix)).write_text(f"Personal {folder}: {label}") + (self.root / folder / (label + suffix)).write_text( + label + '\n' if folder == 'Persp' else f"Personal {folder}: {label}") (self.root / "Persp" / "Ex 1").write_text("Old aggregate rubric") self.client = SimpleNamespace(models=SimpleNamespace(generate_content=Mock())) @@ -113,8 +114,11 @@ class SelectiveGeminiTests(unittest.TestCase): self.assertEqual(gemini.refine_existing(self.workspace, "persp", api_client=self.client), ExitCode.SUCCESS) after = self.snapshot() self.assertEqual({key: value for key, value in before.items() if not key.startswith("Persp/")}, - {key: value for key, value in after.items() if not key.startswith("Persp/")}) - self.assertFalse((self.root / "Persp" / "Ex 1").exists()) + {key: value for key, value in after.items() + if not key.startswith(("Persp/", "Cache/"))}) + responses = list((self.root / 'Cache' / 'rubric_responses').glob('*.json')) + self.assertEqual(len(responses), 2) + self.assertEqual((self.root / "Persp" / "Ex 1").read_bytes(), before['Persp/Ex 1']) for label in self.labels: self.assertIn("Barème Gemini", (self.root / "Persp" / label).read_text()) for call in self.client.models.generate_content.call_args_list: @@ -126,13 +130,22 @@ class SelectiveGeminiTests(unittest.TestCase): self.assertIn("caractères mathématiques Unicode", gemini.PROMPT_4) self.assertIn(r"$\lfloor \sqrt{k} \rfloor$", gemini.PROMPT_4) - def test_incomplete_or_failed_rubrics_preserve_entire_persp(self): + def test_incomplete_or_failed_rubrics_keep_completed_groups(self): before = self.snapshot() for last in (self.response({"rubrics": []}), RuntimeError("API unavailable")): + for label in self.labels[:2]: + (self.root / 'Persp' / label).write_bytes(before[f'Persp/{label}']) self.client.models.generate_content.side_effect = [self.rubric_response(self.labels[:2]), last] with self.assertRaises((CliError, RuntimeError)): gemini.refine_existing(self.workspace, "persp", api_client=self.client) - self.assertEqual(self.snapshot(), before) + after = self.snapshot() + for label in self.labels[:2]: + self.assertIn(b'Bar\xc3\xa8me Gemini', after[f'Persp/{label}']) + self.assertEqual(after['Persp/Ex 2'], before['Persp/Ex 2']) + self.assertEqual(after['Persp/Ex 1'], before['Persp/Ex 1']) + self.assertEqual({key: value for key, value in before.items() if not key.startswith('Persp/')}, + {key: value for key, value in after.items() + if not key.startswith(('Persp/', 'Cache/'))}) def test_rubric_latex_and_line_breaks_survive_json_and_file_writes(self): expected = ( @@ -161,15 +174,96 @@ class SelectiveGeminiTests(unittest.TestCase): {'Ex 2': '$' + command + '$'}) self.assertEqual(self.client.models.generate_content.call_count, 1) - def test_unrepairable_control_character_preserves_all_existing_rubrics(self): + def test_control_character_keeps_progress_and_saves_diagnostics(self): before = self.snapshot() broken = self.response({'rubrics': [{'label': 'Ex 2', 'rubric_content': 'invalid \x01 control'}]}) self.client.models.generate_content.side_effect = [ self.rubric_response(self.labels[:2]), broken] - with self.assertRaisesRegex(ValueError, 'control characters'): + output = io.StringIO() + with patch('sys.stdout', output), self.assertRaisesRegex(ValueError, 'control characters'): gemini.refine_existing(self.workspace, 'persp', api_client=self.client) + after = self.snapshot() + self.assertEqual(after['Persp/Ex 2'], before['Persp/Ex 2']) + for label in self.labels[:2]: + self.assertIn(b'Bar\xc3\xa8me Gemini', after[f'Persp/{label}']) + self.assertIn('Question Ex 2', output.getvalue()) + self.assertIn('U+0001 at offset 8', output.getvalue()) + self.assertIn(r'invalid \x01 control', output.getvalue()) + responses = list((self.root / 'Cache' / 'rubric_responses').glob('*.json')) + self.assertEqual(len(responses), 2) + self.assertIn(broken.text, [path.read_text() for path in responses]) + log = (self.root / 'Cache' / 'rubric_errors.log').read_text() + self.assertIn('Question Ex 2', log) + self.assertIn(r'invalid \x01 control', log) + + def test_invalid_json_is_saved_exactly_and_reported(self): + raw = '{"rubrics": [this is malformed JSON' + self.client.models.generate_content.return_value = SimpleNamespace(text=raw) + output = io.StringIO() + with patch('sys.stdout', output), self.assertRaises(ValueError): + gemini.generate_rubrics(self.client, 'context', workspace=self.workspace, + expected_labels=['Ex 2']) + responses = list((self.root / 'Cache' / 'rubric_responses').glob('*.json')) + self.assertEqual(len(responses), 1) + self.assertEqual(responses[0].read_text(), raw) + self.assertIn(raw, output.getvalue()) + self.assertIn(str(responses[0]), output.getvalue()) + + def test_missing_question_only_preserves_existing_and_keeps_group_context(self): + kept = self.root / 'Persp' / self.labels[0] + kept.write_text(self.labels[0] + '\nManually reviewed rubric') + before = kept.read_bytes() + (self.root / 'Persp' / self.labels[1]).write_text('') + (self.root / 'Persp' / self.labels[2]).unlink() + self.client.models.generate_content.side_effect = [ + self.rubric_response([self.labels[1]]), self.rubric_response([self.labels[2]])] + gemini.refine_existing(self.workspace, 'persp', api_client=self.client) + self.assertEqual(kept.read_bytes(), before) + call = self.client.models.generate_content.call_args_list[0] + text = call.kwargs['contents'][0].parts[1].text + self.assertIn('UNIQUEMENT pour les labels suivants : ' + self.labels[1] + '\n', text) + self.assertIn('Question [' + self.labels[0] + ']', text) + self.assertEqual(self.client.models.generate_content.call_count, 2) + + def test_retry_requests_only_failed_group(self): + self.client.models.generate_content.side_effect = [ + self.rubric_response(self.labels[:2]), RuntimeError('API unavailable')] + with self.assertRaises(RuntimeError): + gemini.refine_existing(self.workspace, 'persp', api_client=self.client) + before = {label: (self.root / 'Persp' / label).read_bytes() for label in self.labels[:2]} + self.client.models.generate_content.reset_mock() + self.client.models.generate_content.side_effect = [self.rubric_response(self.labels[2:])] + gemini.refine_existing(self.workspace, 'persp', api_client=self.client) + self.assertEqual(self.client.models.generate_content.call_count, 1) + for label, content in before.items(): + self.assertEqual((self.root / 'Persp' / label).read_bytes(), content) + + def test_complete_persp_needs_no_api_key_or_requests(self): + for label in self.labels: + (self.root / 'Persp' / label).write_text(label + '\nReviewed rubric') + before = self.snapshot() + with patch.object(gemini, 'api_key', ''), patch.object(gemini.genai, 'Client') as client: + self.assertEqual(gemini.refine_existing(self.workspace, 'persp'), ExitCode.SUCCESS) + client.assert_not_called() self.assertEqual(self.snapshot(), before) + def test_delete_persp_forces_regeneration_and_removes_old_files(self): + for label in self.labels: + (self.root / 'Persp' / label).write_text(label + '\nReviewed rubric') + self.client.models.generate_content.side_effect = [ + self.rubric_response(self.labels[:2]), self.rubric_response(self.labels[2:])] + gemini.refine_existing(self.workspace, 'persp', api_client=self.client, delete_persp=True) + self.assertEqual(self.client.models.generate_content.call_count, 2) + self.assertEqual({path.name for path in (self.root / 'Persp').iterdir()}, set(self.labels)) + + def test_cli_delete_persp_dispatch(self): + with patch.object(gemini, 'refine_existing', return_value=ExitCode.SUCCESS) as refine: + self.assertEqual(gemini.main([str(self.root), '--persp-only', '--delete-persp']), ExitCode.SUCCESS) + self.assertEqual(refine.call_args.kwargs, {'delete_persp': True}) + refine.reset_mock() + self.assertEqual(gemini.main([str(self.root), '--delete-persp']), ExitCode.INVALID_ARGUMENTS) + refine.assert_not_called() + def test_cli_dispatches_selective_modes_without_full_extraction(self): with patch.object(gemini, "refine_existing", return_value=ExitCode.SUCCESS) as refine, patch.object( gemini, "process_exam" @@ -251,6 +345,28 @@ class PersonalStatementGuiTests(unittest.TestCase): self.assertIn(flag, app.command_var.get()) self.assertTrue(app.current_step.optional) self.assertFalse(app.runner.running) + (evaluation / 'labels').write_text('Ex 1\nEx 2\n') + (evaluation / 'Persp').mkdir() + (evaluation / 'Persp' / 'Ex 1').write_text('Ex 1\nReviewed rubric') + (evaluation / 'Persp' / 'Ex 2').write_text('Ex 2\n') + app._refresh_persp_inventory() + self.assertIn('Barèmes présents : 1/2', app.persp_inventory_label.cget('text')) + self.assertIn('Manquants : Ex 2', app.persp_inventory_label.cget('text')) + self.assertFalse(app.arg_vars['delete_persp'].get()) + app.arg_vars['delete_persp'].set(True) + self.assertIn('--delete-persp', app.command_var.get()) + app._save_current_form() + app._render_step() + self.assertFalse(app.arg_vars['delete_persp'].get()) + self.assertNotIn('--delete-persp', app.command_var.get()) + app.tree.selection_set('statement') + app.update() + app._select_variant(1) + app.arg_vars['restart'].set(True) + app._save_current_form() + app._render_step() + self.assertFalse(app.arg_vars['restart'].get()) + self.assertNotIn('--restart', app.command_var.get()) finally: for callback in app.tk.splitlist(app.tk.call("after", "info")): app.after_cancel(callback) diff --git a/tests/test_statement_resume.py b/tests/test_statement_resume.py new file mode 100644 index 0000000..f46c93a --- /dev/null +++ b/tests/test_statement_resume.py @@ -0,0 +1,259 @@ +import io +import json +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from copienator import CliError, EvaluationWorkspace, ExitCode +from copienator.commands import gemini_for_enonce as gemini +from copienator.statement_checkpoints import extract_stage + + +def response(**data): + return SimpleNamespace(text=json.dumps(data)) + + +class ExtractionCheckpointTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + + def solutions(self, request): + return extract_stage(self.root, 'solutions', gemini.SolutionOnlyItem, + 'solution_content', request, labels=['1)', '2)']) + + def test_valid_sibling_survives_and_retry_only_requests_missing(self): + request = Mock(return_value=response(solutions=[ + {'label': '1)', 'solution_content': 'good'}, + {'label': '2)', 'solution_content': 'bad\x01'}])) + output = io.StringIO() + with patch('sys.stdout', output), self.assertRaises(CliError): + self.solutions(request) + saved = json.loads((self.root / 'Cache/statement_solutions.json').read_text()) + self.assertEqual(saved['solutions'], [{'label': '1)', 'solution_content': 'good'}]) + self.assertIn(r'bad\x01', output.getvalue()) + self.assertIn('U+0001', (self.root / 'Cache/statement_errors.log').read_text()) + request = Mock(return_value=response(solutions=[{'label': '2)', 'solution_content': 'fixed'}])) + items = self.solutions(request) + request.assert_called_once_with(['2)']) + self.assertEqual([item.solution_content for item in items], ['good', 'fixed']) + request.reset_mock() + self.solutions(request) + request.assert_not_called() + + def test_malformed_json_and_api_failure_preserve_accepted_items(self): + request = Mock(return_value=response(solutions=[{'label': '1)', 'solution_content': 'good'}])) + with self.assertRaises(CliError): + self.solutions(request) + before = (self.root / 'Cache/statement_solutions.json').read_bytes() + raw = '{"solutions": [broken' + request.return_value = SimpleNamespace(text=raw) + with self.assertRaises(CliError): + self.solutions(request) + self.assertEqual((self.root / 'Cache/statement_solutions.json').read_bytes(), before) + self.assertIn(raw, [p.read_text() for p in (self.root / 'Cache/statement_responses').iterdir()]) + request.side_effect = RuntimeError('offline') + with self.assertRaises(CliError): + self.solutions(request) + self.assertEqual((self.root / 'Cache/statement_solutions.json').read_bytes(), before) + + def test_bad_legacy_cache_is_archived_then_not_replayed(self): + (self.root / 'Cache').mkdir() + legacy = self.root / 'Cache/gemini_solutions.json' + legacy.write_text('broken JSON') + request = Mock(return_value=response(solutions=[ + {'label': label, 'solution_content': 'fixed'} for label in ['1)', '2)']])) + with self.assertRaises(CliError): + self.solutions(request) + request.assert_not_called() + self.solutions(request) + request.assert_called_once_with(['1)', '2)']) + self.assertEqual(legacy.read_text(), 'broken JSON') + + def test_questions_retry_full_inventory_keep_valid_text_and_order(self): + request = Mock(return_value=response(questions=[ + {'label': '1)', 'question_content': 'good'}, + {'label': '2)', 'question_content': 'bad\x01'}])) + def run(): + return extract_stage(self.root, 'questions', gemini.QuestionOnlyItem, 'question_content', request) + with self.assertRaises(CliError): + run() + request.return_value = response(questions=[{'label': '2)', 'question_content': 'fixed'}]) + with self.assertRaises(CliError): + run() # Cannot silently lose the known first question. + request.return_value = response(questions=[{'label': '1)', 'question_content': 'good'}]) + with self.assertRaises(CliError): + run() # Cannot silently omit the previously rejected question either. + request.return_value = response(questions=[ + {'label': '1)', 'question_content': 'rewritten'}, + {'label': '2)', 'question_content': 'fixed'}]) + items = run() + self.assertEqual([item.question_content for item in items], ['good', 'fixed']) + self.assertEqual(request.call_args.args, (None,)) + + def test_contexts_retry_bad_target_and_allow_no_context(self): + request = Mock(return_value=response(contexts=[ + {'target_question_label': '1)', 'last_question_label': '2)', 'context_content': 'good'}, + {'target_question_label': '2)', 'last_question_label': '2)', 'context_content': 'bad\x01'}])) + def run(): + return extract_stage(self.root, 'contexts', gemini.ExtractedContext, 'context_content', + request, labels=['1)', '2)']) + with self.assertRaises(CliError): + run() + request.return_value = response(contexts=[]) + with self.assertRaises(CliError): + run() + request.return_value = response(contexts=[ + {'target_question_label': '2)', 'last_question_label': '2)', 'context_content': 'fixed'}]) + self.assertEqual(len(run()), 2) + self.assertEqual(request.call_args.args, (['2)'],)) + with tempfile.TemporaryDirectory() as root: + self.assertEqual(extract_stage(Path(root), 'contexts', gemini.ExtractedContext, + 'context_content', Mock(return_value=response(contexts=[])), + labels=['1)']), []) + + def test_duplicate_and_unknown_solutions_are_not_silently_accepted(self): + request = Mock(return_value=response(solutions=[ + {'label': '1)', 'solution_content': 'good'}, + {'label': '1)', 'solution_content': 'duplicate'}, + {'label': 'unknown', 'solution_content': 'bad'}])) + with self.assertRaises(CliError): + self.solutions(request) + saved = json.loads((self.root / 'Cache/statement_solutions.json').read_text()) + self.assertFalse(saved['complete']) + self.assertEqual(saved['solutions'], [{'label': '1)', 'solution_content': 'good'}]) + + +class FullStatementResumeTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.workspace = EvaluationWorkspace(self.root) + (self.root / 'enonce.pdf').write_bytes(b'pdf') + (self.root / 'enonce.tex').write_text('source') + (self.root / 'correction.tex').write_text('correction') + self.client = SimpleNamespace(models=SimpleNamespace(generate_content=Mock())) + self.editor = Mock() + self.compiler = Mock(side_effect=lambda content, path, **kwargs: Path(path).write_bytes(b'PDF')) + + def run_exam(self, **kwargs): + with patch.object(gemini.utils, 'edit_file_and_enter', self.editor), patch.object( + gemini, 'compile_to_pdf', self.compiler): + return gemini.process_exam(self.workspace, api_client=self.client, **kwargs) + + def responses(self): + return [response(questions=[{'label': 'Ex 1', 'question_content': 'question'}]), + response(solutions=[{'label': 'Ex 1', 'solution_content': 'answer'}]), + response(contexts=[]), + response(rubrics=[{'label': 'Ex 1', 'rubric_content': '4 points : correct'}])] + + def test_rubric_failure_retry_keeps_grouping_edits_and_sources(self): + self.client.models.generate_content.side_effect = [*self.responses()[:3], RuntimeError('offline')] + self.assertEqual(self.run_exam(), ExitCode.PARTIAL) + self.assertTrue((self.root / 'Cache/statement_grouping.json').is_file()) + self.assertEqual((self.root / 'labels').read_text(), 'Ex 1\n') + tex = self.root / 'Text2/Ex 1.tex' + tex.write_text('manually edited source') + edited_items = (self.root / 'Tmp/exam_items.txt').read_bytes() + self.client.models.generate_content.reset_mock() + self.client.models.generate_content.side_effect = [self.responses()[-1]] + self.editor.reset_mock() + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + self.editor.assert_not_called() + self.assertEqual(self.client.models.generate_content.call_count, 1) + self.assertEqual(tex.read_text(), 'manually edited source') + self.assertEqual((self.root / 'Tmp/exam_items.txt').read_bytes(), edited_items) + + def test_failed_pdf_only_is_compiled_on_retry(self): + self.client.models.generate_content.side_effect = self.responses() + def compile_pdf(content, path, **kwargs): + if Path(path).parent.name == 'Sol2': + Path(path).write_bytes(b'partial PDF from failed compilation') + raise RuntimeError('compiler failure') + Path(path).write_bytes(b'PDF') + self.compiler.side_effect = compile_pdf + self.assertEqual(self.run_exam(), ExitCode.PARTIAL) + good_pdf = (self.root / 'Text2/Ex 1.pdf').read_bytes() + self.compiler.reset_mock() + self.compiler.side_effect = lambda content, path, **kwargs: Path(path).write_bytes(b'PDF') + self.client.models.generate_content.reset_mock() + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + self.client.models.generate_content.assert_not_called() + self.assertEqual(self.compiler.call_count, 1) + self.assertEqual(self.compiler.call_args.args[1], self.root / 'Sol2/Ex 1.pdf') + self.assertEqual((self.root / 'Text2/Ex 1.pdf').read_bytes(), good_pdf) + + def test_interrupted_editor_keeps_edits_until_approval(self): + self.client.models.generate_content.side_effect = self.responses()[:3] + def interrupt(path): + Path(path).write_text('Custom label ### manually edited question\n') + raise KeyboardInterrupt() + self.editor.side_effect = interrupt + with self.assertRaises(KeyboardInterrupt): + self.run_exam() + self.editor.side_effect = None + self.client.models.generate_content.side_effect = [response(rubrics=[ + {'label': 'Custom label', 'rubric_content': '4 points : correct'}])] + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + self.assertIn('manually edited question', (self.root / 'Text2/Custom label.tex').read_text()) + + def test_restart_archives_prior_work_and_reextracts(self): + self.client.models.generate_content.side_effect = self.responses() + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + old = (self.root / 'Persp/Ex 1').read_bytes() + self.client.models.generate_content.side_effect = self.responses() + self.editor.reset_mock() + self.assertEqual(self.run_exam(restart=True), ExitCode.SUCCESS) + self.editor.assert_called_once() + archived = list((self.root / 'Cache/statement_history').glob('*/Persp/Ex 1')) + self.assertEqual(len(archived), 1) + self.assertEqual(archived[0].read_bytes(), old) + + def test_failed_restart_then_resume_still_replaces_old_outputs(self): + self.client.models.generate_content.side_effect = self.responses() + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + self.client.models.generate_content.side_effect = [SimpleNamespace(text='broken JSON')] + with self.assertRaises(CliError): + self.run_exam(restart=True) + self.assertTrue((self.root / 'Persp/Ex 1').exists()) + self.client.models.generate_content.side_effect = self.responses() + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + self.assertTrue(list((self.root / 'Cache/statement_history').glob('*/Persp/Ex 1'))) + + def test_rubric_valid_sibling_saved_even_on_invalid_content(self): + self.client.models.generate_content.return_value = response(rubrics=[ + {'label': 'Ex 1', 'rubric_content': 'good'}, + {'label': 'Ex 2', 'rubric_content': 'bad\x01'}]) + with self.assertRaises(ValueError): + gemini.generate_rubrics(self.client, 'context', workspace=self.workspace, + expected_labels=['Ex 1', 'Ex 2']) + self.assertEqual((self.root / 'Persp/Ex 1').read_text(), 'Ex 1\ngood') + self.assertFalse((self.root / 'Persp/Ex 2').exists()) + + def test_rubric_schema_error_also_preserves_valid_sibling(self): + self.client.models.generate_content.return_value = response(rubrics=[ + {'label': 'Ex 1', 'rubric_content': 'good'}, {'label': 'Ex 2'}]) + with self.assertRaises(ValueError): + gemini.generate_rubrics(self.client, 'context', workspace=self.workspace, + expected_labels=['Ex 1', 'Ex 2']) + self.assertEqual((self.root / 'Persp/Ex 1').read_text(), 'Ex 1\ngood') + + def test_extraction_failure_leaves_existing_outputs_untouched(self): + (self.root / 'Text2').mkdir() + existing = self.root / 'Text2/reviewed.tex' + existing.write_text('manually reviewed question') + self.client.models.generate_content.side_effect = [self.responses()[0], response(solutions=[ + {'label': 'Ex 1', 'solution_content': 'bad\x01'}])] + with self.assertRaises(CliError): + self.run_exam() + self.assertEqual(existing.read_text(), 'manually reviewed question') + self.assertTrue(json.loads((self.root / 'Cache/statement_questions.json').read_text())['complete']) + self.client.models.generate_content.reset_mock() + self.client.models.generate_content.side_effect = self.responses()[1:] + self.assertEqual(self.run_exam(), ExitCode.SUCCESS) + self.assertEqual(self.client.models.generate_content.call_count, 3) + self.assertEqual(existing.read_text(), 'manually reviewed question')