Standardisation des scripts

copienator/__init__.py

@@ -1,5 +1,13 @@
 """Core building blocks shared by Copienator scripts and interfaces."""
 
+from .cli import (
+    CliError,
+    ExitCode,
+    evaluation_parser,
+    evaluation_workspace,
+    execute,
+    workspace_from_args,
+)
 from .json_io import (
     JsonLockTimeout,
     atomic_update_json,
@@ -14,12 +22,18 @@ from .workspace import (
 )
 
 __all__ = [
+    "CliError",
     "EvaluationWorkspace",
+    "ExitCode",
     "JsonLockTimeout",
     "WorkspaceNotFoundError",
     "WorkspaceValidationError",
     "atomic_update_json",
     "atomic_write_json",
     "atomic_write_text",
+    "evaluation_parser",
+    "evaluation_workspace",
+    "execute",
     "read_json",
+    "workspace_from_args",
 ]

copienator/cli.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import argparse
+import sys
+import traceback
+from collections.abc import Callable, Sequence
+from enum import IntEnum
+from pathlib import Path
+from typing import Any
+
+from .workspace import (
+    EvaluationWorkspace,
+    WorkspaceNotFoundError,
+    WorkspaceValidationError,
+)
+
+
+class ExitCode(IntEnum):
+    SUCCESS = 0
+    FAILURE = 1
+    INVALID_ARGUMENTS = 2
+    INVALID_WORKSPACE = 3
+    PARTIAL = 4
+    INTERRUPTED = 130
+
+
+class CliError(Exception):
+    def __init__(self, message: str, exit_code: ExitCode = ExitCode.FAILURE) -> None:
+        self.exit_code = exit_code
+        super().__init__(message)
+
+
+def evaluation_parser(description: str) -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=description)
+    parser.add_argument(
+        "evaluation",
+        type=Path,
+        help="Evaluation directory",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Show a traceback when an unexpected error occurs",
+    )
+    return parser
+
+
+def evaluation_workspace(
+    path: str | Path,
+    *,
+    repository: str | Path | None = None,
+) -> EvaluationWorkspace:
+    root = Path(path).expanduser().resolve()
+    if not root.exists():
+        raise CliError(
+            f"Evaluation directory does not exist: {root}",
+            ExitCode.INVALID_WORKSPACE,
+        )
+    if not root.is_dir():
+        raise CliError(
+            f"Evaluation path is not a directory: {root}",
+            ExitCode.INVALID_WORKSPACE,
+        )
+    return EvaluationWorkspace(root, Path(repository) if repository is not None else None)
+
+
+def workspace_from_args(
+    args: argparse.Namespace,
+    *,
+    repository: str | Path | None = None,
+) -> EvaluationWorkspace:
+    return evaluation_workspace(args.evaluation, repository=repository)
+
+
+def execute(
+    parser: argparse.ArgumentParser,
+    argv: Sequence[str] | None,
+    handler: Callable[[argparse.Namespace], int | ExitCode | None],
+) -> int:
+    args = parser.parse_args(argv)
+    try:
+        result = handler(args)
+    except CliError as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        return int(exc.exit_code)
+    except (WorkspaceNotFoundError, WorkspaceValidationError) as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        return int(ExitCode.INVALID_WORKSPACE)
+    except KeyboardInterrupt:
+        print("Interrupted by user.", file=sys.stderr)
+        return int(ExitCode.INTERRUPTED)
+    except Exception as exc:  # noqa: BLE001 - executable boundary
+        if getattr(args, "verbose", False):
+            traceback.print_exc()
+        else:
+            print(f"Error: {exc}", file=sys.stderr)
+        return int(ExitCode.FAILURE)
+    return int(ExitCode.SUCCESS if result is None else result)
+
+
+def stderr(message: Any) -> None:
+    print(message, file=sys.stderr)

copienator/workspace.py

@@ -186,6 +186,24 @@ class EvaluationWorkspace:
         if missing:
             raise WorkspaceValidationError(self.root, missing)
 
+    def require_files(self, *relative_paths: str) -> None:
+        missing = [
+            relative_path
+            for relative_path in relative_paths
+            if not (self.root / relative_path).is_file()
+        ]
+        if missing:
+            raise WorkspaceValidationError(self.root, missing)
+
+    def require_directories(self, *relative_paths: str) -> None:
+        missing = [
+            relative_path
+            for relative_path in relative_paths
+            if not (self.root / relative_path).is_dir()
+        ]
+        if missing:
+            raise WorkspaceValidationError(self.root, missing)
+
     def ensure_control_directories(self) -> None:
         self.logs_dir.mkdir(parents=True, exist_ok=True)
         self.runs_dir.mkdir(parents=True, exist_ok=True)