bass-foundry/scripts/release.py

#!/usr/bin/env python3
"""Validate independent product tags and publish Windows assets to Gitea."""
import argparse
import hashlib
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request
import uuid
import zipfile
from plugin_catalog import ROOT, archive_name, catalog, tag_version

class SafeRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, request, fp, code, msg, headers, newurl):
        redirect = super().redirect_request(request, fp, code, msg, headers, newurl)
        if redirect and urllib.parse.urlsplit(request.full_url)[:2] != urllib.parse.urlsplit(newurl)[:2]:
            redirect.remove_header("Authorization")
        return redirect

class Gitea:
    def __init__(self, server, repository, token):
        parts = repository.split("/")
        if len(parts) != 2 or not all(parts):
            raise ValueError("GITEA_REPOSITORY must be owner/repository")
        self.server = server.rstrip("/")
        self.base = self.server + "/api/v1/repos/" + "/".join(urllib.parse.quote(p, safe="") for p in parts)
        self.token = token
        self.opener = urllib.request.build_opener(SafeRedirect())

    def fetch(self, method, url, data=None, content_type=None, missing=False):
        headers = {}
        if urllib.parse.urlsplit(url)[:2] == urllib.parse.urlsplit(self.server)[:2]:
            headers["Authorization"] = "token " + self.token
        if content_type:
            headers["Content-Type"] = content_type
        request = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            with self.opener.open(request, timeout=120) as response:
                return response.read()
        except urllib.error.HTTPError as error:
            if error.code == 404 and missing:
                return None
            raise RuntimeError(f"Gitea {method}: HTTP {error.code}") from None

    def request(self, method, path, data=None, missing=False):
        raw = self.fetch(method, self.base + path,
                         json.dumps(data).encode() if data is not None else None,
                         "application/json" if data is not None else None, missing)
        return json.loads(raw) if raw else None

    def download(self, url):
        return self.fetch("GET", url)

    def upload(self, release_id, path):
        boundary = "rivet-patina-" + uuid.uuid4().hex
        body = (
            f"--{boundary}\r\nContent-Disposition: form-data; name=\"attachment\"; filename=\"{path.name}\"\r\n"
            "Content-Type: application/octet-stream\r\n\r\n"
        ).encode() + path.read_bytes() + f"\r\n--{boundary}--\r\n".encode()
        url = self.base + f"/releases/{release_id}/assets?name=" + urllib.parse.quote(path.name, safe="")
        return json.loads(self.fetch("POST", url, body, f"multipart/form-data; boundary={boundary}"))

def release_notes(plugin, release_version, root=ROOT):
    notes = (root / "docs" / plugin / "CHANGELOG.md").read_text()
    match = re.search(r"^## " + re.escape(release_version) + r"\s*\n(.*?)(?=^## |\Z)", notes, re.M | re.S)
    if not match:
        raise ValueError(f"Add a {release_version} section to docs/{plugin}/CHANGELOG.md")
    return match[1].strip()

def plan(tag, root=ROOT):
    tag = tag.removeprefix("refs/tags/")
    plugin = next((p for p in catalog(root) if tag.startswith(p + "-v")), None)
    if plugin is None:
        raise ValueError("Use a <plugin>-v<version> tag, for example " + ", ".join(p + "-v1.0.0" for p in catalog(root)))
    release_version = tag_version(plugin, tag, root)
    release_notes(plugin, release_version, root)
    return {"plugin": plugin, "version": release_version, "tag": tag,
            "archive": archive_name(plugin, release_version)}

def publish(plugin, tag, client, root=ROOT):
    metadata = plan(tag, root)
    if metadata["plugin"] != plugin:
        raise ValueError("Selected plugin does not match the release tag")
    tag = metadata["tag"]
    product = catalog(root)[plugin]
    archive = root / "dist" / metadata["archive"]
    sidecar = archive.with_suffix(".zip.sha256")
    expected = f"{hashlib.sha256(archive.read_bytes()).hexdigest()}  {archive.name}\n"
    if sidecar.read_text() != expected:
        raise ValueError("Archive checksum does not match its sidecar")
    with zipfile.ZipFile(archive) as package:
        if package.testzip() is not None or [n for n in package.namelist() if n.endswith(".clap")] != [f"{product['name']}.clap"]:
            raise ValueError("Release ZIP must contain exactly the selected plugin")
        sums = package.read("SHA256SUMS").decode().splitlines()
        for line in sums:
            digest, name = line.split("  ", 1)
            if hashlib.sha256(package.read(name)).hexdigest() != digest:
                raise ValueError("Internal package checksum mismatch")
    escaped = urllib.parse.quote(tag, safe="")
    # Do not let CreateRelease silently manufacture a missing repository tag.
    if client.request("GET", "/tags/" + escaped, missing=True) is None:
        raise ValueError("The release tag must already exist in Gitea")
    release = client.request("GET", "/releases/tags/" + escaped, missing=True)
    if release is None:
        release = client.request("POST", "/releases", {
            "name": f"{product['name']} {metadata['version']}", "tag_name": tag,
            "body": release_notes(plugin, metadata["version"], root),
            "draft": True, "prerelease": "-" in metadata["version"],
        })
    assets = release.get("assets", [])
    for path in [archive, sidecar]:
        existing = next((asset for asset in assets if asset["name"] == path.name), None)
        if existing is not None:
            if client.download(existing["browser_download_url"]) != path.read_bytes():
                raise ValueError(f"Refusing to replace release asset with different content: {path.name}")
        else:
            client.upload(release["id"], path)
    # New releases remain drafts until both assets have uploaded successfully.
    if release.get("draft"):
        client.request("PATCH", f"/releases/{release['id']}", {"draft": False})
    print(f"Published {tag}: {archive.name}")

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=["plan", "validate", "publish"])
    parser.add_argument("--plugin", choices=catalog())
    parser.add_argument("--tag", required=True)
    args = parser.parse_args()
    metadata = plan(args.tag)
    if args.plugin is not None and metadata["plugin"] != args.plugin:
        raise ValueError("Selected plugin does not match the release tag")
    if args.action in ["plan", "validate"]:
        print(f"Validated {metadata['tag']}")
        output = os.environ.get("GITHUB_OUTPUT") or os.environ.get("GITEA_OUTPUT")
        if args.action == "plan" and output:
            with open(output, "a", encoding="utf-8") as stream:
                for name, value in metadata.items():
                    stream.write(f"{name}={value}\n")
        return
    if os.environ.get("GITEA_REF") != "refs/tags/" + metadata["tag"]:
        raise ValueError("Publishing requires a matching Gitea tag workflow")
    client = Gitea(os.environ["GITEA_SERVER_URL"], os.environ["GITEA_REPOSITORY"], os.environ["GITEA_TOKEN"])
    publish(metadata["plugin"], metadata["tag"], client)

if __name__ == "__main__":
    try:
        main()
    except (ValueError, RuntimeError, KeyError, OSError) as error:
        raise SystemExit(str(error)) from None