chatSummaries/tools/release.py

#!/usr/bin/env python3
"""Publishes a built .plugin file to a Gitea release.

Creates the release for the given tag if it does not exist, replaces any
same-named asset already attached, then uploads the artifact. Re-running is
safe: the release is reused and the asset is replaced rather than duplicated.

Stdlib only, so it runs on any runner image without a pip install step.

Usage:
    tools/release.py --api <repo-api-url> --tag v1.0.0 --artifact dist/x.plugin

The token is read from GITEA_TOKEN (or GITHUB_TOKEN as a fallback).
"""
from __future__ import annotations

import argparse
import json
import mimetypes
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Optional, Tuple

REPO_ROOT = Path(__file__).resolve().parent.parent


def fail(message: str) -> "None":
    print(f"error: {message}", file=sys.stderr)
    raise SystemExit(1)


def request(
    method: str,
    url: str,
    token: str,
    body: Optional[bytes] = None,
    content_type: Optional[str] = None,
) -> Tuple[int, bytes]:
    """Performs one API call, returning (status, body) instead of raising."""
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Authorization", f"token {token}")
    req.add_header("Accept", "application/json")
    if content_type:
        req.add_header("Content-Type", content_type)
    try:
        with urllib.request.urlopen(req) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read()
    except urllib.error.URLError as exc:
        fail(f"{method} {url} failed: {exc.reason}")
        raise  # unreachable, keeps type checkers quiet


def find_release(api: str, tag: str, token: str) -> Optional[dict]:
    status, payload = request("GET", f"{api}/releases/tags/{tag}", token)
    if status == 404:
        return None
    if status != 200:
        fail(f"looking up release {tag} returned HTTP {status}: {payload[:400]!r}")
    return json.loads(payload)


def create_release(api: str, tag: str, token: str, commitish: str, notes: str) -> dict:
    """Creates the release, which also creates the tag if it does not exist."""
    body = json.dumps(
        {
            "tag_name": tag,
            "target_commitish": commitish,
            "name": tag,
            "body": notes,
            "draft": False,
            "prerelease": False,
        }
    ).encode()
    status, payload = request("POST", f"{api}/releases", token, body, "application/json")
    if status != 201:
        fail(f"creating release {tag} returned HTTP {status}: {payload[:400]!r}")
    return json.loads(payload)


def update_release(api: str, release_id: int, token: str, commitish: str, notes: str) -> dict:
    """Refreshes an existing release so a rebuild reports the commit it came from.

    target_commitish is deliberately not changed: the tag already points at the
    commit that first introduced this version, and moving it would silently
    rewrite history for anyone who already fetched it.
    """
    body = json.dumps({"body": notes}).encode()
    status, payload = request(
        "PATCH", f"{api}/releases/{release_id}", token, body, "application/json"
    )
    if status != 200:
        fail(f"updating release {release_id} returned HTTP {status}: {payload[:400]!r}")
    return json.loads(payload)


def delete_existing_assets(api: str, release_id: int, name: str, token: str) -> int:
    """Removes same-named assets, since Gitea keeps duplicates side by side."""
    status, payload = request("GET", f"{api}/releases/{release_id}/assets", token)
    if status != 200:
        fail(f"listing assets returned HTTP {status}: {payload[:400]!r}")
    removed = 0
    for asset in json.loads(payload) or []:
        if asset.get("name") != name:
            continue
        status, payload = request(
            "DELETE", f"{api}/releases/{release_id}/assets/{asset['id']}", token
        )
        if status not in (204, 404):
            fail(f"deleting asset {asset['id']} returned HTTP {status}: {payload[:400]!r}")
        removed += 1
    return removed


def upload_asset(api: str, release_id: int, path: Path, name: str, token: str) -> dict:
    """Uploads the artifact as multipart/form-data."""
    boundary = uuid.uuid4().hex
    mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
    payload = b"".join(
        [
            f"--{boundary}\r\n".encode(),
            f'Content-Disposition: form-data; name="attachment"; filename="{name}"\r\n'.encode(),
            f"Content-Type: {mime}\r\n\r\n".encode(),
            path.read_bytes(),
            f"\r\n--{boundary}--\r\n".encode(),
        ]
    )
    url = f"{api}/releases/{release_id}/assets?name={urllib.parse.quote(name)}"
    status, response = request(
        "POST", url, token, payload, f"multipart/form-data; boundary={boundary}"
    )
    if status != 201:
        fail(f"uploading {name} returned HTTP {status}: {response[:400]!r}")
    return json.loads(response)


def main(argv: Optional[list] = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api", required=True, help="repo API base, e.g. https://host/api/v1/repos/owner/repo")
    parser.add_argument("--tag", required=True, help="release tag")
    parser.add_argument("--artifact", required=True, help="file to attach")
    parser.add_argument("--commitish", default="", help="target commit for a new tag")
    parser.add_argument("--asset-name", default="", help="attachment name; defaults to the file name")
    args = parser.parse_args(argv)

    token = os.environ.get("GITEA_TOKEN") or os.environ.get("GITHUB_TOKEN") or ""
    if not token:
        fail("GITEA_TOKEN is not set")

    artifact = Path(args.artifact)
    if not artifact.is_file() or artifact.stat().st_size == 0:
        fail(f"artifact {artifact} is missing or empty")
    asset_name = args.asset_name or artifact.name

    api = args.api.rstrip("/")
    version = json.loads((REPO_ROOT / "plugin.json").read_text(encoding="utf-8"))["version"]

    notes = [f"Automated build of version {version}."]
    if args.commitish:
        notes.append(f"Built from `{args.commitish[:12]}`.")
    notes.append("Install by opening the attached `.plugin` file in exteraGram.")
    notes_text = "\n\n".join(notes)

    release = find_release(api, args.tag, token)
    if release is None:
        release = create_release(api, args.tag, token, args.commitish, notes_text)
        print(f"created release {args.tag}")
    else:
        # Version unchanged since the last push: keep the same tag and refresh
        # it in place so the attachment always matches the newest commit.
        release = update_release(api, int(release["id"]), token, args.commitish, notes_text)
        print(f"rebuilding existing release {args.tag}")

    release_id = int(release["id"])
    removed = delete_existing_assets(api, release_id, asset_name, token)
    if removed:
        print(f"removed {removed} previous copy/copies of {asset_name}")

    asset = upload_asset(api, release_id, artifact, asset_name, token)
    print(f"uploaded {asset_name} ({artifact.stat().st_size} bytes) -> {asset.get('browser_download_url', '')}")
    return 0


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