diff --git a/Script.org b/Script.org index 0ace07d..f7ebe47 100644 --- a/Script.org +++ b/Script.org @@ -9,7 +9,7 @@ les parcours alternatifs. - [[file:Readme.org][Guide de démarrage]] - [[file:Architecture.org][Architecture et conventions de développement]] -* Étapes et Script +* Étapes Utiliser `python -m copienator gui` ou `python -m copienator gui Interro` pour lancer un GUI qui suit automatiquement les étapes décrites ci-dessous. @@ -314,6 +314,37 @@ OU + update the copies from =miqmacs.fr/admin=. 6. (gestion perso) Impression d'une copie. Via Evince » print to pdf. +** Archivage et nettoyage + +Une fois l'évaluation terminée, =python -m copienator clean Interro= +supprime les fichiers intermédiaires et régénérables. La commande ne +conserve que : + + + les PDF =Copies/*.pdf= produits après le découpage des pages ; + + les sources et sorties textuelles du prétraitement de l'énoncé : + =enonce.tex=, =correction.tex=, =labels=, =label_groups=, =Text=, + =Sol=, =Persp=, les fichiers TeX de =Text2= et =Sol2=, =Cache= et + =Tmp= ; + + le résultat final =correction.json= ; + + les journaux de =.copienator/logs= et les journaux placés à la + racine, comme =correction_log= ; + + les images et fichiers =score.json= présents dans =A Rendre=. + +Les liens symboliques conservés dans =A Rendre= sont remplacés par de +véritables fichiers avant la suppression de leurs cibles. La commande +refuse de démarrer si les copies traitées, =correction.json= ou une +image/un score d'élève sont absents. Elle affiche d'abord un résumé et +demande de saisir le nom de l'évaluation pour confirmer. + +Dans le GUI, cette commande apparaît comme dernière étape facultative +dans la section =Archivage=. Un avertissement rappelle que la +progression du GUI et les données binaires permettant de reprendre les +étapes seront définitivement perdues. + +Utiliser =python -m copienator clean Interro --dry-run= pour afficher +le plan sans rien supprimer, et =--yes= pour omettre la confirmation +interactive. + * Autres ** Recorrection d'une seule copie (peu testé) diff --git a/copienator/commands/clean.py b/copienator/commands/clean.py new file mode 100644 index 0000000..161ba79 --- /dev/null +++ b/copienator/commands/clean.py @@ -0,0 +1,329 @@ +from __future__ import annotations + +import argparse +import os +import shutil +import uuid +from collections import Counter +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from pathlib import Path + +from copienator import ( + CliError, + EvaluationWorkspace, + ExitCode, + evaluation_parser, + execute, + workspace_from_args, +) + + +IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".webp"} +STATEMENT_ROOT_FILES = {"enonce.tex", "correction.tex", "labels", "label_groups"} +STATEMENT_TEXT_DIRECTORIES = {"Text", "Sol", "Persp", "Cache", "Tmp"} + + +@dataclass(frozen=True, slots=True) +class CleanupPlan: + kept_files: tuple[Path, ...] + deleted_files: tuple[Path, ...] + deleted_directories: tuple[Path, ...] + bytes_to_delete: int + + +def _workspace_entries(root: Path) -> tuple[list[Path], list[Path]]: + """List files/symlinks and real directories without following symlinks.""" + files: list[Path] = [] + directories: list[Path] = [] + for current, directory_names, file_names in os.walk( + root, topdown=True, followlinks=False + ): + current_path = Path(current) + traversable: list[str] = [] + for name in directory_names: + path = current_path / name + if path.is_symlink(): + files.append(path) + else: + directories.append(path) + traversable.append(name) + directory_names[:] = traversable + files.extend(current_path / name for name in file_names) + return files, directories + + +def _return_artifacts(workspace: EvaluationWorkspace) -> list[Path]: + return_dir = workspace.return_dir + if not return_dir.is_dir() or return_dir.is_symlink(): + raise CliError( + f"Return directory not found or invalid: {return_dir}", + ExitCode.INVALID_WORKSPACE, + ) + + student_directories = sorted( + ( + path + for path in return_dir.iterdir() + if path.is_dir() and not path.is_symlink() + ), + key=lambda path: path.name.casefold(), + ) + if not student_directories: + raise CliError( + f"No student directories found in {return_dir}", + ExitCode.INVALID_WORKSPACE, + ) + + artifacts: list[Path] = [] + incomplete: list[str] = [] + for directory in student_directories: + files = [path for path in directory.rglob("*") if path.is_file()] + images = [ + path for path in files if path.suffix.casefold() in IMAGE_SUFFIXES + ] + scores = [path for path in files if path.name.casefold() == "score.json"] + if not images or not scores: + missing = [] + if not images: + missing.append("image") + if not scores: + missing.append("score.json") + incomplete.append(f"{directory.name} ({', '.join(missing)})") + artifacts.extend(images) + artifacts.extend(scores) + + if incomplete: + details = "\n".join(f" - {item}" for item in incomplete) + raise CliError( + "A Rendre is incomplete; cleanup was refused:\n" + details, + ExitCode.INVALID_WORKSPACE, + ) + return artifacts + + +def _is_statement_text_file(workspace: EvaluationWorkspace, path: Path) -> bool: + relative = path.relative_to(workspace.root) + if len(relative.parts) == 1: + return relative.name in STATEMENT_ROOT_FILES + top_level = relative.parts[0] + if top_level in STATEMENT_TEXT_DIRECTORIES: + return path.suffix.casefold() in {"", ".json", ".tex", ".txt"} + if top_level in {"Text2", "Sol2"}: + return path.suffix.casefold() == ".tex" + return False + + +def _is_log_file(workspace: EvaluationWorkspace, path: Path) -> bool: + if path.is_relative_to(workspace.logs_dir): + return True + relative = path.relative_to(workspace.root) + return len(relative.parts) == 1 and ( + path.suffix.casefold() == ".log" + or path.name.casefold().endswith("_log") + ) + + +def build_cleanup_plan(workspace: EvaluationWorkspace) -> CleanupPlan: + processed_copies = sorted( + ( + path + for path in workspace.copies_dir.glob("*.pdf") + if path.is_file() + ), + key=lambda path: path.name.casefold(), + ) + if not processed_copies: + raise CliError( + f"No processed PDF copies found in {workspace.copies_dir}", + ExitCode.INVALID_WORKSPACE, + ) + if not workspace.correction_file.is_file(): + raise CliError( + f"Correction result not found: {workspace.correction_file}", + ExitCode.INVALID_WORKSPACE, + ) + + files, directories = _workspace_entries(workspace.root) + + kept_files = set(processed_copies) + kept_files.add(workspace.correction_file) + kept_files.update(_return_artifacts(workspace)) + kept_files.update( + path + for path in files + if _is_statement_text_file(workspace, path) + or _is_log_file(workspace, path) + ) + + kept_directories = {workspace.root} + for path in kept_files: + kept_directories.update( + parent + for parent in path.parents + if parent == workspace.root or workspace.root in parent.parents + ) + + deleted_files = tuple(sorted(set(files) - kept_files, key=str)) + deleted_directories = tuple( + sorted( + set(directories) - kept_directories, + key=lambda path: (len(path.parts), str(path)), + reverse=True, + ) + ) + bytes_to_delete = sum( + path.stat(follow_symlinks=False).st_size + for path in deleted_files + if path.exists() or path.is_symlink() + ) + return CleanupPlan( + kept_files=tuple(sorted(kept_files, key=str)), + deleted_files=deleted_files, + deleted_directories=deleted_directories, + bytes_to_delete=bytes_to_delete, + ) + + +def _human_size(size: int) -> str: + value = float(size) + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if value < 1024 or unit == "TiB": + return f"{value:.1f} {unit}" + value /= 1024 + return f"{size} B" + + +def _top_level_counts( + workspace: EvaluationWorkspace, paths: Iterable[Path] +) -> Counter[str]: + counts: Counter[str] = Counter() + for path in paths: + relative = path.relative_to(workspace.root) + counts[relative.parts[0]] += 1 + return counts + + +def print_plan(workspace: EvaluationWorkspace, plan: CleanupPlan, *, verbose: bool) -> None: + processed_count = sum(path.parent == workspace.copies_dir for path in plan.kept_files) + return_count = sum( + path.is_relative_to(workspace.return_dir) for path in plan.kept_files + ) + statement_count = sum( + _is_statement_text_file(workspace, path) for path in plan.kept_files + ) + log_count = sum(_is_log_file(workspace, path) for path in plan.kept_files) + print(f"Evaluation: {workspace.root}") + print("Will keep:") + print(f" - {processed_count} processed PDF copies in Copies") + print(f" - {statement_count} textual statement files") + print(" - correction.json") + print(f" - {log_count} log files") + print(f" - {return_count} image/score artifacts in A Rendre") + print( + f"Will delete {len(plan.deleted_files)} files and " + f"{len(plan.deleted_directories)} directories " + f"({_human_size(plan.bytes_to_delete)} in file entries)." + ) + counts = _top_level_counts(workspace, plan.deleted_files) + if counts: + print("Files removed by top-level location:") + for name, count in sorted(counts.items(), key=lambda item: item[0].casefold()): + print(f" - {name}: {count}") + if verbose: + print("Deletion list:") + for path in plan.deleted_files: + print(f" - {path.relative_to(workspace.root)}") + + +def _materialize_return_links(workspace: EvaluationWorkspace, paths: Iterable[Path]) -> None: + for path in paths: + if not path.is_relative_to(workspace.return_dir) or not path.is_symlink(): + continue + try: + source = path.resolve(strict=True) + except OSError as exc: + raise CliError(f"Broken return link {path}: {exc}") from exc + if not source.is_file(): + raise CliError(f"Return link does not target a file: {path} -> {source}") + temporary = path.with_name(f".{path.name}.materialize-{uuid.uuid4().hex}.tmp") + try: + shutil.copy2(source, temporary) + temporary.replace(path) + finally: + temporary.unlink(missing_ok=True) + print(f"Materialized: {path.relative_to(workspace.root)}") + + +def apply_cleanup(workspace: EvaluationWorkspace, plan: CleanupPlan) -> None: + _materialize_return_links(workspace, plan.kept_files) + for path in plan.deleted_files: + path.unlink(missing_ok=True) + for path in plan.deleted_directories: + try: + path.rmdir() + except FileNotFoundError: + pass + print( + f"Cleanup complete: deleted {len(plan.deleted_files)} files and " + f"{len(plan.deleted_directories)} directories." + ) + + +def run( + workspace: EvaluationWorkspace, + *, + dry_run: bool = False, + assume_yes: bool = False, + verbose: bool = False, +) -> ExitCode: + plan = build_cleanup_plan(workspace) + print_plan(workspace, plan, verbose=verbose) + if dry_run: + print("Dry run: nothing was deleted.") + return ExitCode.SUCCESS + if not assume_yes: + expected = workspace.name + answer = input( + f"This cannot be undone. Type {expected!r} to confirm cleanup: " + ).strip() + if answer != expected: + print("Cleanup cancelled; nothing was deleted.") + return ExitCode.SUCCESS + apply_cleanup(workspace, plan) + return ExitCode.SUCCESS + + +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser( + "Archive an evaluation by deleting regenerable and intermediate files." + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be kept and deleted without changing anything", + ) + parser.add_argument( + "--yes", + action="store_true", + help="Skip the interactive evaluation-name confirmation", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + + def handle(args: argparse.Namespace) -> ExitCode: + return run( + workspace_from_args(args), + dry_run=args.dry_run, + assume_yes=args.yes, + verbose=args.verbose, + ) + + return execute(parser, argv, handle) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/copienator/dispatcher.py b/copienator/dispatcher.py index c9ca9df..5c69357 100644 --- a/copienator/dispatcher.py +++ b/copienator/dispatcher.py @@ -39,6 +39,7 @@ COMMANDS: dict[str, Command] = { "giving-names": Command("giving_names", "Name copies and prepare A Rendre"), "update-ods": Command("update_ods", "Update the configured score spreadsheet"), "add-final-score": Command("add_final_score", "Stamp final scores on copies"), + "clean": Command("clean", "Delete intermediate files from a finished evaluation"), "gui": Command("@gui", "Launch the graphical workflow assistant"), } diff --git a/copienator_gui/app.py b/copienator_gui/app.py index 4ff6b99..aef1736 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -802,8 +802,9 @@ class CopienatorApp(tk.Tk): 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 ?", + "Nettoyage irréversible" if step.id == "clean" else "Confirmation requise", + variant.danger_warning + or "Cette commande réinitialise ou supprime des résultats de correction. Continuer ?", icon="warning", ): return @@ -893,6 +894,26 @@ class CopienatorApp(tk.Tk): if not step_id: return status = process_status(return_code, interrupted) + if step_id == "clean" and status == "success": + self._append_console( + f"\n[Terminé — code {return_code} — {STATUS_LABELS[status]}]\n" + ) + self.info_var.set( + "Nettoyage terminé : la progression précédente a été supprimée." + ) + self.active_step_id = None + # Do not recreate .copienator-gui.json after clean removed it. + self.state_store = StateStore() + self._populate_tree() + self._update_controls() + messagebox.showinfo( + "Nettoyage terminé", + "Les fichiers intermédiaires et la progression du GUI ont été " + "supprimés. Les copies traitées, les fichiers textuels de l’énoncé, " + "correction.json, les journaux et les livrables de A Rendre ont " + "été conservés.", + ) + return self.state_store.update_step(step_id, status=status, return_code=return_code) self.state_store.add_history( { diff --git a/copienator_gui/runner.py b/copienator_gui/runner.py index 0b19e90..e23ea1d 100644 --- a/copienator_gui/runner.py +++ b/copienator_gui/runner.py @@ -27,12 +27,15 @@ class ProcessRunner: command: list[str], cwd: Path, environment: dict[str, str], - log_path: Path, + log_path: Path | None, ) -> 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") + if log_path is not None: + log_path.parent.mkdir(parents=True, exist_ok=True) + self._log_file = log_path.open("wb") + else: + self._log_file = None self._interrupted = False kwargs: dict[str, Any] = {} diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index 5d8e071..90b51af 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -32,6 +32,7 @@ class CommandVariant: fixed_args: tuple[str, ...] = () fixed_args_before_positionals: bool = False dangerous: bool = False + danger_warning: str | None = None @dataclass(frozen=True) @@ -473,6 +474,36 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: personal=True, optional=True, ), + StepDefinition( + "clean", + "Archivage", + "Nettoyer les fichiers intermédiaires", + "Supprime définitivement les fichiers permettant de reprendre le parcours. " + "Conserve les PDF traités, les fichiers textuels de l’énoncé, correction.json, " + "les journaux, ainsi que les images et score.json de A Rendre.", + ( + python( + "default", + "Nettoyage définitif", + "clean", + fixed_args=("--yes",), + dangerous=True, + danger_warning=( + "Le nettoyage est irréversible.\n\n" + "Toute la progression du GUI et les fichiers intermédiaires " + "seront supprimés. Il ne sera plus possible de reprendre une " + "étape sans régénérer ses données.\n\n" + "Les PDF traités, les fichiers textuels de l’énoncé, " + "correction.json, les journaux, ainsi que les images et " + "score.json de A Rendre seront conservés.\n\n" + "Continuer ?" + ), + ), + ), + arguments=(arg_target("Dossier de l’évaluation"),), + optional=True, + requires=("Copies", "correction.json", "A Rendre"), + ), ] return [step for step in steps if show_personal_steps or not step.personal] diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 8cc1ee5..ee36de3 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -1652,6 +1652,20 @@ class WorkflowTests(unittest.TestCase): self.assertEqual(rotate[2:], ["-m", "copienator", "copies", "rotate", self.evaluation]) self.assertEqual(rename[2:], ["-m", "copienator", "copies", "rename", self.evaluation]) + def test_clean_is_the_last_dangerous_gui_step(self) -> None: + ordered_steps = build_workflow(True) + clean = ordered_steps[-1] + self.assertEqual(clean.id, "clean") + self.assertTrue(clean.optional) + self.assertTrue(clean.variants[0].dangerous) + self.assertIn("progression", clean.variants[0].danger_warning) + self.assertEqual( + command_arguments( + self.command("clean", "default", {"target": self.evaluation}) + ), + [self.evaluation, "--yes"], + ) + class CrossPlatformFileTests(unittest.TestCase): def test_rotate_all_skips_statement(self) -> None: