hvm-docs/scripts/registry_gc.py

"""Gitea container-registry garbage collection.

Lists package versions for one container package and deletes versions
older than --keep-days. Always preserves:

  - the :latest tag
  - the --keep-latest most-recent date-tagged versions
  - anything pushed in the last --keep-days days

The actual disk reclaim happens on Gitea's next package GC cron (admin
site settings). This script just marks the versions for deletion.

Usage:

    python scripts/registry_gc.py \\
        --owner <user> \\
        --package <product>-docs-mcp \\
        --keep-days 90 \\
        --keep-latest 5

Auth: reads GITEA_TOKEN from env (set in the workflow as a secret).
"""
from __future__ import annotations

import argparse
import os
import sys
from datetime import datetime, timedelta, timezone
from urllib.request import Request, urlopen
from urllib.error import HTTPError
import json


GITEA_HOST = os.environ.get("GITEA_HOST", "https://git.jpaul.io")


def api(token: str, method: str, path: str) -> object:
    req = Request(f"{GITEA_HOST}{path}",
                  headers={"Authorization": f"token {token}"},
                  method=method)
    try:
        with urlopen(req, timeout=30) as r:
            body = r.read()
            return json.loads(body) if body else None
    except HTTPError as e:
        if e.code == 404:
            return None
        raise


def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--owner", required=True)
    p.add_argument("--package", required=True)
    p.add_argument("--keep-days", type=int, default=90)
    p.add_argument("--keep-latest", type=int, default=5)
    p.add_argument("--dry-run", action="store_true")
    args = p.parse_args()

    token = os.environ.get("GITEA_TOKEN")
    if not token:
        print("GITEA_TOKEN not set", file=sys.stderr)
        return 1

    versions = api(token, "GET",
                   f"/api/v1/packages/{args.owner}/container/{args.package}/versions") or []
    if not versions:
        print(f"no versions found for {args.owner}/{args.package}")
        return 0

    cutoff = datetime.now(timezone.utc) - timedelta(days=args.keep_days)

    # Date-tagged versions (YYYY.MM.DD), newest first
    date_tagged = []
    for v in versions:
        tags = v.get("tags") or []
        for t in tags:
            if len(t) == 10 and t[4] == "." and t[7] == ".":
                date_tagged.append((t, v))
                break
    date_tagged.sort(key=lambda kv: kv[0], reverse=True)
    keep_date_tags = {t for t, _ in date_tagged[:args.keep_latest]}

    deleted = 0
    for v in versions:
        tags = v.get("tags") or []
        if "latest" in tags:
            continue
        if any(t in keep_date_tags for t in tags):
            continue
        try:
            created = datetime.fromisoformat(v["created_at"].replace("Z", "+00:00"))
        except (KeyError, ValueError):
            continue
        if created >= cutoff:
            continue
        version_id = v.get("id")
        print(f"  deleting v{version_id}  tags={tags}  created={v['created_at']}")
        if not args.dry_run:
            api(token, "DELETE",
                f"/api/v1/packages/{args.owner}/container/{args.package}/versions/{version_id}")
            deleted += 1
    print(f"done: {deleted} version(s) deleted")
    return 0


if __name__ == "__main__":
    sys.exit(main())