Compare commits
2
Commits
ec8e79bb1a
...
a56fc6e680
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a56fc6e680 | ||
|
|
2a4c5f4f7c |
No files matched your search
@@ -21,6 +21,7 @@ from copienator.commands import grouping
|
||||
from copienator.answer_origins import save_origins
|
||||
from copienator import prompting
|
||||
from copienator.feedback_boxes import valid_feedback_box
|
||||
from copienator.latex_json import repair_latex_json
|
||||
from copienator import (
|
||||
CliError,
|
||||
EvaluationWorkspace,
|
||||
@@ -100,6 +101,16 @@ def report_group_progress(completed: int, total: int) -> None:
|
||||
)
|
||||
|
||||
|
||||
def parse_correction_response(response_text: str):
|
||||
"""Decode model JSON without consuming known LaTeX command prefixes."""
|
||||
return json.loads(
|
||||
repair_latex_json(
|
||||
response_text,
|
||||
additional_content_fields={"text"},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def request_graceful_stop(_signum=None, _frame=None) -> None:
|
||||
"""Stop scheduling Gemini calls while allowing in-flight calls to finish."""
|
||||
if not stop_requested.is_set():
|
||||
@@ -319,7 +330,7 @@ def correct_boxes_with_gemini(pid, label, pdf_path, original_feedbacks,
|
||||
|
||||
contents, config = prompting.request_for_box_correction(pdf_path, original_feedbacks)
|
||||
response_text = call_gemini_with_retries(MODEL_ID_flash, contents, config)
|
||||
corrected_feedbacks = json.loads(response_text)
|
||||
corrected_feedbacks = parse_correction_response(response_text)
|
||||
|
||||
global_feedbacks = [f for f in original_feedbacks if not f["box_2d"]]
|
||||
|
||||
@@ -507,7 +518,7 @@ def process_single_task(task_tuple, precomputed_response=None):
|
||||
tprint(f"Asking Gemini {'Flash' if use_flash else 'Pro '}: {label} {group_name}")
|
||||
full_response_text = call_gemini_with_retries(model_to_use, contents, config)
|
||||
|
||||
json_data = json.loads(full_response_text)
|
||||
json_data = parse_correction_response(full_response_text)
|
||||
|
||||
# Ensure consistency of answer placements
|
||||
for p in json_data:
|
||||
|
||||
@@ -554,11 +554,18 @@ def run(
|
||||
client = genai.Client(api_key=api_key)
|
||||
|
||||
groups = group_images(images)
|
||||
total_groups = len(groups)
|
||||
active: dict[str, tuple[str, float]] = {}
|
||||
progress_lock = Lock()
|
||||
incomplete: list[Path] = []
|
||||
run_started = time.monotonic()
|
||||
|
||||
def print_group_progress(completed: int) -> None:
|
||||
print(
|
||||
f"Label detection progress: {completed}/{total_groups} copy groups completed.",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
def report_progress(group: str, description: str, started: float) -> None:
|
||||
with progress_lock:
|
||||
active[group] = (description, started)
|
||||
@@ -582,6 +589,7 @@ def run(
|
||||
for group_key, files in groups.items()
|
||||
}
|
||||
pending = set(futures)
|
||||
print_group_progress(0)
|
||||
while pending:
|
||||
done, pending = wait(pending, timeout=PROGRESS_INTERVAL)
|
||||
for future in done:
|
||||
@@ -591,6 +599,7 @@ def run(
|
||||
future.result()
|
||||
except IncompleteCopyError as exc:
|
||||
incomplete.extend(exc.images)
|
||||
print_group_progress(total_groups - len(pending))
|
||||
if pending:
|
||||
now = time.monotonic()
|
||||
with progress_lock:
|
||||
|
||||
@@ -313,6 +313,7 @@ class ImageViewer:
|
||||
# Bindings
|
||||
self.root.bind(PLOTTING_KB["OK"], self.on_enter)
|
||||
self.root.bind(PLOTTING_KB["previous"], self.on_previous)
|
||||
self.root.bind("P", self.on_previous)
|
||||
self.root.bind(PLOTTING_KB["edit"], self.on_edit)
|
||||
self.root.bind(PLOTTING_KB["open pdf"], self.on_open_pdf)
|
||||
self.root.bind(PLOTTING_KB["open original pdf"], self.on_open_ori_pdf)
|
||||
|
||||
@@ -16,6 +16,7 @@ from copienator import (
|
||||
read_json,
|
||||
workspace_from_args,
|
||||
)
|
||||
from copienator.latex_json import repair_decoded_latex_controls
|
||||
|
||||
WORD_LIST_FILE = Path(__file__).resolve().parents[1] / "data" / "liste_francais.txt"
|
||||
ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]")
|
||||
@@ -47,6 +48,15 @@ def escape_latex_underscores(text: str) -> str:
|
||||
return "".join(parts)
|
||||
|
||||
|
||||
def escape_latex_percents(text: str) -> str:
|
||||
r"""Escape percent signs before annotation text is passed to LaTeX.
|
||||
|
||||
Unlike underscores, percent signs are comments both inside and outside
|
||||
math mode. Collapsing an existing escape run keeps cleanup idempotent.
|
||||
"""
|
||||
return re.sub(r"\\*%", lambda _match: r"\%", text)
|
||||
|
||||
|
||||
def normalize_overescaped_latex_commands(text: str) -> str:
|
||||
r"""Collapse doubled command escapes inside LaTeX math environments.
|
||||
|
||||
@@ -89,6 +99,7 @@ def fix_hex_corruption_safe(text: str) -> str:
|
||||
|
||||
def repair_json_escape_corruption(text: str) -> str:
|
||||
r"""Restore observed LaTeX commands consumed as JSON control escapes."""
|
||||
text = repair_decoded_latex_controls(text)
|
||||
replacements = (
|
||||
("\x0crac", r"\frac"),
|
||||
("\x0ceuille", r"\equiv"),
|
||||
@@ -116,7 +127,7 @@ def clean_string(text: str, lookup: dict[str, str]) -> str:
|
||||
text = fast_fix(text, lookup).replace("\x00", "")
|
||||
text = repair_json_escape_corruption(text)
|
||||
text = normalize_overescaped_latex_commands(text)
|
||||
return escape_latex_underscores(text)
|
||||
return escape_latex_percents(escape_latex_underscores(text))
|
||||
|
||||
|
||||
def clean_obj(value: Any, lookup: dict[str, str]) -> Any:
|
||||
@@ -132,6 +143,33 @@ def clean_obj(value: Any, lookup: dict[str, str]) -> Any:
|
||||
return value
|
||||
|
||||
|
||||
def unresolved_math_control_paths(value: Any, path: str = "$") -> list[str]:
|
||||
"""Locate destructive JSON control escapes left in mathematical text."""
|
||||
if isinstance(value, str):
|
||||
return (
|
||||
[path]
|
||||
if any(
|
||||
control in {"\b", "\f"}
|
||||
for match in MATH_PATTERN.finditer(value)
|
||||
for control in match.group(0)
|
||||
)
|
||||
else []
|
||||
)
|
||||
if isinstance(value, list):
|
||||
return [
|
||||
issue
|
||||
for index, item in enumerate(value)
|
||||
for issue in unresolved_math_control_paths(item, f"{path}[{index}]")
|
||||
]
|
||||
if isinstance(value, dict):
|
||||
return [
|
||||
issue
|
||||
for key, item in value.items()
|
||||
for issue in unresolved_math_control_paths(item, f"{path}[{key!r}]")
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def run(
|
||||
workspace: EvaluationWorkspace,
|
||||
*,
|
||||
@@ -141,11 +179,18 @@ def run(
|
||||
lookup = build_lookup_map(word_list_path)
|
||||
data = read_json(workspace.correction_file)
|
||||
cleaned = clean_obj(data, lookup)
|
||||
unresolved = unresolved_math_control_paths(cleaned)
|
||||
backup = workspace.root / "correction_precleanup.json"
|
||||
atomic_write_bytes(backup, workspace.correction_file.read_bytes())
|
||||
print(f"Original JSON backed up to {backup}")
|
||||
atomic_write_json(workspace.correction_file, cleaned)
|
||||
print(f"Fixed JSON saved to {workspace.correction_file}")
|
||||
if unresolved:
|
||||
print(
|
||||
"Warning: unresolved JSON control characters remain in mathematical "
|
||||
"text at: " + ", ".join(unresolved)
|
||||
)
|
||||
return ExitCode.PARTIAL
|
||||
return ExitCode.SUCCESS
|
||||
|
||||
|
||||
|
||||
@@ -13,6 +13,9 @@ _CONTENT_FIELDS = {'question_content', 'solution_content', 'context_content', 'r
|
||||
_CONTENT_KEYS = {f'"{field}"' for field in _CONTENT_FIELDS}
|
||||
_STRING = re.compile(r'"(?:\\.|[^"\\])*"', re.DOTALL)
|
||||
_COMMAND = re.compile(r'[a-zA-Z]+')
|
||||
_MATH = re.compile(r"(\$\$.*?\$\$|\$.*?\$|\\\(.*?\\\)|\\\[.*?\\\])", re.DOTALL)
|
||||
_DECODED_COMMAND = re.compile(r"[\b\f\n\r\t]([A-Za-z]+)")
|
||||
_JSON_CONTROL_PREFIX = {"\b": "b", "\f": "f", "\n": "n", "\r": "r", "\t": "t"}
|
||||
|
||||
|
||||
def normalize_literal_newlines(text: str) -> str:
|
||||
@@ -44,6 +47,29 @@ underbrace underline underset uparrow updownarrow upsilon usepackage
|
||||
'''.split())
|
||||
|
||||
|
||||
def repair_decoded_latex_controls(text: str) -> str:
|
||||
r"""Restore known TeX commands after JSON decoded their first letter.
|
||||
|
||||
Only math spans and whitelisted command names are changed. This avoids
|
||||
treating ordinary paragraph newlines or tabs as damaged LaTeX.
|
||||
"""
|
||||
def repair_math(math_match: re.Match[str]) -> str:
|
||||
def repair_command(command_match: re.Match[str]) -> str:
|
||||
candidate = (
|
||||
_JSON_CONTROL_PREFIX[command_match.group(0)[0]]
|
||||
+ command_match.group(1)
|
||||
)
|
||||
return (
|
||||
f"\\{candidate}"
|
||||
if candidate in _AMBIGUOUS_COMMANDS
|
||||
else command_match.group(0)
|
||||
)
|
||||
|
||||
return _DECODED_COMMAND.sub(repair_command, math_match.group(0))
|
||||
|
||||
return _MATH.sub(repair_math, text)
|
||||
|
||||
|
||||
def _repair_string(token: str) -> str:
|
||||
output = []
|
||||
index = 0
|
||||
@@ -71,7 +97,9 @@ def _repair_string(token: str) -> str:
|
||||
return ''.join(output)
|
||||
|
||||
|
||||
def repair_latex_json(raw: str) -> str:
|
||||
def repair_latex_json(
|
||||
raw: str, *, additional_content_fields: set[str] | None = None
|
||||
) -> str:
|
||||
"""Repair single TeX backslashes without decoding or rewriting other JSON.
|
||||
|
||||
Preserve standard escapes, Unicode escapes, and already escaped TeX.
|
||||
@@ -79,14 +107,22 @@ def repair_latex_json(raw: str) -> str:
|
||||
as a broken ``\\not`` command; known TeX command names take precedence.
|
||||
Raw responses/caches should be retained to allow reviewing such cases.
|
||||
"""
|
||||
additional_content_keys = {
|
||||
f'"{field}"' for field in (additional_content_fields or set())
|
||||
}
|
||||
output = []
|
||||
end = 0
|
||||
previous = None
|
||||
for match in _STRING.finditer(raw):
|
||||
token = match.group()
|
||||
if (previous is not None and raw[previous.end():match.start()].strip() == ':'
|
||||
and previous.group() in _CONTENT_KEYS):
|
||||
token = _repair_string(token)
|
||||
if previous is not None and raw[previous.end():match.start()].strip() == ':':
|
||||
if previous.group() in _CONTENT_KEYS:
|
||||
token = _repair_string(token)
|
||||
elif previous.group() in additional_content_keys:
|
||||
token = _MATH.sub(
|
||||
lambda math_match: _repair_string(math_match.group(0)),
|
||||
token,
|
||||
)
|
||||
output.extend((raw[end:match.start()], token))
|
||||
end = match.end()
|
||||
previous = match
|
||||
|
||||
@@ -141,6 +141,13 @@ import subprocess
|
||||
|
||||
from .configuration import LATEX_AFTER, LATEX_BEFORE
|
||||
|
||||
LATEX_WARNING_MARKER = "[Copienator:latex-warning] "
|
||||
|
||||
|
||||
def report_latex_warning(output_pdf_path) -> None:
|
||||
"""Emit a stable, machine-readable path alongside human diagnostics."""
|
||||
print(f"{LATEX_WARNING_MARKER}{Path(output_pdf_path).resolve()}")
|
||||
|
||||
|
||||
def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
"""Wraps text in standalone header/footer templates and compiles it to PDF."""
|
||||
@@ -185,6 +192,7 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
shutil.move(generated_pdf, output_pdf_path)
|
||||
|
||||
if result.returncode != 0:
|
||||
report_latex_warning(output_pdf_path)
|
||||
lines = result.stdout.splitlines()
|
||||
first = next((i for i, line in enumerate(lines) if line.lstrip().startswith("!")), None)
|
||||
detail = "\n" + "\n".join(lines[first:first + 5]) if first is not None else ""
|
||||
@@ -196,10 +204,12 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
if strict:
|
||||
raise RuntimeError(f"LaTeX failed for {output_pdf_path}{detail}")
|
||||
elif not pdf_generated:
|
||||
report_latex_warning(output_pdf_path)
|
||||
print(f"Warning: LaTeX compilation produced no PDF for {output_pdf_path}")
|
||||
if strict:
|
||||
raise RuntimeError(f"No PDF generated for {output_pdf_path}")
|
||||
except Exception as e:
|
||||
if strict:
|
||||
raise
|
||||
report_latex_warning(output_pdf_path)
|
||||
print(f"Compilation error for {output_pdf_path}: {e}")
|
||||
+121
-5
@@ -20,6 +20,7 @@ from copienator.platform import (
|
||||
open_path,
|
||||
validate_windows_labels,
|
||||
)
|
||||
from copienator.utils import LATEX_WARNING_MARKER
|
||||
|
||||
from .diagnostics import collect_diagnostics
|
||||
from .batch_monitor import BatchMonitor
|
||||
@@ -281,6 +282,8 @@ class CopienatorApp(tk.Tk):
|
||||
self.pending_refaire_commands: list[list[str]] = []
|
||||
self.refaire_command_index = 0
|
||||
self._correction_progress_buffer = ""
|
||||
self._latex_warning_buffer = ""
|
||||
self._latex_warning_paths: list[Path] = []
|
||||
|
||||
self.title("Copienator — assistant de correction")
|
||||
self.geometry("1180x820")
|
||||
@@ -325,7 +328,7 @@ class CopienatorApp(tk.Tk):
|
||||
|
||||
top = ttk.Frame(self, padding=(10, 8))
|
||||
top.grid(row=0, column=0, sticky="ew")
|
||||
top.columnconfigure(5, weight=1)
|
||||
top.columnconfigure(6, weight=1)
|
||||
ttk.Button(top, text="Charger", command=self._load_evaluation).grid(
|
||||
row=0, column=0, padx=(0, 8)
|
||||
)
|
||||
@@ -333,14 +336,19 @@ class CopienatorApp(tk.Tk):
|
||||
self.evaluation_entry = ttk.Entry(top, textvariable=self.evaluation_var, width=42)
|
||||
self.evaluation_entry.grid(row=0, column=2, sticky="w")
|
||||
self.evaluation_entry.bind("<Return>", lambda _event: self._load_evaluation())
|
||||
self.browse_evaluation_button = ttk.Button(
|
||||
top, text="Parcourir…", command=self._browse_evaluation
|
||||
)
|
||||
self.browse_evaluation_button.grid(row=0, column=3, padx=(6, 0))
|
||||
self.open_evaluation_button = ttk.Button(
|
||||
top, text="Ouvrir le dossier", command=self._open_evaluation_folder
|
||||
)
|
||||
self.open_evaluation_button.grid(row=0, column=3, padx=(6, 0))
|
||||
ttk.Button(top, text="Parcourir…", command=self._browse_evaluation).grid(
|
||||
row=0, column=4, padx=6
|
||||
self.open_evaluation_button.grid(row=0, column=4, padx=(12, 0))
|
||||
self.open_statement_button = ttk.Button(
|
||||
top, text="Ouvrir l’énoncé", command=self._open_statement_pdf
|
||||
)
|
||||
ttk.Button(top, text="Diagnostic…", command=self._show_diagnostics).grid(row=0, column=6, padx=(6, 0))
|
||||
self.open_statement_button.grid(row=0, column=5, padx=(6, 0))
|
||||
ttk.Button(top, text="Diagnostic…", command=self._show_diagnostics).grid(row=0, column=7, padx=(6, 0))
|
||||
|
||||
ttk.Label(top, text="Clé Gemini").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=(7, 0))
|
||||
ttk.Entry(top, textvariable=self.api_key_var, show="•", width=32).grid(
|
||||
@@ -488,6 +496,13 @@ class CopienatorApp(tk.Tk):
|
||||
clipboard_actions.grid(row=1, column=0, columnspan=6, sticky="w", pady=(5, 0))
|
||||
ttk.Button(clipboard_actions, text="Copier la sélection", command=self._copy_console_selection).pack(side="left")
|
||||
ttk.Button(clipboard_actions, text="Copier toute la sortie", command=self._copy_console_all).pack(side="left", padx=6)
|
||||
self.latex_warning_button = ttk.Button(
|
||||
clipboard_actions,
|
||||
text="Problèmes LaTeX…",
|
||||
command=self._show_current_latex_warnings,
|
||||
)
|
||||
self.latex_warning_button.pack(side="left", padx=6)
|
||||
self.latex_warning_button.pack_forget()
|
||||
ttk.Label(console_actions, text="Réponse au script").grid(row=0, column=0, padx=(0, 6))
|
||||
self.stdin_var = tk.StringVar()
|
||||
self.stdin_entry = ttk.Entry(console_actions, textvariable=self.stdin_var)
|
||||
@@ -560,6 +575,11 @@ class CopienatorApp(tk.Tk):
|
||||
def _open_evaluation_folder(self) -> None:
|
||||
self._open_desktop_path(self.evaluation, "dossier d’évaluation")
|
||||
|
||||
def _open_statement_pdf(self) -> None:
|
||||
evaluation = self.evaluation
|
||||
statement = EvaluationWorkspace(evaluation).statement_pdf_file if evaluation else None
|
||||
self._open_desktop_path(statement, "fichier enonce.pdf")
|
||||
|
||||
def _toggle_proxy(self) -> None:
|
||||
self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled")
|
||||
|
||||
@@ -572,6 +592,7 @@ class CopienatorApp(tk.Tk):
|
||||
messagebox.showerror("Dossier invalide", "Choisissez un dossier d’évaluation existant.")
|
||||
return
|
||||
self.batch_monitor.stop()
|
||||
self._reset_latex_warnings()
|
||||
self._save_current_form()
|
||||
self.state_store.load(evaluation)
|
||||
for step in self.steps:
|
||||
@@ -1522,6 +1543,8 @@ class CopienatorApp(tk.Tk):
|
||||
last_run_values=run_values,
|
||||
)
|
||||
self.active_step_id = step.id
|
||||
if step.id == "statement":
|
||||
self._reset_latex_warnings()
|
||||
if step.id == "correction":
|
||||
self._correction_progress_buffer = ""
|
||||
self.info_var.set("Correction : préparation des groupes…")
|
||||
@@ -1632,6 +1655,7 @@ class CopienatorApp(tk.Tk):
|
||||
if event == "output":
|
||||
output = str(payload)
|
||||
self._track_correction_progress(output)
|
||||
self._track_latex_warnings(output)
|
||||
self._append_console(output)
|
||||
elif event == "input_echo":
|
||||
self._append_console(f"> {payload}")
|
||||
@@ -1660,6 +1684,88 @@ class CopienatorApp(tk.Tk):
|
||||
f"({percent} %)."
|
||||
)
|
||||
|
||||
def _reset_latex_warnings(self) -> None:
|
||||
self._latex_warning_buffer = ""
|
||||
self._latex_warning_paths = []
|
||||
if hasattr(self, "latex_warning_button"):
|
||||
self.latex_warning_button.pack_forget()
|
||||
|
||||
def _track_latex_warnings(self, output: str, *, flush: bool = False) -> None:
|
||||
if self.active_step_id != "statement":
|
||||
return
|
||||
combined = self._latex_warning_buffer + output
|
||||
lines = combined.splitlines(keepends=True)
|
||||
self._latex_warning_buffer = ""
|
||||
if lines and not lines[-1].endswith(("\n", "\r")) and not flush:
|
||||
self._latex_warning_buffer = lines.pop()
|
||||
for line in lines:
|
||||
marker_at = line.find(LATEX_WARNING_MARKER)
|
||||
if marker_at < 0:
|
||||
continue
|
||||
raw_path = line[marker_at + len(LATEX_WARNING_MARKER):].strip()
|
||||
if not raw_path:
|
||||
continue
|
||||
path = Path(raw_path).expanduser()
|
||||
if not path.is_absolute() and self.evaluation:
|
||||
path = self.evaluation / path
|
||||
path = path.resolve()
|
||||
if path not in self._latex_warning_paths:
|
||||
self._latex_warning_paths.append(path)
|
||||
if self._latex_warning_paths:
|
||||
self.latex_warning_button.pack(side="left", padx=6)
|
||||
|
||||
def _show_current_latex_warnings(self) -> None:
|
||||
if not self._latex_warning_paths:
|
||||
messagebox.showinfo("Compilation LaTeX", "Aucun problème de compilation détecté.")
|
||||
return
|
||||
dialog = tk.Toplevel(self)
|
||||
dialog.title("Problèmes de compilation LaTeX")
|
||||
dialog.geometry("900x380")
|
||||
dialog.transient(self)
|
||||
dialog.columnconfigure(0, weight=1)
|
||||
dialog.rowconfigure(1, weight=1)
|
||||
ttk.Label(
|
||||
dialog,
|
||||
text=(
|
||||
f"{len(self._latex_warning_paths)} fichier(s) n’ont pas été compilés "
|
||||
"correctement. Sélectionnez-en un pour ouvrir ses éléments."
|
||||
),
|
||||
padding=10,
|
||||
wraplength=850,
|
||||
justify="left",
|
||||
).grid(row=0, column=0, sticky="ew")
|
||||
paths = tk.Listbox(dialog, exportselection=False)
|
||||
for pdf_path in self._latex_warning_paths:
|
||||
try:
|
||||
label = str(pdf_path.relative_to(self.evaluation)) if self.evaluation else str(pdf_path)
|
||||
except ValueError:
|
||||
label = str(pdf_path)
|
||||
paths.insert("end", label)
|
||||
paths.grid(row=1, column=0, sticky="nsew", padx=10)
|
||||
paths.selection_set(0)
|
||||
|
||||
actions = ttk.Frame(dialog, padding=10)
|
||||
actions.grid(row=2, column=0, sticky="ew")
|
||||
|
||||
def selected_pdf() -> Path:
|
||||
selection = paths.curselection()
|
||||
index = selection[0] if selection else 0
|
||||
return self._latex_warning_paths[index]
|
||||
|
||||
def open_folder() -> None:
|
||||
self._open_desktop_path(selected_pdf().parent, "dossier de compilation")
|
||||
|
||||
def open_source() -> None:
|
||||
self._open_desktop_path(selected_pdf().with_suffix(".tex"), "fichier source LaTeX")
|
||||
|
||||
def open_pdf() -> None:
|
||||
self._open_desktop_path(selected_pdf(), "fichier PDF compilé")
|
||||
|
||||
ttk.Button(actions, text="Ouvrir le dossier", command=open_folder).pack(side="left")
|
||||
ttk.Button(actions, text="Ouvrir le .tex", command=open_source).pack(side="left", padx=6)
|
||||
ttk.Button(actions, text="Ouvrir le PDF", command=open_pdf).pack(side="left")
|
||||
ttk.Button(actions, text="Fermer", command=dialog.destroy).pack(side="right")
|
||||
|
||||
def _batch_watch_status_changed(self, message: str) -> None:
|
||||
self.batch_watch_status.set(message)
|
||||
self._update_controls()
|
||||
@@ -1700,6 +1806,8 @@ class CopienatorApp(tk.Tk):
|
||||
step_id = self.active_step_id
|
||||
if not step_id:
|
||||
return
|
||||
if step_id == "statement":
|
||||
self._track_latex_warnings("", flush=True)
|
||||
status = process_status(return_code, interrupted)
|
||||
if status == "success" and self.pending_refaire_commands:
|
||||
command = self.pending_refaire_commands.pop(0)
|
||||
@@ -1780,9 +1888,17 @@ class CopienatorApp(tk.Tk):
|
||||
title = self.step_by_id[step_id].title
|
||||
self._append_console(f"\n[Terminé — code {return_code} — {STATUS_LABELS[status]}]\n")
|
||||
self.info_var.set(f"{title} : {STATUS_LABELS[status].lower()} (code {return_code}).")
|
||||
if step_id == "statement" and self._latex_warning_paths:
|
||||
count = len(self._latex_warning_paths)
|
||||
self.info_var.set(
|
||||
f"{title} : {STATUS_LABELS[status].lower()} (code {return_code}) — "
|
||||
f"{count} problème(s) LaTeX."
|
||||
)
|
||||
self.active_step_id = None
|
||||
self._populate_tree()
|
||||
self._update_controls()
|
||||
if step_id == "statement" and self._latex_warning_paths:
|
||||
self.after_idle(self._show_current_latex_warnings)
|
||||
if step_id == "personal_read":
|
||||
if status == "success":
|
||||
self.info_var.set("Commande terminée. Poursuivez la relecture, puis cliquez sur « Marquer réussie ».")
|
||||
|
||||
@@ -4,12 +4,13 @@ import subprocess
|
||||
import threading
|
||||
from tkinter import ttk, messagebox
|
||||
|
||||
SERVICE = "syncthing@sebastien.service"
|
||||
SERVICE = "syncthing.service"
|
||||
SYSTEMCTL = ("systemctl", "--user", "--no-ask-password")
|
||||
|
||||
|
||||
def service_status():
|
||||
result = subprocess.run(
|
||||
["systemctl", "show", SERVICE, "--property=ActiveState", "--value"],
|
||||
[*SYSTEMCTL, "show", SERVICE, "--property=ActiveState", "--value"],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
)
|
||||
if result.returncode:
|
||||
@@ -22,7 +23,7 @@ def control_service(action):
|
||||
return "active"
|
||||
command = "start" if action == "ensure" else action
|
||||
result = subprocess.run(
|
||||
["systemctl", "--no-ask-password", command, SERVICE],
|
||||
[*SYSTEMCTL, command, SERVICE],
|
||||
capture_output=True, text=True, timeout=60,
|
||||
)
|
||||
if result.returncode:
|
||||
@@ -87,13 +88,18 @@ class SyncthingPanel(ttk.Frame):
|
||||
self.start_button.configure(state="normal")
|
||||
self.stop_button.configure(state="normal")
|
||||
self.status.configure(text=f"{SERVICE} : {error or status}")
|
||||
if error and action != "status":
|
||||
self.pending_export = None
|
||||
messagebox.showerror("Syncthing", error)
|
||||
elif action == "ensure":
|
||||
if action == "ensure":
|
||||
callback, self.pending_export = self.pending_export, None
|
||||
if error:
|
||||
messagebox.showwarning(
|
||||
"Syncthing",
|
||||
f"Impossible de démarrer Syncthing : {error}\n\n"
|
||||
"L’export va continuer.",
|
||||
)
|
||||
if callback:
|
||||
callback()
|
||||
elif error and action != "status":
|
||||
messagebox.showerror("Syncthing", error)
|
||||
elif self.pending_export:
|
||||
self.request("ensure")
|
||||
if self.winfo_exists():
|
||||
|
||||
@@ -706,7 +706,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
|
||||
"personal_deploy",
|
||||
"Étapes personnelles",
|
||||
"Déployer et publier les copies",
|
||||
"Étape manuelle : déployer miqmacs-copies-assets puis mettre à jour les copies depuis l’administration.",
|
||||
"Étape manuelle : déployer miqmacs-copies-assets et la section `Interro` (inutile : mettre à jour les copies depuis l’administration).",
|
||||
(manual("deploy"),),
|
||||
personal=True,
|
||||
),
|
||||
|
||||
@@ -9,6 +9,7 @@ from copienator_gui.app import CopienatorApp
|
||||
from copienator_gui.workflow import command_display
|
||||
from copienator.commands.page_splitter import _selected_inputs
|
||||
from copienator import EvaluationWorkspace
|
||||
from copienator.utils import LATEX_WARNING_MARKER
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get("DISPLAY"), "Tk requires a display")
|
||||
@@ -99,15 +100,51 @@ class GuiConvenienceTests(unittest.TestCase):
|
||||
self.assertEqual(self.app.state_store.step("statement_persp")["status"], expected)
|
||||
self.assertEqual(self.app.state_store.step("statement_groups")["status"], "ready")
|
||||
|
||||
def test_compact_evaluation_input_and_open_folder_button(self):
|
||||
def test_statement_latex_warnings_are_tracked_for_both_variants(self):
|
||||
for variant in ("gemini", "personal"):
|
||||
with self.subTest(variant=variant):
|
||||
self.app._reset_latex_warnings()
|
||||
self.app.state_store.update_step("statement", last_run_variant=variant)
|
||||
self.app.active_step_id = "statement"
|
||||
first = self.evaluation / "Text2" / "Ex 1.pdf"
|
||||
second = self.evaluation / "Sol2" / "Ex 2.pdf"
|
||||
marker = LATEX_WARNING_MARKER
|
||||
self.app._track_latex_warnings(f"noise\n{marker}{first}")
|
||||
self.assertEqual(self.app._latex_warning_paths, [])
|
||||
self.app._track_latex_warnings(f"\n{marker}{second}\n{marker}{first}\n")
|
||||
self.assertEqual(self.app._latex_warning_paths, [first, second])
|
||||
self.assertTrue(self.app.latex_warning_button.winfo_manager())
|
||||
with patch.object(self.app, "after_idle") as after_idle:
|
||||
self.app._finish_process(4, False)
|
||||
self.assertTrue(
|
||||
any(call.args[0] == self.app._show_current_latex_warnings
|
||||
for call in after_idle.call_args_list)
|
||||
)
|
||||
self.assertIn("2 problème(s) LaTeX", self.app.info_var.get())
|
||||
|
||||
def test_evaluation_buttons_are_ordered_and_open_folder_and_statement(self):
|
||||
self.assertEqual(int(self.app.evaluation_entry.cget("width")), 42)
|
||||
self.assertEqual(
|
||||
int(self.app.open_evaluation_button.grid_info()["column"]),
|
||||
int(self.app.browse_evaluation_button.grid_info()["column"]),
|
||||
int(self.app.evaluation_entry.grid_info()["column"]) + 1,
|
||||
)
|
||||
self.assertEqual(
|
||||
int(self.app.open_evaluation_button.grid_info()["column"]),
|
||||
int(self.app.browse_evaluation_button.grid_info()["column"]) + 1,
|
||||
)
|
||||
self.assertEqual(
|
||||
int(self.app.open_statement_button.grid_info()["column"]),
|
||||
int(self.app.open_evaluation_button.grid_info()["column"]) + 1,
|
||||
)
|
||||
statement = self.evaluation / "enonce.pdf"
|
||||
statement.touch()
|
||||
with patch("copienator_gui.app.open_path") as opened:
|
||||
self.app.open_evaluation_button.invoke()
|
||||
opened.assert_called_once_with(self.evaluation)
|
||||
self.app.open_statement_button.invoke()
|
||||
self.assertEqual(
|
||||
opened.call_args_list,
|
||||
[unittest.mock.call(self.evaluation), unittest.mock.call(statement)],
|
||||
)
|
||||
|
||||
def test_manual_resolution_panel_follows_command_target(self):
|
||||
manual = self.evaluation / "manual_resolutions.txt"
|
||||
|
||||
@@ -21,6 +21,7 @@ from pypdf import PdfReader, PdfWriter
|
||||
from copienator import (
|
||||
CliError,
|
||||
EvaluationWorkspace,
|
||||
ExitCode,
|
||||
WorkspaceNotFoundError,
|
||||
WorkspaceValidationError,
|
||||
atomic_update_json,
|
||||
@@ -1058,6 +1059,26 @@ class StandardCliTests(unittest.TestCase):
|
||||
previous_image, previous_json, previous_meta
|
||||
)
|
||||
|
||||
def test_plotting_accepts_lowercase_and_uppercase_previous_shortcuts(self) -> None:
|
||||
module = self.modules["plotting"]
|
||||
root = Mock()
|
||||
root.winfo_screenwidth.return_value = 1200
|
||||
root.winfo_screenheight.return_value = 800
|
||||
label = Mock()
|
||||
|
||||
with patch.object(module.tk, "Label", return_value=label), patch.object(
|
||||
module.ImageViewer, "poll_queue"
|
||||
):
|
||||
viewer = module.ImageViewer(
|
||||
root,
|
||||
SimpleNamespace(root=Path("/evaluation")),
|
||||
set(),
|
||||
queue.Queue(),
|
||||
)
|
||||
|
||||
root.bind.assert_any_call("p", viewer.on_previous)
|
||||
root.bind.assert_any_call("P", viewer.on_previous)
|
||||
|
||||
def test_plotting_validates_plain_and_directional_labels(self) -> None:
|
||||
module = self.modules["plotting"]
|
||||
self.assertEqual(
|
||||
@@ -1317,6 +1338,16 @@ class StandardCliTests(unittest.TestCase):
|
||||
self.assertEqual(result["list"][0]["label"], "??Ex 10 : 1)a)")
|
||||
self.assertIn("keeping them with a ?? prefix", output.getvalue())
|
||||
|
||||
def test_correction_response_preserves_ambiguous_latex_commands(self) -> None:
|
||||
module = self.modules["correction"]
|
||||
parsed = module.parse_correction_response(
|
||||
r'[{"result":{"feedback":[{"text":"$\forall x, f \big(x\big) \to x$"}]}}]'
|
||||
)
|
||||
self.assertEqual(
|
||||
parsed[0]["result"]["feedback"][0]["text"],
|
||||
r"$\forall x, f \big(x\big) \to x$",
|
||||
)
|
||||
|
||||
def test_correction_overwrite_keeps_previous_state_until_a_commit(self) -> None:
|
||||
module = self.modules["correction"]
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
@@ -1754,6 +1785,18 @@ class StandardCliTests(unittest.TestCase):
|
||||
r"valid $\begin{cases} A=0 \\ B=1 \end{cases}$",
|
||||
)
|
||||
self.assertEqual(module.clean_string(cleaned_text, {}), cleaned_text)
|
||||
percent_text = r"Code `n % p`, prose 50%, math $n \% p$ and $50%$."
|
||||
escaped_percent_text = (
|
||||
r"Code `n \% p`, prose 50\%, math $n \% p$ and $50\%$."
|
||||
)
|
||||
self.assertEqual(
|
||||
module.clean_string(percent_text, {}),
|
||||
escaped_percent_text,
|
||||
)
|
||||
self.assertEqual(
|
||||
module.clean_string(escaped_percent_text, {}),
|
||||
escaped_percent_text,
|
||||
)
|
||||
corrupted_json_escapes = (
|
||||
"Formulas $"
|
||||
+ "\x0c"
|
||||
@@ -1774,6 +1817,23 @@ class StandardCliTests(unittest.TestCase):
|
||||
r"Formulas $\frac{1}{2}$, $\theta \equiv 0 [2\pi]$, "
|
||||
r"$e \times x$, and [\negthinspace[0,n]",
|
||||
)
|
||||
decoded_math_commands = (
|
||||
"$"
|
||||
+ "\x0corall x, f "
|
||||
+ "\x08ig(x\x08ig) "
|
||||
+ "\to "
|
||||
+ "\text{false}$"
|
||||
)
|
||||
self.assertEqual(
|
||||
module.clean_string(decoded_math_commands, {}),
|
||||
r"$\forall x, f \big(x\big) \to \text{false}$",
|
||||
)
|
||||
self.assertEqual(
|
||||
module.unresolved_math_control_paths(
|
||||
{"feedback": ["Plain\fprose", "$unknown\fform$"]}
|
||||
),
|
||||
["$['feedback'][1]"],
|
||||
)
|
||||
self.assertEqual(
|
||||
module.clean_string("x ∈ A and B ⊂ C", {}),
|
||||
r"x \ensuremath{\in} A and B \ensuremath{\subset} C",
|
||||
@@ -1800,6 +1860,19 @@ class StandardCliTests(unittest.TestCase):
|
||||
self.assertEqual((evaluation / "correction.json").read_bytes(), cleaned)
|
||||
self.assertEqual(backup.read_bytes(), original)
|
||||
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
evaluation = Path(directory) / "Exam"
|
||||
evaluation.mkdir()
|
||||
atomic_write_json(
|
||||
evaluation / "correction.json",
|
||||
{"text": "$unknown\fcommand$"},
|
||||
)
|
||||
output = io.StringIO()
|
||||
with redirect_stdout(output):
|
||||
result = module.run(EvaluationWorkspace(evaluation))
|
||||
self.assertEqual(result, ExitCode.PARTIAL)
|
||||
self.assertIn("$['text']", output.getvalue())
|
||||
|
||||
def test_manual_resolution_with_no_actions_is_safe(self) -> None:
|
||||
module = self.modules["resolve_manual"]
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
|
||||
@@ -104,6 +104,8 @@ class LabelTimeoutTests(unittest.TestCase):
|
||||
patch.object(labels, "wait", side_effect=wait), redirect_stdout(output):
|
||||
result = labels.run(self.workspace, [], client=self.client, sleep=lambda _: None)
|
||||
self.assertEqual(result, ExitCode.SUCCESS)
|
||||
self.assertIn("Label detection progress: 0/1 copy groups completed.", output.getvalue())
|
||||
self.assertIn("Label detection progress: 1/1 copy groups completed.", output.getvalue())
|
||||
self.assertIn("1 copies outstanding", output.getvalue())
|
||||
self.assertIn("Copie01_01.jpg, attempt 1/3", output.getvalue())
|
||||
self.assertIn("elapsed for image", output.getvalue())
|
||||
@@ -1,7 +1,11 @@
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from copienator.latex_json import repair_latex_json, normalize_literal_newlines
|
||||
from copienator.latex_json import (
|
||||
normalize_literal_newlines,
|
||||
repair_decoded_latex_controls,
|
||||
repair_latex_json,
|
||||
)
|
||||
from copienator.commands.gemini_for_enonce import ExamQuestions, ExamSolutions, ExamContext, GroupRubrics
|
||||
|
||||
|
||||
@@ -44,6 +48,50 @@ class LatexJsonTests(unittest.TestCase):
|
||||
self.assertEqual(repair_latex_json(raw),
|
||||
r'{"label":"\not", "other":"\frac", "rubric_content":"\\frac"}')
|
||||
|
||||
def test_feedback_text_can_be_repaired_before_json_decoding(self):
|
||||
raw = (
|
||||
r'{"text":"$\forall x, f \big(x\big) \neq x \to \text{false}$",'
|
||||
r'"other":"\forall"}'
|
||||
)
|
||||
repaired = repair_latex_json(
|
||||
raw,
|
||||
additional_content_fields={"text"},
|
||||
)
|
||||
parsed = json.loads(repaired)
|
||||
self.assertEqual(
|
||||
parsed["text"],
|
||||
r"$\forall x, f \big(x\big) \neq x \to \text{false}$",
|
||||
)
|
||||
self.assertEqual(parsed["other"], "\x0corall")
|
||||
|
||||
valid_prose = json.dumps({"text": "First line\ntext outside math"})
|
||||
self.assertEqual(
|
||||
repair_latex_json(
|
||||
valid_prose,
|
||||
additional_content_fields={"text"},
|
||||
),
|
||||
valid_prose,
|
||||
)
|
||||
|
||||
def test_decoded_control_repair_is_limited_to_known_commands_in_math(self):
|
||||
corrupted = (
|
||||
"$"
|
||||
+ "\x0corall x, f "
|
||||
+ "\x08ig(x\x08ig) "
|
||||
+ "\neq x "
|
||||
+ "\to "
|
||||
+ "\text{false}$"
|
||||
)
|
||||
self.assertEqual(
|
||||
repair_decoded_latex_controls(corrupted),
|
||||
r"$\forall x, f \big(x\big) \neq x \to \text{false}$",
|
||||
)
|
||||
ordinary_whitespace = "First paragraph\nnext line\tindented"
|
||||
self.assertEqual(
|
||||
repair_decoded_latex_controls(ordinary_whitespace),
|
||||
ordinary_whitespace,
|
||||
)
|
||||
|
||||
def test_does_not_repair_structural_errors(self):
|
||||
with self.assertRaises(json.JSONDecodeError):
|
||||
json.loads(repair_latex_json(r'{"question_content":"\frac", broken}'))
|
||||
+43
-3
@@ -1,8 +1,16 @@
|
||||
import subprocess
|
||||
import os
|
||||
import tkinter as tk
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from copienator_gui.syncthing import SERVICE, control_service
|
||||
from copienator_gui.syncthing import (
|
||||
SERVICE,
|
||||
SYSTEMCTL,
|
||||
SyncthingPanel,
|
||||
control_service,
|
||||
service_status,
|
||||
)
|
||||
|
||||
|
||||
class SyncthingTests(unittest.TestCase):
|
||||
@@ -16,7 +24,7 @@ class SyncthingTests(unittest.TestCase):
|
||||
with patch("copienator_gui.syncthing.service_status", side_effect=["inactive", "active"]), \
|
||||
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
|
||||
self.assertEqual(control_service("ensure"), "active")
|
||||
self.assertEqual(run.call_args.args[0], ["systemctl", "--no-ask-password", "start", SERVICE])
|
||||
self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "start", SERVICE])
|
||||
|
||||
def test_failed_start_reports_error(self):
|
||||
with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \
|
||||
@@ -28,4 +36,36 @@ class SyncthingTests(unittest.TestCase):
|
||||
with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \
|
||||
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
|
||||
self.assertEqual(control_service("stop"), "inactive")
|
||||
self.assertEqual(run.call_args.args[0], ["systemctl", "--no-ask-password", "stop", SERVICE])
|
||||
self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "stop", SERVICE])
|
||||
|
||||
def test_status_uses_the_non_privileged_user_service(self):
|
||||
with patch(
|
||||
"copienator_gui.syncthing.subprocess.run",
|
||||
return_value=subprocess.CompletedProcess([], 0, stdout="active\n"),
|
||||
) as run:
|
||||
self.assertEqual(service_status(), "active")
|
||||
self.assertEqual(
|
||||
run.call_args.args[0],
|
||||
[*SYSTEMCTL, "show", SERVICE, "--property=ActiveState", "--value"],
|
||||
)
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get("DISPLAY"), "Tk requires a display")
|
||||
class SyncthingPanelTests(unittest.TestCase):
|
||||
def test_failed_automatic_start_warns_and_continues_export(self):
|
||||
root = tk.Tk()
|
||||
self.addCleanup(root.destroy)
|
||||
with patch("copienator_gui.syncthing.threading.Thread.start"):
|
||||
panel = SyncthingPanel(root)
|
||||
panel.pack()
|
||||
continued = []
|
||||
panel.pending_export = lambda: continued.append(True)
|
||||
panel.busy = True
|
||||
panel.events.put(("ensure", None, "Access denied"))
|
||||
|
||||
with patch("copienator_gui.syncthing.messagebox.showwarning") as warning:
|
||||
panel.poll()
|
||||
|
||||
warning.assert_called_once()
|
||||
self.assertIn("L’export va continuer", warning.call_args.args[1])
|
||||
self.assertEqual(continued, [True])
|
||||
@@ -38,6 +38,7 @@ class CompileToPdfTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(output.read_bytes(), b"partial")
|
||||
self.assertIn("Warning: LaTeX compilation reported errors", console.getvalue())
|
||||
self.assertIn(utils.LATEX_WARNING_MARKER + str(output.resolve()), console.getvalue())
|
||||
self.assertIn("a PDF was generated", console.getvalue())
|
||||
self.assertIn("missing \\item", console.getvalue())
|
||||
|
||||
@@ -69,6 +70,7 @@ class CompileToPdfTests(unittest.TestCase):
|
||||
|
||||
self.assertFalse(output.exists())
|
||||
self.assertIn("produced no PDF", console.getvalue())
|
||||
self.assertIn(utils.LATEX_WARNING_MARKER + str(output.resolve()), console.getvalue())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in new issue
Block a user