silver_pose/v1/release.py

"""Build and verify a credential-free Silver Pose V1 release directory."""

import argparse
import hashlib
import json
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Iterable


class ReleaseError(ValueError):
    """Raised when a release package cannot be safely created or verified."""


_EXCLUDED_NAMES = {"config.local.json", "artifacts", "__pycache__", ".pytest_cache", "models"}


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with Path(path).open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def build_release_package(
    project_root: Path, model_path: Path, expected_sha256: str, output_dir: Path
) -> Dict:
    """Copy the V1 runtime and a locked model without local customer artifacts."""

    root = Path(project_root).resolve()
    source_model = Path(model_path).resolve()
    output = Path(output_dir).resolve()
    expected = _validate_model(source_model, expected_sha256)
    if output.exists():
        raise ReleaseError("output directory already exists: {0}".format(output))

    output.mkdir(parents=True)
    _copy_runtime_tree(root / "v1", output / "v1")
    _copy_required_file(root / "README.md", output / "README.md")
    _copy_required_file(root / "init.ps1", output / "init.ps1")
    _copy_required_file(root / "docs" / "demo-runbook.md", output / "docs" / "demo-runbook.md")
    destination_model = output / "v1" / "models" / "best.pt"
    destination_model.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(str(source_model), str(destination_model))
    _write_release_config(root / "v1" / "config.example.json", output / "v1", expected)
    manifest = _manifest(output, expected)
    (output / "release-manifest.json").write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8"
    )
    return manifest


def verify_release_package(package_dir: Path) -> Dict:
    package = Path(package_dir).resolve()
    manifest_path = package / "release-manifest.json"
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise ReleaseError("cannot read release manifest: {0}".format(exc))
    for relative_path, expected in manifest.get("files", {}).items():
        actual_path = package / relative_path
        if not actual_path.is_file() or sha256_file(actual_path) != expected:
            raise ReleaseError("release file SHA-256 does not match: {0}".format(relative_path))
    model = package / "v1" / "models" / "best.pt"
    if not model.is_file() or sha256_file(model) != manifest.get("model_sha256"):
        raise ReleaseError("model SHA-256 does not match release manifest")
    return manifest


def _validate_model(model_path: Path, expected_sha256: str) -> str:
    expected = str(expected_sha256).lower()
    if len(expected) != 64 or any(char not in "0123456789abcdef" for char in expected):
        raise ReleaseError("expected model SHA-256 must be 64 hexadecimal characters")
    if not model_path.is_file():
        raise ReleaseError("model file does not exist: {0}".format(model_path))
    if sha256_file(model_path) != expected:
        raise ReleaseError("model SHA-256 does not match expected value")
    return expected


def _copy_runtime_tree(source: Path, destination: Path) -> None:
    if not source.is_dir():
        raise ReleaseError("V1 source directory does not exist: {0}".format(source))
    shutil.copytree(
        str(source),
        str(destination),
        ignore=shutil.ignore_patterns(*_EXCLUDED_NAMES),
    )


def _copy_required_file(source: Path, destination: Path) -> None:
    if not source.is_file():
        raise ReleaseError("required release file does not exist: {0}".format(source))
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(str(source), str(destination))


def _write_release_config(source: Path, destination_dir: Path, model_sha256: str) -> None:
    raw = json.loads(source.read_text(encoding="utf-8"))
    raw["model"]["sha256"] = model_sha256
    (destination_dir / "config.release.example.json").write_text(
        json.dumps(raw, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )


def _manifest(package: Path, model_sha256: str) -> Dict:
    files = {}
    for path in _files(package):
        files[path.relative_to(package).as_posix()] = sha256_file(path)
    return {
        "format": "silver-pose-v1-release-1",
        "created_at_utc": datetime.now(timezone.utc).isoformat(),
        "model_sha256": model_sha256,
        "files": files,
    }


def _files(root: Path) -> Iterable[Path]:
    for path in sorted(root.rglob("*")):
        if path.is_file() and path.name != "release-manifest.json":
            yield path


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description="Build or verify a Silver Pose V1 release")
    commands = parser.add_subparsers(dest="command", required=True)
    build = commands.add_parser("build")
    build.add_argument("--project-root", default=".")
    build.add_argument("--model-path", required=True)
    build.add_argument("--model-sha256", required=True)
    build.add_argument("--output", required=True)
    verify = commands.add_parser("verify")
    verify.add_argument("--package", required=True)
    args = parser.parse_args(argv)
    if args.command == "build":
        manifest = build_release_package(
            Path(args.project_root), Path(args.model_path), args.model_sha256, Path(args.output)
        )
        print("release package built: {0}".format(args.output))
        print("model SHA-256: {0}".format(manifest["model_sha256"]))
        return 0
    verify_release_package(Path(args.package))
    print("release package verified: {0}".format(args.package))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())