railiance-platform/scripts/forgejo_package_prune.py

#!/usr/bin/env python3
"""Forgejo package retention prune — keep newest N versions per package."""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

DEFAULT_BASE = "https://forgejo.coulomb.social"
DEFAULT_OWNER = "coulomb"
DEFAULT_TYPES = ("container", "pypi", "npm", "generic")
DEFAULT_MAX_VERSIONS = 3
DEFAULT_APPS_ROOT = Path.home() / "railiance-apps"
FORGEJO_IMAGE_RE = re.compile(
    r"^forgejo\.coulomb\.social/(?:coulomb/)?(?P<name>[^:/]+)(?::(?P<tag>[^/\s]+))?$",
    re.IGNORECASE,
)


@dataclass(frozen=True)
class VersionRef:
    package_type: str
    name: str
    version: str

    def key(self) -> tuple[str, str, str]:
        return (self.package_type, self.name, self.version)


@dataclass(frozen=True)
class DeletePlan:
    package_type: str
    name: str
    version: str
    created_at: str
    protected: bool
    reason: str


def _parse_created_at(value: str | None) -> datetime:
    if not value:
        return datetime.min
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return datetime.min


def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:
    protected: set[tuple[str, str, str]] = set()
    if not apps_root.is_dir():
        return protected

    patterns = [
        apps_root / "helm" / "*-values.yaml",
        apps_root / "charts" / "*" / "values.yaml",
    ]
    paths: list[Path] = []
    for pattern in patterns:
        paths.extend(sorted(pattern.parent.glob(pattern.name)))

    try:
        import yaml  # type: ignore
    except ImportError:
        yaml = None

    for path in paths:
        text = path.read_text(encoding="utf-8")
        if yaml is not None:
            try:
                data = yaml.safe_load(text) or {}
            except Exception:
                data = {}
            image = data.get("image") if isinstance(data, dict) else None
            if isinstance(image, dict):
                repo = str(image.get("repository") or "").strip()
                tag = str(image.get("tag") or "").strip()
                if repo and tag:
                    match = FORGEJO_IMAGE_RE.match(repo) or FORGEJO_IMAGE_RE.match(
                        f"{repo}:{tag}"
                    )
                    if match:
                        name = match.group("name")
                        protected.add(("container", name, tag))
            continue

        repo_match = re.search(
            r"repository:\s*forgejo\.coulomb\.social/coulomb/([^\s]+)",
            text,
            re.IGNORECASE,
        )
        tag_match = re.search(r'^\s*tag:\s*"?([^"\s#]+)"?\s*$', text, re.MULTILINE)
        if repo_match and tag_match:
            protected.add(("container", repo_match.group(1), tag_match.group(1)))

    return protected


def _api_request(
    method: str,
    url: str,
    token: str,
    *,
    timeout: float = 30.0,
) -> Any:
    req = urllib.request.Request(
        url,
        method=method,
        headers={
            "Authorization": f"token {token}",
            "Accept": "application/json",
        },
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        body = resp.read().decode("utf-8")
        return json.loads(body) if body else None


def _paginate(
    base_url: str,
    token: str,
    path: str,
    *,
    page_size: int = 50,
) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    page = 1
    while True:
        query = urllib.parse.urlencode({"limit": page_size, "page": page})
        url = f"{base_url.rstrip('/')}{path}?{query}"
        payload = _api_request("GET", url, token)
        if not payload:
            break
        if isinstance(payload, list):
            batch = payload
        else:
            batch = payload.get("data") or payload.get("items") or []
        if not batch:
            break
        items.extend(batch)
        if len(batch) < page_size:
            break
        page += 1
    return items


def list_packages(
    base_url: str,
    token: str,
    owner: str,
    package_type: str,
) -> list[dict[str, Any]]:
    owner_q = urllib.parse.quote(owner)
    items: list[dict[str, Any]] = []
    page = 1
    page_size = 50
    while True:
        query = urllib.parse.urlencode(
            {"limit": page_size, "page": page, "type": package_type}
        )
        url = f"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}"
        payload = _api_request("GET", url, token)
        batch = payload if isinstance(payload, list) else []
        if not batch:
            break
        items.extend(batch)
        if len(batch) < page_size:
            break
        page += 1
    return items


def list_versions(
    base_url: str,
    token: str,
    owner: str,
    package_type: str,
    name: str,
) -> list[dict[str, Any]]:
    owner_q = urllib.parse.quote(owner)
    type_q = urllib.parse.quote(package_type)
    name_q = urllib.parse.quote(name, safe="")
    return _paginate(
        base_url,
        token,
        f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/versions",
    )


def delete_version(
    base_url: str,
    token: str,
    owner: str,
    package_type: str,
    name: str,
    version: str,
    *,
    dry_run: bool,
) -> None:
    owner_q = urllib.parse.quote(owner)
    type_q = urllib.parse.quote(package_type)
    name_q = urllib.parse.quote(name, safe="")
    version_q = urllib.parse.quote(version, safe="")
    path = f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/{version_q}"
    if dry_run:
        return
    url = f"{base_url.rstrip('/')}{path}"
    _api_request("DELETE", url, token)


def build_delete_plans(
    *,
    base_url: str,
    token: str,
    owner: str,
    package_types: list[str],
    max_versions: int,
    protected: set[tuple[str, str, str]],
) -> tuple[list[DeletePlan], list[str]]:
    plans: list[DeletePlan] = []
    errors: list[str] = []

    for package_type in package_types:
        try:
            packages = list_packages(base_url, token, owner, package_type)
        except urllib.error.HTTPError as exc:
            errors.append(f"list {package_type}: HTTP {exc.code}")
            continue
        except Exception as exc:  # noqa: BLE001
            errors.append(f"list {package_type}: {exc}")
            continue

        for package in packages:
            name = str(package.get("name") or package.get("package_name") or "")
            if not name:
                continue
            try:
                versions = list_versions(base_url, token, owner, package_type, name)
            except urllib.error.HTTPError as exc:
                errors.append(f"versions {package_type}/{name}: HTTP {exc.code}")
                continue
            except Exception as exc:  # noqa: BLE001
                errors.append(f"versions {package_type}/{name}: {exc}")
                continue

            sorted_versions = sorted(
                versions,
                key=lambda item: _parse_created_at(str(item.get("created_at") or "")),
                reverse=True,
            )
            keep = {
                str(item.get("version") or "")
                for item in sorted_versions[:max_versions]
                if str(item.get("version") or "")
            }
            for item in sorted_versions[max_versions:]:
                version = str(item.get("version") or "")
                if not version or version in keep:
                    continue
                key = (package_type, name, version)
                is_protected = key in protected
                plans.append(
                    DeletePlan(
                        package_type=package_type,
                        name=name,
                        version=version,
                        created_at=str(item.get("created_at") or ""),
                        protected=is_protected,
                        reason="protected_production_tag" if is_protected else "beyond_retention_depth",
                    )
                )
    return plans, errors


def load_token() -> str:
    for env_name in ("FORGEJO_TOKEN", "FORGEJO_ADMIN_TOKEN"):
        token = os.environ.get(env_name, "").strip()
        if token:
            return token
    token_file = os.environ.get(
        "FORGEJO_TOKEN_FILE",
        os.environ.get("FORGEJO_ADMIN_TOKEN_FILE", "/tmp/forgejo-tegwick-api-token"),
    )
    path = Path(token_file).expanduser()
    if path.is_file():
        return path.read_text(encoding="utf-8").strip()
    raise SystemExit(
        "ERROR: Forgejo API token required (read:package + write:package).\n"
        "  export FORGEJO_TOKEN=<pat>     # or FORGEJO_ADMIN_TOKEN\n"
        "  # or save PAT to /tmp/forgejo-tegwick-api-token (chmod 600)\n"
        "  Generate: https://forgejo.coulomb.social/user/settings/applications\n"
        "  See: railiance-platform/docs/forgejo-package-prune.md"
    )


def emit_summary(
    *,
    owner: str,
    max_versions: int,
    package_types: list[str],
    protected: set[tuple[str, str, str]],
    plans: list[DeletePlan],
    errors: list[str],
    apply: bool,
    deleted: list[DeletePlan],
) -> dict[str, Any]:
    would_delete = [p for p in plans if not p.protected]
    skipped_protected = [p for p in plans if p.protected]
    return {
        "kind": "forgejo_package_prune",
        "owner": owner,
        "max_versions": max_versions,
        "package_types": package_types,
        "protected_count": len(protected),
        "candidate_count": len(would_delete),
        "skipped_protected_count": len(skipped_protected),
        "deleted_count": len(deleted),
        "apply": apply,
        "would_delete": [
            {
                "type": p.package_type,
                "name": p.name,
                "version": p.version,
                "created_at": p.created_at,
            }
            for p in would_delete
        ],
        "skipped_protected": [
            {
                "type": p.package_type,
                "name": p.name,
                "version": p.version,
                "reason": p.reason,
            }
            for p in skipped_protected
        ],
        "deleted": [
            {
                "type": p.package_type,
                "name": p.name,
                "version": p.version,
            }
            for p in deleted
        ],
        "errors": errors,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Prune old Forgejo package versions")
    parser.add_argument("--owner", default=DEFAULT_OWNER)
    parser.add_argument("--base-url", default=os.environ.get("FORGEJO_BASE_URL", DEFAULT_BASE))
    parser.add_argument("--max-versions", type=int, default=DEFAULT_MAX_VERSIONS)
    parser.add_argument(
        "--types",
        default=",".join(DEFAULT_TYPES),
        help="Comma-separated package types",
    )
    parser.add_argument("--apps-root", type=Path, default=DEFAULT_APPS_ROOT)
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--json", action="store_true", help="Emit JSON summary on stdout")
    args = parser.parse_args(argv)

    apply = bool(args.apply)
    dry_run = not apply
    package_types = [part.strip() for part in args.types.split(",") if part.strip()]
    token = load_token()
    protected = collect_protected_versions(args.apps_root.expanduser())

    print(f"Forgejo package prune — owner={args.owner} keep={args.max_versions}", file=sys.stderr)
    print(f"Protected production tags: {len(protected)}", file=sys.stderr)

    plans, errors = build_delete_plans(
        base_url=args.base_url,
        token=token,
        owner=args.owner,
        package_types=package_types,
        max_versions=max(1, args.max_versions),
        protected=protected,
    )

    deleted: list[DeletePlan] = []
    for plan in plans:
        if plan.protected:
            print(
                f"  skip protected  {plan.package_type}/{plan.name}:{plan.version}",
                file=sys.stderr,
            )
            continue
        if dry_run:
            print(
                f"  would delete    {plan.package_type}/{plan.name}:{plan.version}",
                file=sys.stderr,
            )
            continue
        try:
            delete_version(
                args.base_url,
                token,
                args.owner,
                plan.package_type,
                plan.name,
                plan.version,
                dry_run=False,
            )
            deleted.append(plan)
            print(
                f"  deleted         {plan.package_type}/{plan.name}:{plan.version}",
                file=sys.stderr,
            )
        except urllib.error.HTTPError as exc:
            errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: HTTP {exc.code}")
        except Exception as exc:  # noqa: BLE001
            errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: {exc}")

    summary = emit_summary(
        owner=args.owner,
        max_versions=args.max_versions,
        package_types=package_types,
        protected=protected,
        plans=plans,
        errors=errors,
        apply=apply,
        deleted=deleted,
    )

    if args.json or not sys.stdout.isatty():
        print(json.dumps(summary, indent=2))
    else:
        print(json.dumps(summary, indent=2))

    return 1 if errors else 0


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