diff --git a/copienator_gui/__init__.py b/copienator_gui/__init__.py new file mode 100644 index 0000000..92a1179 --- /dev/null +++ b/copienator_gui/__init__.py @@ -0,0 +1,5 @@ +"""Interface graphique pour piloter le workflow Copienator.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/copienator_gui/__pycache__/__init__.cpython-314.pyc b/copienator_gui/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..95e00ea Binary files /dev/null and b/copienator_gui/__pycache__/__init__.cpython-314.pyc differ diff --git a/copienator_gui/__pycache__/app.cpython-314.pyc b/copienator_gui/__pycache__/app.cpython-314.pyc new file mode 100644 index 0000000..e70b05a Binary files /dev/null and b/copienator_gui/__pycache__/app.cpython-314.pyc differ diff --git a/copienator_gui/__pycache__/runner.cpython-314.pyc b/copienator_gui/__pycache__/runner.cpython-314.pyc new file mode 100644 index 0000000..2ae4ddc Binary files /dev/null and b/copienator_gui/__pycache__/runner.cpython-314.pyc differ diff --git a/copienator_gui/__pycache__/state.cpython-314.pyc b/copienator_gui/__pycache__/state.cpython-314.pyc new file mode 100644 index 0000000..be67cac Binary files /dev/null and b/copienator_gui/__pycache__/state.cpython-314.pyc differ diff --git a/copienator_gui/__pycache__/workflow.cpython-314.pyc b/copienator_gui/__pycache__/workflow.cpython-314.pyc new file mode 100644 index 0000000..5164e16 Binary files /dev/null and b/copienator_gui/__pycache__/workflow.cpython-314.pyc differ diff --git a/copienator_gui/app.py b/copienator_gui/app.py new file mode 100644 index 0000000..f7c7a07 --- /dev/null +++ b/copienator_gui/app.py @@ -0,0 +1,660 @@ +from __future__ import annotations + +import os +import queue +import re +import tkinter as tk +from datetime import datetime +from pathlib import Path +from tkinter import filedialog, messagebox, ttk +from typing import Any + +from .runner import ProcessRunner +from .state import StateStore +from .workflow import ( + CommandVariant, + StepDefinition, + build_command, + build_workflow, + command_display, + evaluation_argument, + value_for_default, +) + +STATUS_LABELS = { + "ready": "Prête", + "running": "En cours", + "success": "Réussie", + "failed": "Échouée", + "interrupted": "Interrompue", + "skipped": "Ignorée", + "stale": "À revalider", + "detected": "Détectée", +} + + +class CopienatorApp(tk.Tk): + def __init__( + self, + repository: Path, + show_personal_steps: bool, + initial_evaluation: Path | None = None, + ) -> None: + super().__init__() + self.repository = repository.resolve() + self.steps = build_workflow(show_personal_steps) + self.step_by_id = {step.id: step for step in self.steps} + self.state_store = StateStore() + self.runner = ProcessRunner() + self.active_step_id: str | None = None + self.current_step: StepDefinition | None = None + self.arg_vars: dict[str, tk.Variable] = {} + self._rendering = False + + self.title("Copienator — assistant de correction") + self.geometry("1180x820") + self.minsize(900, 640) + self.protocol("WM_DELETE_WINDOW", self._on_close) + + self.evaluation_var = tk.StringVar() + self.api_key_var = tk.StringVar(value=os.environ.get("GEMINI_API_KEY", "")) + self.proxy_var = tk.StringVar(value=os.environ.get("HTTPS_PROXY", "")) + self.variant_var = tk.StringVar() + self.extra_var = tk.StringVar() + self.command_var = tk.StringVar(value="Sélectionnez une étape.") + self.info_var = tk.StringVar(value="Choisissez un dossier d’évaluation.") + + self._build_ui(show_personal_steps) + self.extra_var.trace_add("write", lambda *_args: self._update_command_preview()) + self.after(60, self._poll_runner) + + if initial_evaluation and initial_evaluation.is_dir(): + self.evaluation_var.set(str(initial_evaluation.resolve())) + self._load_evaluation() + else: + self._populate_tree() + + @property + def evaluation(self) -> Path | None: + value = self.evaluation_var.get().strip() + return Path(value).expanduser().resolve() if value else None + + def _build_ui(self, show_personal_steps: bool) -> None: + self.columnconfigure(0, weight=1) + self.rowconfigure(1, weight=3) + self.rowconfigure(2, weight=2) + + top = ttk.Frame(self, padding=(10, 8)) + top.grid(row=0, column=0, sticky="ew") + top.columnconfigure(1, weight=1) + ttk.Label(top, text="Évaluation").grid(row=0, column=0, sticky="w", padx=(0, 8)) + path_entry = ttk.Entry(top, textvariable=self.evaluation_var) + path_entry.grid(row=0, column=1, sticky="ew") + path_entry.bind("", lambda _event: self._load_evaluation()) + ttk.Button(top, text="Parcourir…", command=self._browse_evaluation).grid(row=0, column=2, padx=6) + ttk.Button(top, text="Charger", command=self._load_evaluation).grid(row=0, column=3) + + 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( + row=1, column=1, sticky="w", pady=(7, 0) + ) + environment = ttk.Frame(top) + environment.grid(row=1, column=2, columnspan=2, sticky="e", pady=(7, 0)) + ttk.Label(environment, text="HTTPS_PROXY").pack(side="left", padx=(0, 6)) + ttk.Entry(environment, textvariable=self.proxy_var, width=28).pack(side="left") + profile = "standard + personnel" if show_personal_steps else "standard" + ttk.Label(environment, text=f"Profil : {profile}").pack(side="left", padx=(12, 0)) + + main_pane = ttk.Panedwindow(self, orient="horizontal") + main_pane.grid(row=1, column=0, sticky="nsew", padx=10) + + navigation = ttk.Frame(main_pane, padding=(0, 5, 6, 5)) + self.tree = ttk.Treeview(navigation, columns=("status",), show="tree headings", selectmode="browse") + self.tree.heading("#0", text="Étape") + self.tree.heading("status", text="État") + self.tree.column("#0", minwidth=230, width=280) + self.tree.column("status", minwidth=90, width=105, anchor="center") + tree_scroll = ttk.Scrollbar(navigation, orient="vertical", command=self.tree.yview) + self.tree.configure(yscrollcommand=tree_scroll.set) + self.tree.pack(side="left", fill="both", expand=True) + tree_scroll.pack(side="right", fill="y") + self.tree.bind("<>", self._on_tree_select) + main_pane.add(navigation, weight=1) + + self.detail = ttk.Frame(main_pane, padding=(12, 6, 4, 4)) + self.detail.columnconfigure(0, weight=1) + self.detail.rowconfigure(2, weight=1) + self.title_label = ttk.Label(self.detail, text="Sélectionnez une étape", font=("TkDefaultFont", 15, "bold")) + self.title_label.grid(row=0, column=0, sticky="w") + self.description_label = ttk.Label(self.detail, text="", wraplength=680, justify="left") + self.description_label.grid(row=1, column=0, sticky="ew", pady=(5, 8)) + self.form = ttk.Frame(self.detail) + self.form.grid(row=2, column=0, sticky="nsew") + self.form.columnconfigure(1, weight=1) + + command_box = ttk.LabelFrame(self.detail, text="Commande", padding=6) + command_box.grid(row=3, column=0, sticky="ew", pady=(8, 5)) + command_box.columnconfigure(0, weight=1) + ttk.Label(command_box, textvariable=self.command_var, wraplength=690, justify="left").grid( + row=0, column=0, sticky="ew" + ) + + buttons = ttk.Frame(self.detail) + buttons.grid(row=4, column=0, sticky="ew", pady=(5, 0)) + self.previous_button = ttk.Button(buttons, text="← Précédente", command=lambda: self._move_selection(-1)) + self.previous_button.pack(side="left") + self.next_button = ttk.Button(buttons, text="Suivante →", command=lambda: self._move_selection(1)) + self.next_button.pack(side="left", padx=6) + self.skip_button = ttk.Button(buttons, text="Marquer ignorée", command=self._skip_step) + self.skip_button.pack(side="right") + self.run_button = ttk.Button(buttons, text="Exécuter", command=self._run_current_step) + self.run_button.pack(side="right", padx=6) + main_pane.add(self.detail, weight=3) + + console_frame = ttk.LabelFrame(self, text="Sortie en temps réel", padding=(7, 5)) + console_frame.grid(row=2, column=0, sticky="nsew", padx=10, pady=(6, 8)) + console_frame.columnconfigure(0, weight=1) + console_frame.rowconfigure(0, weight=1) + self.console = tk.Text( + console_frame, + height=14, + wrap="word", + background="#171717", + foreground="#efefef", + insertbackground="white", + state="disabled", + ) + console_scroll = ttk.Scrollbar(console_frame, orient="vertical", command=self.console.yview) + self.console.configure(yscrollcommand=console_scroll.set) + self.console.grid(row=0, column=0, sticky="nsew") + console_scroll.grid(row=0, column=1, sticky="ns") + + console_actions = ttk.Frame(console_frame) + console_actions.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(5, 0)) + console_actions.columnconfigure(1, weight=1) + 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) + self.stdin_entry.grid(row=0, column=1, sticky="ew") + self.stdin_entry.bind("", lambda _event: self._send_input()) + self.send_button = ttk.Button(console_actions, text="Envoyer", command=self._send_input, state="disabled") + self.send_button.grid(row=0, column=2, padx=5) + self.interrupt_button = ttk.Button( + console_actions, text="Interrompre", command=self._interrupt, state="disabled" + ) + self.interrupt_button.grid(row=0, column=3, padx=5) + self.force_button = ttk.Button( + console_actions, text="Forcer l’arrêt", command=self._force_stop, state="disabled" + ) + self.force_button.grid(row=0, column=4) + ttk.Button(console_actions, text="Effacer la console", command=self._clear_console).grid( + row=0, column=5, padx=(8, 0) + ) + + status = ttk.Label(self, textvariable=self.info_var, anchor="w", relief="sunken", padding=(6, 3)) + status.grid(row=3, column=0, sticky="ew") + + def _browse_evaluation(self) -> None: + selected = filedialog.askdirectory(initialdir=self.evaluation_var.get() or self.repository) + if selected: + self.evaluation_var.set(selected) + self._load_evaluation() + + def _load_evaluation(self) -> None: + evaluation = self.evaluation + if not evaluation or not evaluation.is_dir(): + messagebox.showerror("Dossier invalide", "Choisissez un dossier d’évaluation existant.") + return + self._save_current_form() + self.state_store.load(evaluation) + for step in self.steps: + if self.state_store.step(step.id).get("status") == "running": + self.state_store.update_step(step.id, status="interrupted") + self._populate_tree() + missing = [name for name in ("enonce.pdf", "enonce.tex", "correction.tex") if not (evaluation / name).exists()] + names_found = (evaluation / "names").exists() or (self.repository / "names").exists() + details = [] + if missing: + details.append("manquants : " + ", ".join(missing)) + if not names_found: + details.append("fichier names introuvable") + if not self.api_key_var.get(): + details.append("clé Gemini non renseignée") + self.info_var.set( + f"Évaluation chargée : {evaluation}" + + (" — " + " ; ".join(details) if details else " — prérequis principaux détectés") + ) + first = self.steps[0].id if self.steps else None + if first: + self.tree.selection_set(first) + self.tree.see(first) + + def _populate_tree(self) -> None: + selected = self.current_step.id if self.current_step else None + self.tree.delete(*self.tree.get_children()) + section_items: dict[str, str] = {} + for step in self.steps: + if step.section not in section_items: + section_id = f"section:{len(section_items)}" + section_items[step.section] = section_id + self.tree.insert("", "end", iid=section_id, text=step.section, values=("",), open=True) + status = self._step_status(step) + suffix = " (facultative)" if step.optional else "" + self.tree.insert( + section_items[step.section], + "end", + iid=step.id, + text=step.title + suffix, + values=(STATUS_LABELS.get(status, status),), + ) + if selected and self.tree.exists(selected): + self.tree.selection_set(selected) + + def _step_status(self, step: StepDefinition) -> str: + if self.state_store.evaluation: + saved = self.state_store.step(step.id).get("status") + if saved: + return str(saved) + if self._artifacts_exist(step): + return "detected" + if step.id == "inputs" and not self._missing_requirements(step): + return "detected" + return "ready" + + def _artifacts_exist(self, step: StepDefinition) -> bool: + evaluation = self.evaluation + if not evaluation or not step.artifacts: + return False + for pattern in step.artifacts: + if any(char in pattern for char in "*?["): + if next(evaluation.glob(pattern), None) is not None: + return True + elif (evaluation / pattern).exists(): + return True + return False + + def _missing_requirements(self, step: StepDefinition) -> list[str]: + evaluation = self.evaluation + if not evaluation: + return list(step.requires) + missing = [] + for pattern in step.requires: + if any(char in pattern for char in "*?["): + exists = next(evaluation.glob(pattern), None) is not None + else: + exists = (evaluation / pattern).exists() + if not exists: + missing.append(pattern) + return missing + + def _on_tree_select(self, _event: tk.Event[Any] | None = None) -> None: + selection = self.tree.selection() + if not selection or selection[0].startswith("section:"): + return + step = self.step_by_id.get(selection[0]) + if not step: + return + if self.current_step and self.current_step.id != step.id: + self._save_current_form() + self.current_step = step + self._render_step() + + def _render_step(self) -> None: + step = self.current_step + if not step: + return + self._rendering = True + for child in self.form.winfo_children(): + child.destroy() + self.arg_vars.clear() + entry = self.state_store.step(step.id) if self.state_store.evaluation else {} + saved_variant = entry.get("variant", step.variants[0].id) + if saved_variant not in {variant.id for variant in step.variants}: + saved_variant = step.variants[0].id + self.variant_var.set(saved_variant) + self.extra_var.set(str(entry.get("extra", ""))) + self.title_label.configure(text=step.title) + missing = self._missing_requirements(step) + description = step.description + if missing: + description += "\nPrérequis non détectés : " + ", ".join(missing) + self.description_label.configure(text=description) + + row = 0 + if len(step.variants) > 1: + ttk.Label(self.form, text="Mode").grid(row=row, column=0, sticky="w", pady=4, padx=(0, 8)) + labels = [variant.label for variant in step.variants] + combo = ttk.Combobox(self.form, values=labels, state="readonly") + variant_index = [variant.id for variant in step.variants].index(saved_variant) + combo.current(variant_index) + combo.grid(row=row, column=1, sticky="ew", pady=4) + combo.bind("<>", lambda _event: self._select_variant(combo.current())) + row += 1 + + values = entry.get("values", {}) if isinstance(entry.get("values", {}), dict) else {} + variant = self._current_variant() + evaluation_arg = self._evaluation_arg() + for spec in step.arguments: + if spec.variants and variant.id not in spec.variants: + continue + value = values.get(spec.name, value_for_default(spec.default, evaluation_arg)) + ttk.Label(self.form, text=spec.label).grid(row=row, column=0, sticky="nw", pady=4, padx=(0, 8)) + if spec.kind == "bool": + variable: tk.Variable = tk.BooleanVar(value=bool(value)) + widget = ttk.Checkbutton(self.form, variable=variable) + widget.grid(row=row, column=1, sticky="w", pady=4) + elif spec.kind == "choice": + variable = tk.StringVar(value=str(value)) + widget = ttk.Combobox(self.form, textvariable=variable, values=spec.choices, state="readonly") + widget.grid(row=row, column=1, sticky="ew", pady=4) + else: + variable = tk.StringVar(value=str(value)) + field = ttk.Frame(self.form) + field.grid(row=row, column=1, sticky="ew", pady=4) + field.columnconfigure(0, weight=1) + widget = ttk.Entry(field, textvariable=variable) + widget.grid(row=0, column=0, sticky="ew") + if spec.kind == "path": + ttk.Button(field, text="Fichier…", command=lambda var=variable: self._browse_target_file(var)).grid( + row=0, column=1, padx=(5, 0) + ) + ttk.Button(field, text="Dossier…", command=lambda var=variable: self._browse_target_dir(var)).grid( + row=0, column=2, padx=(5, 0) + ) + self.arg_vars[spec.name] = variable + variable.trace_add("write", lambda *_args: self._update_command_preview()) + if spec.help: + ttk.Label(self.form, text=spec.help, foreground="#666666", wraplength=540).grid( + row=row + 1, column=1, sticky="w" + ) + row += 1 + row += 1 + + if not step.is_manual: + ttk.Label(self.form, text="Arguments supplémentaires").grid( + row=row, column=0, sticky="w", pady=(10, 4), padx=(0, 8) + ) + ttk.Entry(self.form, textvariable=self.extra_var).grid(row=row, column=1, sticky="ew", pady=(10, 4)) + + self.run_button.configure(text="Marquer terminée" if step.is_manual else "Exécuter") + self.skip_button.configure(state="normal" if step.optional else "disabled") + self._rendering = False + self._update_command_preview() + self._update_controls() + + def _select_variant(self, index: int) -> None: + if not self.current_step or index < 0: + return + self._save_current_form() + self.variant_var.set(self.current_step.variants[index].id) + if self.state_store.evaluation: + self.state_store.update_step(self.current_step.id, variant=self.variant_var.get()) + self._render_step() + + def _current_variant(self) -> CommandVariant: + assert self.current_step is not None + selected = self.variant_var.get() + return next((variant for variant in self.current_step.variants if variant.id == selected), self.current_step.variants[0]) + + def _values(self) -> dict[str, object]: + return {name: variable.get() for name, variable in self.arg_vars.items()} + + def _save_current_form(self) -> None: + if self._rendering or not self.current_step or not self.state_store.evaluation: + return + saved_values = self.state_store.step(self.current_step.id).get("values", {}) + merged_values = dict(saved_values) if isinstance(saved_values, dict) else {} + merged_values.update(self._values()) + self.state_store.update_step( + self.current_step.id, + variant=self.variant_var.get() or self.current_step.variants[0].id, + values=merged_values, + extra=self.extra_var.get(), + ) + + def _evaluation_arg(self) -> str: + evaluation = self.evaluation + return evaluation_argument(self.repository, evaluation) if evaluation else "<évaluation>" + + def _make_command(self) -> list[str]: + if not self.current_step: + return [] + return build_command( + self.repository, + self.current_step, + self._current_variant(), + self._values(), + self._evaluation_arg(), + self.extra_var.get(), + ) + + def _update_command_preview(self) -> None: + if self._rendering or not self.current_step: + return + if self.current_step.is_manual: + self.command_var.set("Étape manuelle — aucune commande ne sera exécutée.") + return + try: + self.command_var.set(command_display(self._make_command())) + except ValueError as exc: + self.command_var.set(f"Arguments invalides : {exc}") + + def _browse_target_file(self, variable: tk.Variable) -> None: + selected = filedialog.askopenfilename(initialdir=self.evaluation or self.repository) + if selected: + variable.set(selected) + + def _browse_target_dir(self, variable: tk.Variable) -> None: + selected = filedialog.askdirectory(initialdir=self.evaluation or self.repository) + if selected: + variable.set(selected) + + def _validate_arguments(self) -> bool: + if not self.current_step: + return False + variant = self._current_variant() + for spec in self.current_step.arguments: + if spec.variants and variant.id not in spec.variants: + continue + value = self.arg_vars.get(spec.name) + if spec.kind == "int" and value and str(value.get()).strip(): + try: + int(str(value.get())) + except ValueError: + messagebox.showerror("Argument invalide", f"« {spec.label} » doit être un entier.") + return False + try: + self._make_command() + except ValueError as exc: + messagebox.showerror("Arguments invalides", str(exc)) + return False + return True + + def _run_current_step(self) -> None: + step = self.current_step + evaluation = self.evaluation + if not step or not evaluation or not self.state_store.evaluation: + messagebox.showerror("Évaluation absente", "Chargez d’abord un dossier d’évaluation.") + return + if self.runner.running: + messagebox.showwarning("Traitement en cours", "Interrompez le traitement actuel avant d’en lancer un autre.") + return + if step.is_manual: + self._mark_step("success") + return + if not self._validate_arguments(): + return + missing = self._missing_requirements(step) + if missing and not messagebox.askyesno( + "Prérequis non détectés", + "Les éléments suivants n’ont pas été trouvés :\n\n" + + "\n".join(missing) + + "\n\nLancer tout de même la commande ?", + ): + return + variant = self._current_variant() + command = self._make_command() + if variant.dangerous and not messagebox.askyesno( + "Confirmation requise", + "Cette commande réinitialise ou supprime des résultats de correction. Continuer ?", + icon="warning", + ): + return + + self._save_current_form() + ordered_ids = [item.id for item in self.steps] + self.state_store.invalidate_after(ordered_ids, step.id) + self.state_store.update_step(step.id, status="running", command=command_display(command)) + self.active_step_id = step.id + timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") + safe_step = re.sub(r"[^A-Za-z0-9_.-]+", "_", step.id) + log_path = evaluation / ".copienator" / "logs" / f"{timestamp}-{safe_step}.log" + + environment = os.environ.copy() + environment["PYTHONUNBUFFERED"] = "1" + environment["PYTHONIOENCODING"] = "utf-8" + if self.api_key_var.get().strip(): + environment["GEMINI_API_KEY"] = self.api_key_var.get().strip() + if self.proxy_var.get().strip(): + environment["HTTPS_PROXY"] = self.proxy_var.get().strip() + + self._append_console(f"\n$ {command_display(command)}\n") + try: + self.runner.start(command, self.repository, environment, log_path) + except (OSError, RuntimeError) as exc: + self.state_store.update_step(step.id, status="failed") + self.state_store.add_history( + {"step": step.id, "command": command_display(command), "status": "failed", "error": str(exc)} + ) + self.active_step_id = None + self._append_console(f"Impossible de lancer la commande : {exc}\n") + messagebox.showerror("Échec du lancement", str(exc)) + self._populate_tree() + self._update_controls() + + def _mark_step(self, status: str) -> None: + if not self.current_step or not self.state_store.evaluation: + return + self._save_current_form() + self.state_store.invalidate_after([item.id for item in self.steps], self.current_step.id) + self.state_store.update_step(self.current_step.id, status=status) + self.state_store.add_history({"step": self.current_step.id, "status": status, "manual": True}) + self._populate_tree() + self.info_var.set(f"{self.current_step.title} : {STATUS_LABELS.get(status, status).lower()}.") + + def _skip_step(self) -> None: + if self.current_step and self.current_step.optional: + self._mark_step("skipped") + + def _poll_runner(self) -> None: + while True: + try: + event, payload = self.runner.events.get_nowait() + except queue.Empty: + break + if event == "output": + self._append_console(str(payload)) + elif event == "input_echo": + self._append_console(f"> {payload}") + elif event == "runner_error": + self._append_console(f"\nErreur du lanceur : {payload}\n") + elif event == "finished": + return_code, interrupted = payload + self._finish_process(int(return_code), bool(interrupted)) + self.after(60, self._poll_runner) + + def _finish_process(self, return_code: int, interrupted: bool) -> None: + step_id = self.active_step_id + if not step_id: + return + if interrupted: + status = "interrupted" + else: + status = "success" if return_code == 0 else "failed" + self.state_store.update_step(step_id, status=status, return_code=return_code) + self.state_store.add_history( + { + "step": step_id, + "command": self.state_store.step(step_id).get("command", ""), + "status": status, + "return_code": return_code, + } + ) + 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}).") + self.active_step_id = None + self._populate_tree() + self._update_controls() + + def _send_input(self) -> None: + text = self.stdin_var.get() + try: + self.runner.send_input(text) + except RuntimeError as exc: + messagebox.showinfo("Aucune saisie attendue", str(exc)) + return + self.stdin_var.set("") + + def _interrupt(self) -> None: + if self.runner.running and messagebox.askyesno( + "Interrompre", "Envoyer une interruption au script en cours ?" + ): + try: + self.runner.interrupt() + self.info_var.set("Interruption demandée. Utilisez « Forcer l’arrêt » si le script ne répond pas.") + except OSError as exc: + messagebox.showerror("Interruption impossible", str(exc)) + + def _force_stop(self) -> None: + if self.runner.running and messagebox.askyesno( + "Forcer l’arrêt", "Forcer l’arrêt du script et de ses processus enfants ?", icon="warning" + ): + try: + self.runner.force_stop() + except OSError as exc: + messagebox.showerror("Arrêt impossible", str(exc)) + + def _append_console(self, text: str) -> None: + self.console.configure(state="normal") + self.console.insert("end", text) + self.console.see("end") + self.console.configure(state="disabled") + + def _clear_console(self) -> None: + self.console.configure(state="normal") + self.console.delete("1.0", "end") + self.console.configure(state="disabled") + + def _update_controls(self) -> None: + running = self.runner.running + self.run_button.configure(state="disabled" if running or not self.current_step else "normal") + self.interrupt_button.configure(state="normal" if running else "disabled") + self.force_button.configure(state="normal" if running else "disabled") + self.send_button.configure(state="normal" if running else "disabled") + self.stdin_entry.configure(state="normal" if running else "disabled") + + def _move_selection(self, delta: int) -> None: + if not self.current_step: + return + ids = [step.id for step in self.steps] + try: + index = ids.index(self.current_step.id) + except ValueError: + return + target = max(0, min(len(ids) - 1, index + delta)) + self.tree.selection_set(ids[target]) + self.tree.see(ids[target]) + + def _on_close(self) -> None: + if self.runner.running: + if not messagebox.askyesno( + "Traitement en cours", "Un script est encore actif. Le forcer à s’arrêter et fermer l’application ?" + ): + return + try: + self.runner.force_stop() + except OSError: + pass + self._save_current_form() + self.destroy() diff --git a/copienator_gui/runner.py b/copienator_gui/runner.py new file mode 100644 index 0000000..0b19e90 --- /dev/null +++ b/copienator_gui/runner.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import os +import queue +import signal +import subprocess +import threading +from pathlib import Path +from typing import Any + + +class ProcessRunner: + """Run one subprocess and expose thread-safe events for the Tk loop.""" + + def __init__(self) -> None: + self.events: queue.Queue[tuple[str, Any]] = queue.Queue() + self.process: subprocess.Popen[bytes] | None = None + self._interrupted = False + self._log_file = None + + @property + def running(self) -> bool: + return self.process is not None and self.process.poll() is None + + def start( + self, + command: list[str], + cwd: Path, + environment: dict[str, str], + log_path: Path, + ) -> None: + if self.running: + raise RuntimeError("Un processus est déjà en cours") + log_path.parent.mkdir(parents=True, exist_ok=True) + self._log_file = log_path.open("wb") + self._interrupted = False + + kwargs: dict[str, Any] = {} + if os.name == "nt": + kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + else: + kwargs["start_new_session"] = True + + self.process = subprocess.Popen( + command, + cwd=cwd, + env=environment, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + bufsize=0, + **kwargs, + ) + thread = threading.Thread(target=self._read_process, daemon=True) + thread.start() + + def _read_process(self) -> None: + process = self.process + if process is None or process.stdout is None: + return + return_code = -1 + try: + while True: + chunk = process.stdout.read(4096) + if not chunk: + break + if self._log_file: + self._log_file.write(chunk) + self._log_file.flush() + self.events.put(("output", chunk.decode("utf-8", errors="replace"))) + return_code = process.wait() + except (OSError, ValueError) as exc: # pragma: no cover - defensive reporting + self.events.put(("runner_error", str(exc))) + return_code = process.wait() + finally: + if self._log_file: + self._log_file.close() + self._log_file = None + process.stdout.close() + if process.stdin: + process.stdin.close() + self.events.put(("finished", (return_code, self._interrupted))) + + def send_input(self, text: str) -> None: + if not self.running or not self.process or not self.process.stdin: + raise RuntimeError("Aucun processus n’attend de saisie") + data = (text + "\n").encode("utf-8") + self.process.stdin.write(data) + self.process.stdin.flush() + self.events.put(("input_echo", text + "\n")) + + def interrupt(self) -> None: + if not self.running or not self.process: + return + self._interrupted = True + if os.name == "nt": + self.process.send_signal(signal.CTRL_BREAK_EVENT) + else: + os.killpg(self.process.pid, signal.SIGINT) + + def force_stop(self) -> None: + if not self.running or not self.process: + return + self._interrupted = True + if os.name == "nt": + subprocess.run( + ["taskkill", "/PID", str(self.process.pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + else: + os.killpg(self.process.pid, signal.SIGKILL) diff --git a/copienator_gui/state.py b/copienator_gui/state.py new file mode 100644 index 0000000..b2f1b04 --- /dev/null +++ b/copienator_gui/state.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +STATE_FILENAME = ".copienator-gui.json" + + +class StateStore: + def __init__(self) -> None: + self.evaluation: Path | None = None + self.data: dict[str, Any] = self._empty_data() + + @staticmethod + def _empty_data() -> dict[str, Any]: + return {"version": 1, "steps": {}, "history": []} + + def load(self, evaluation: Path) -> None: + self.evaluation = evaluation + path = evaluation / STATE_FILENAME + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + self.data = loaded if isinstance(loaded, dict) else self._empty_data() + except (OSError, json.JSONDecodeError): + self.data = self._empty_data() + self.data.setdefault("version", 1) + self.data.setdefault("steps", {}) + self.data.setdefault("history", []) + + def save(self) -> None: + if not self.evaluation: + return + path = self.evaluation / STATE_FILENAME + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text( + json.dumps(self.data, indent=2, ensure_ascii=False), encoding="utf-8" + ) + temporary.replace(path) + + def step(self, step_id: str) -> dict[str, Any]: + return self.data["steps"].setdefault(step_id, {}) + + def update_step(self, step_id: str, **changes: Any) -> None: + entry = self.step(step_id) + entry.update(changes) + entry["updated_at"] = datetime.now(timezone.utc).isoformat() + self.save() + + def add_history(self, entry: dict[str, Any]) -> None: + entry = dict(entry) + entry["timestamp"] = datetime.now(timezone.utc).isoformat() + self.data["history"].append(entry) + self.data["history"] = self.data["history"][-200:] + self.save() + + def invalidate_after(self, ordered_ids: list[str], current_id: str) -> None: + try: + index = ordered_ids.index(current_id) + except ValueError: + return + for step_id in ordered_ids[index + 1 :]: + entry = self.step(step_id) + if entry.get("status") in {"success", "detected", "skipped"}: + entry["status"] = "stale" + self.save() diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py new file mode 100644 index 0000000..50d7e6e --- /dev/null +++ b/copienator_gui/workflow.py @@ -0,0 +1,485 @@ +from __future__ import annotations + +import os +import shlex +import sys +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path + +EVALUATION = "${evaluation}" + + +@dataclass(frozen=True) +class ArgumentSpec: + name: str + label: str + kind: str = "text" # text, int, bool, choice, path + flag: str | None = None + default: object = "" + choices: tuple[str, ...] = () + help: str = "" + positional: bool = False + variants: tuple[str, ...] = () + + +@dataclass(frozen=True) +class CommandVariant: + id: str + label: str + program: str | None + kind: str = "python" # python, shell, external, manual + fixed_args: tuple[str, ...] = () + dangerous: bool = False + + +@dataclass(frozen=True) +class StepDefinition: + id: str + section: str + title: str + description: str + variants: tuple[CommandVariant, ...] + arguments: tuple[ArgumentSpec, ...] = () + optional: bool = False + personal: bool = False + requires: tuple[str, ...] = () + artifacts: tuple[str, ...] = () + + @property + def is_manual(self) -> bool: + return all(variant.kind == "manual" for variant in self.variants) + + +def arg_target(help_text: str = "Dossier d’évaluation ou fichier à traiter") -> ArgumentSpec: + return ArgumentSpec( + "target", + "Cible", + kind="path", + default=EVALUATION, + positional=True, + help=help_text, + ) + + +def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: + python = lambda ident, label, script, **kwargs: CommandVariant( + ident, label, script, "python", **kwargs + ) + manual = lambda ident, label="Étape manuelle": CommandVariant( + ident, label, None, "manual" + ) + + steps = [ + StepDefinition( + "inputs", + "Préparation", + "Vérifier les fichiers d’entrée", + "Le dossier doit contenir enonce.pdf, enonce.tex et correction.tex. " + "Le fichier names doit être présent dans l’évaluation ou à la racine du projet.", + (manual("check"),), + requires=("enonce.pdf", "enonce.tex", "correction.tex"), + ), + StepDefinition( + "statement", + "Prétraitement de l’énoncé", + "Analyser l’énoncé", + "Détecte les questions, leurs groupes, le corrigé et les indications de barème.", + ( + python("gemini", "Analyse avec Gemini", "gemini_for_enonce.py"), + python("personal", "Alternative enonce_info.py", "enonce_info.py"), + ), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec( + "restart", + "Ignorer le cache (--restart)", + kind="bool", + flag="--restart", + variants=("gemini",), + ), + ), + requires=("enonce.pdf", "enonce.tex", "correction.tex"), + artifacts=("labels", "Text", "Sol", "Persp"), + ), + StepDefinition( + "review_persp", + "Prétraitement de l’énoncé", + "Relire les barèmes dans Persp", + "Étape manuelle facultative : vérifier et modifier les instructions de correction.", + (manual("review"),), + optional=True, + requires=("Persp",), + ), + StepDefinition( + "rotate", + "Prétraitement des copies", + "Retourner toutes les copies", + "Rotation facultative de 180° lorsque les scans sont à l’envers.", + (CommandVariant("rotate", "Rotation", "rotate_all.sh", "shell"),), + arguments=(arg_target("Dossier de l’évaluation"),), + optional=True, + ), + StepDefinition( + "rename", + "Prétraitement des copies", + "Renommer les PDF en CopieXX.pdf", + "Renomme les PDF du dossier d’évaluation dans leur ordre courant.", + (CommandVariant("rename", "Renommage", "rename_to_copie.sh", "shell"),), + arguments=(arg_target("Dossier de l’évaluation"),), + ), + StepDefinition( + "page_splitter", + "Prétraitement des copies", + "Séparer et réordonner les pages", + "Ouvre l’outil interactif de découpage A3 vers A4. La cible peut être un dossier ou un PDF.", + (python("default", "Séparation des pages", "page_splitter.py"),), + arguments=(arg_target(),), + artifacts=("Copies", "Copies Originales"), + ), + StepDefinition( + "cutleft", + "Prétraitement des copies", + "Découper la marge des labels", + "Produit les images de la partie gauche des copies. Une copie précise peut être ciblée.", + (python("default", "Découpe", "cutleft.py"),), + arguments=( + arg_target(), + ArgumentSpec("fullpage", "Toujours utiliser la page entière", "bool", "--fullpage"), + ), + requires=("Copies",), + artifacts=("Cutleft",), + ), + StepDefinition( + "labels", + "Labels et regroupement", + "Détecter les labels avec Gemini", + "Identifie les labels dans les images produites par la découpe gauche.", + (python("default", "Détection des labels", "gemini_for_labels.py"),), + arguments=( + arg_target(), + ArgumentSpec("overwrite", "Régénérer les résultats", "bool", "--overwrite"), + ), + requires=("labels", "Copies", "Cutleft"), + artifacts=("Copies/*.json",), + ), + StepDefinition( + "plotting", + "Labels et regroupement", + "Vérifier visuellement les labels", + "Ouvre la fenêtre de vérification. La cible peut être toute l’évaluation ou une copie.", + (python("default", "Vérification", "plotting.py"),), + arguments=(arg_target(),), + requires=("labels", "Cutleft"), + artifacts=("Copies/Copie*.json",), + ), + StepDefinition( + "splitting", + "Labels et regroupement", + "Découper les réponses par question", + "Découpe les copies à partir des coordonnées de labels vérifiées.", + (python("default", "Découpage", "splitting_int.py"),), + arguments=(arg_target(),), + requires=("Copies",), + artifacts=("Copies/Copie*/*",), + ), + StepDefinition( + "grouping", + "Labels et regroupement", + "Regrouper les réponses", + "Regroupe les réponses portant le même label pour préparer les requêtes.", + (python("default", "Regroupement", "grouping.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + requires=("Copies",), + artifacts=("Par label",), + ), + StepDefinition( + "correction", + "Correction", + "Lancer ou intégrer la correction", + "Choisir une correction immédiate, batch, hybride, une recorrection, ou l’intégration d’un batch.", + ( + python("live", "Correction immédiate", "correction.py"), + python("batch", "Préparer toutes les requêtes batch", "correction.py", fixed_args=("--batch",)), + python("hybrid", "Batch à partir d’un label", "correction.py"), + python("refaire", "Recorrection depuis refaire.json", "correction.py", fixed_args=("--refaire",)), + python( + "integrate", + "Intégrer les résultats batch", + "correction.py", + fixed_args=("--deal-with-batched",), + ), + python("reset", "Réinitialiser les corrections", "correction.py", fixed_args=("--reset",), dangerous=True), + ), + arguments=( + arg_target("Évaluation ou image Group_X.jpg"), + ArgumentSpec("overwrite", "Écraser les corrections existantes", "bool", "--overwrite", variants=("live",)), + ArgumentSpec("limit", "Limite d’appels Pro", "int", "--limit", variants=("live",)), + ArgumentSpec("batch_from", "Premier label envoyé en batch", "text", "--batch-from", variants=("hybrid",)), + ), + requires=("Par label", "Persp", "labels"), + artifacts=("correction.json", "batch_requests_*.jsonl"), + ), + StepDefinition( + "submit_batches", + "Correction", + "Envoyer les batchs", + "Envoie à Gemini les fichiers JSONL produits par le mode batch.", + (python("default", "Envoi", "submit_batches.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + optional=True, + artifacts=("batch_jobs.json",), + ), + StepDefinition( + "batch_status", + "Correction", + "Consulter l’état des batchs", + "Affiche les jobs Gemini en cours. L’identifiant de téléchargement est facultatif.", + (python("default", "État des batchs", "batch_status.py"),), + arguments=(ArgumentSpec("download", "Télécharger le job", "text", "--download"),), + optional=True, + ), + StepDefinition( + "fetch_batches", + "Correction", + "Récupérer les résultats batch", + "Télécharge et rassemble les réponses des jobs terminés.", + (python("default", "Récupération", "fetch_batched_results.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + optional=True, + ), + StepDefinition( + "post_correction", + "Correction", + "Nettoyer la correction", + "Corrige certains problèmes d’encodage et prépare le texte pour LaTeX.", + (python("default", "Post-correction", "post-correction.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + requires=("correction.json",), + ), + StepDefinition( + "manual_resolution", + "Correction", + "Résoudre les conflits manuels", + "Étape conditionnelle, uniquement si manual_resolutions.txt contient des conflits à traiter.", + (python("default", "Résolution", "resolve_manual.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + optional=True, + requires=("manual_resolutions.txt", "correction.json"), + ), + StepDefinition( + "annotation", + "Génération des annotations", + "Générer les copies annotées", + "Les trois modes sont exclusifs pour un parcours donné.", + ( + python("simple", "Annotations simples (Anot)", "annotating.py"), + python("checks", "Annotations avec cases (Bnot)", "annotating_with_checks.py"), + python("grouped", "Annotations groupées (BGnot)", "annotating_by_label.py"), + ), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec("overwrite", "Écraser les sorties", "bool", "--overwrite"), + ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire", variants=("checks",)), + ), + requires=("correction.json",), + artifacts=("Anot", "Bnot", "BGnot"), + ), + StepDefinition( + "export", + "Génération des annotations", + "Exporter vers la tablette", + "Exporte les groupes vers le dossier EXPORT_DIR défini dans config.py.", + (python("default", "Export", "export.py"),), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire"), + ), + optional=True, + ), + StepDefinition( + "tablet", + "Correction manuscrite", + "Annoter les PDF sur la tablette", + "Étape manuelle : enregistrer chaque résultat sous le nom Concat_annotated.pdf.", + (manual("tablet"),), + ), + StepDefinition( + "import", + "Correction manuscrite", + "Importer les annotations manuscrites", + "Copie les PDF présents dans IMPORT_DIR vers l’évaluation.", + (python("default", "Import", "import.py"),), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire"), + ), + ), + StepDefinition( + "read_annotations", + "Finalisation", + "Lire les annotations manuscrites", + "Le mode doit correspondre au mode choisi lors de la génération des annotations.", + ( + python("standard", "Lecture Bnot", "reading_annotations.py"), + python("grouped", "Lecture BGnot", "reading_grouped_annotations.py"), + ), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec("update_score", "Réappliquer les score.json", "bool", "--update-score"), + ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire", variants=("grouped",)), + ), + ), + StepDefinition( + "giving_names", + "Finalisation", + "Attribuer les noms et préparer A Rendre", + "Crée le dossier A Rendre à partir du dossier d’annotations choisi.", + (python("default", "Attribution des noms", "giving_names.py"),), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec( + "annotation_dir", + "Dossier d’annotations", + "choice", + default="BGnot", + choices=("BGnot", "Bnot", "Anot"), + positional=True, + ), + ), + artifacts=("A Rendre",), + ), + StepDefinition( + "personal_create", + "Étapes personnelles", + "Créer l’interro dans gestion_classe", + "Exécute gestion_classe ne.", + (CommandVariant("default", "gestion_classe ne", "gestion_classe", "external", ("ne",)),), + personal=True, + ), + StepDefinition( + "personal_scale", + "Étapes personnelles", + "Renseigner le barème", + "Exécute gestion_classe we.", + (CommandVariant("default", "gestion_classe we", "gestion_classe", "external", ("we",)),), + personal=True, + ), + StepDefinition( + "update_ods", + "Étapes personnelles", + "Mettre à jour le fichier ODS", + "Transfère les scores, avec la possibilité de n’écrire que leur somme.", + (python("default", "Mise à jour ODS", "update_ods.py"),), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec("sum", "Écrire seulement la somme", "bool", "--sum"), + ), + personal=True, + ), + StepDefinition( + "personal_read", + "Étapes personnelles", + "Relire avec gestion_classe", + "Exécute gestion_classe re.", + (CommandVariant("default", "gestion_classe re", "gestion_classe", "external", ("re",)),), + personal=True, + ), + StepDefinition( + "personal_sent", + "Étapes personnelles", + "Marquer comme envoyé", + "Exécute gestion_classe wsent.", + (CommandVariant("default", "gestion_classe wsent", "gestion_classe", "external", ("wsent",)),), + personal=True, + ), + StepDefinition( + "final_score", + "Étapes personnelles", + "Ajouter le score final", + "Génère les fichiers de diffusion avec le score final.", + (python("default", "Score final", "add_final_score.py"),), + arguments=(arg_target("Dossier de l’évaluation"),), + personal=True, + ), + 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.", + (manual("deploy"),), + personal=True, + ), + StepDefinition( + "personal_print", + "Étapes personnelles", + "Imprimer une copie", + "Étape manuelle via le lecteur PDF.", + (manual("print"),), + personal=True, + optional=True, + ), + ] + return [step for step in steps if show_personal_steps or not step.personal] + + +def value_for_default(value: object, evaluation_arg: str) -> object: + return evaluation_arg if value == EVALUATION else value + + +def build_command( + repository: Path, + step: StepDefinition, + variant: CommandVariant, + values: dict[str, object], + evaluation_arg: str, + extra_arguments: str = "", +) -> list[str]: + if variant.kind == "manual" or not variant.program: + return [] + + program_path = repository / variant.program + if variant.kind == "python": + command = [sys.executable, "-u", str(program_path)] + elif variant.kind == "shell": + command = [str(program_path)] + else: + command = [variant.program] + + positionals: list[str] = [] + options: list[str] = [] + for spec in step.arguments: + if spec.variants and variant.id not in spec.variants: + continue + value = values.get(spec.name, value_for_default(spec.default, evaluation_arg)) + if spec.kind == "bool": + if bool(value) and spec.flag: + options.append(spec.flag) + continue + if value is None or str(value).strip() == "": + continue + rendered = str(value) + if spec.positional: + positionals.append(rendered) + elif spec.flag: + options.extend((spec.flag, rendered)) + + command.extend(positionals) + command.extend(variant.fixed_args) + command.extend(options) + if extra_arguments.strip(): + command.extend(shlex.split(extra_arguments, posix=os.name != "nt")) + return command + + +def command_display(command: Iterable[str]) -> str: + if os.name == "nt": + return " ".join(f'"{part}"' if " " in part else part for part in command) + return shlex.join(command) + + +def evaluation_argument(repository: Path, evaluation: Path) -> str: + try: + return str(evaluation.resolve().relative_to(repository.resolve())) + except ValueError: + return str(evaluation.resolve()) diff --git a/gui.py b/gui.py new file mode 100644 index 0000000..17cde64 --- /dev/null +++ b/gui.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import argparse +from pathlib import Path + +from copienator_gui.app import CopienatorApp + + +def personal_steps_enabled() -> bool: + try: + from config import SHOW_PERSONAL_STEPS + except (ImportError, AttributeError): + try: + from default_config import SHOW_PERSONAL_STEPS + except (ImportError, AttributeError): + return False + return bool(SHOW_PERSONAL_STEPS) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Interface graphique du workflow Copienator") + parser.add_argument("evaluation", nargs="?", type=Path, help="Dossier d’évaluation à ouvrir") + args = parser.parse_args() + + repository = Path(__file__).resolve().parent + evaluation = args.evaluation.resolve() if args.evaluation else None + app = CopienatorApp(repository, personal_steps_enabled(), evaluation) + app.mainloop() + + +if __name__ == "__main__": + main()