shd.build/tools/game-data-capture/src/game_data_capture/cli.py

"""CLI contracts for the capture utility."""

from __future__ import annotations

import argparse
import json
import sys
from collections.abc import Sequence
from dataclasses import replace
from pathlib import Path

from game_data_capture.config import AppConfig, ConfigError, ServerConfig, load_config
from game_data_capture.doctor import doctor_as_dict, run_doctor
from game_data_capture.paths import PathConfigurationError, RuntimePaths, resolve_runtime_paths
from game_data_capture.runtime import CaptureRuntime, RuntimeFailure

EXIT_OK = 0
EXIT_CONFIG_ERROR = 2
EXIT_UNIMPLEMENTED = 3
EXIT_RUNTIME_ERROR = 4


def _add_config_argument(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--config", type=Path, help="absolute path to a TOML configuration file")


def _port_argument(raw_value: str) -> int:
    try:
        port = int(raw_value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("port must be an integer from 1024 to 65535") from error
    if not 1024 <= port <= 65535:
        raise argparse.ArgumentTypeError("port must be an integer from 1024 to 65535")
    return port


def build_parser() -> argparse.ArgumentParser:
    """Build the stable command surface without initializing runtime resources."""

    parser = argparse.ArgumentParser(
        prog="shd-capture", description="Local SHD.build game-data capture"
    )
    subcommands = parser.add_subparsers(dest="command", required=True)

    doctor = subcommands.add_parser("doctor", help="report local prerequisite status")
    _add_config_argument(doctor)
    doctor.add_argument("--json", action="store_true", help="emit JSON diagnostics")

    serve = subcommands.add_parser("serve", help="start the localhost review service")
    _add_config_argument(serve)
    serve.add_argument("--host", help="loopback IP literal override")
    serve.add_argument("--port", type=_port_argument, help="TCP port override")

    watch = subcommands.add_parser("watch", help="watch the Wayland clipboard for images")
    _add_config_argument(watch)
    watch_modes = watch.add_mutually_exclusive_group()
    watch_modes.add_argument("--arm", action="store_true", help="arm live clipboard capture")
    watch_modes.add_argument(
        "--dry-run", action="store_true", help="validate the capture runtime plan"
    )

    ingest_image = subcommands.add_parser("ingest-image", help="stage one image for capture")
    _add_config_argument(ingest_image)
    ingest_image.add_argument("image", type=Path)

    import_video = subcommands.add_parser("import-video", help="stage a video for frame extraction")
    _add_config_argument(import_video)
    import_video.add_argument("video", type=Path)

    export = subcommands.add_parser("export", help="write a versioned ISAC handoff")
    _add_config_argument(export)
    export.add_argument("destination", type=Path)

    verify = subcommands.add_parser("verify", help="verify an evidence archive")
    _add_config_argument(verify)
    verify.add_argument("archive", type=Path)

    evaluate = subcommands.add_parser("eval", help="evaluate local OCR and classification")
    _add_config_argument(evaluate)
    evaluate.add_argument("corpus", type=Path)

    purge = subcommands.add_parser("purge", help="purge retained local data")
    _add_config_argument(purge)
    purge.add_argument("--confirm", action="store_true", help="confirm a future destructive purge")

    return parser


def _load_cli_config(config_path: Path | None) -> tuple[AppConfig, RuntimePaths]:
    paths = resolve_runtime_paths()
    return load_config(config_path=config_path, paths=paths), paths


def _doctor(arguments: argparse.Namespace) -> int:
    config, paths = _load_cli_config(arguments.config)
    diagnostics = doctor_as_dict(run_doctor(config, paths))
    if arguments.json:
        print(json.dumps(diagnostics, separators=(",", ":"), sort_keys=True))
    else:
        for diagnostic in diagnostics:
            print(f"{diagnostic['name']}: {diagnostic['status']} ({diagnostic['detail']})")
    return EXIT_OK


def _serve_contract(arguments: argparse.Namespace) -> int:
    config, paths = _load_cli_config(arguments.config)
    host = arguments.host if arguments.host is not None else config.server.host
    port = arguments.port if arguments.port is not None else config.server.port
    configured = replace(config, server=ServerConfig(host=host, port=port))
    CaptureRuntime(configured, paths).serve()
    return EXIT_OK


def _unimplemented(command: str) -> int:
    print(
        f"{command}: subsystem is not implemented; requires gdc-runtime-integration",
        file=sys.stderr,
    )
    return EXIT_UNIMPLEMENTED


def _watch(arguments: argparse.Namespace) -> int:
    config, paths = _load_cli_config(arguments.config)
    if not arguments.arm and not arguments.dry_run:
        return _unimplemented("watch")
    runtime = CaptureRuntime(config, paths)
    if arguments.dry_run:
        runtime.validate_plan()
        print("watch dry-run: capture runtime plan is valid")
        return EXIT_OK
    runtime.watch()
    return EXIT_OK


def _ingest_image(arguments: argparse.Namespace) -> int:
    config, paths = _load_cli_config(arguments.config)
    CaptureRuntime(config, paths).ingest_image(arguments.image)
    print("image capture stored")
    return EXIT_OK


def main(argv: Sequence[str] | None = None) -> int:
    """Run one CLI command and return a stable process exit status."""

    parser = build_parser()
    arguments = parser.parse_args(argv)
    try:
        if arguments.command == "doctor":
            return _doctor(arguments)
        if arguments.command == "serve":
            return _serve_contract(arguments)
        if arguments.command == "watch":
            return _watch(arguments)
        if arguments.command == "ingest-image":
            return _ingest_image(arguments)
        _load_cli_config(arguments.config)
        return _unimplemented(arguments.command)
    except (ConfigError, PathConfigurationError) as error:
        print(f"configuration error: {error}", file=sys.stderr)
        return EXIT_CONFIG_ERROR
    except RuntimeFailure:
        print("capture runtime failed", file=sys.stderr)
        return EXIT_RUNTIME_ERROR