Clean command

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é)
 

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())

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"),
 }
 

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",
+            "Nettoyage irréversible" if step.id == "clean" else "Confirmation requise",
-            "Cette commande réinitialise ou supprime des résultats de correction. Continuer ?",
+            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(
             {

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")
+        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] = {}

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]
 

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: