media-assets/scripts/publish_gitea_release_asset.py

#!/usr/bin/env python3
"""Publish or replace a fixed release asset on Gitea."""

from __future__ import annotations

import argparse
import json
import mimetypes
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


def _fail(message: str, *, status: int | None = None, body: str | None = None) -> None:
    print(f"ERROR: {message}", file=sys.stderr)
    if status is not None:
        print(f"HTTP status: {status}", file=sys.stderr)
    if body:
        print(f"Response body: {body}", file=sys.stderr)
    sys.exit(1)


def _request(
    method: str,
    url: str,
    token: str,
    *,
    data: bytes | None = None,
    headers: dict[str, str] | None = None,
    content_type: str | None = None,
) -> tuple[int, str]:
    req_headers = {"Authorization": f"token {token}"}
    if headers:
        req_headers.update(headers)
    if content_type:
        req_headers["Content-Type"] = content_type

    request = urllib.request.Request(url, data=data, headers=req_headers, method=method)
    try:
        with urllib.request.urlopen(request) as response:
            body = response.read().decode("utf-8", errors="replace")
            return response.status, body
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        return exc.code, body


def _api_base(server_url: str, repository: str) -> str:
    base = server_url.rstrip("/")
    owner, repo = repository.split("/", 1)
    return f"{base}/api/v1/repos/{owner}/{repo}"


def _parse_json(body: str) -> dict | list:
    if not body.strip():
        return {}
    return json.loads(body)


def _get_release_by_tag(api_base: str, token: str, tag: str) -> dict | None:
    url = f"{api_base}/releases/tags/{urllib.parse.quote(tag)}"
    status, body = _request("GET", url, token)
    if status == 404:
        return None
    if status >= 400:
        _fail(f"failed to query release for tag {tag}", status=status, body=body)
    return _parse_json(body)  # type: ignore[return-value]


def _create_release(
    api_base: str,
    token: str,
    *,
    tag: str,
    title: str,
    target_commitish: str | None,
) -> dict:
    payload: dict[str, object] = {
        "tag_name": tag,
        "name": title,
        "body": "Stable media-assets bundle generated by CI.",
        "draft": False,
        "prerelease": False,
    }
    if target_commitish:
        payload["target_commitish"] = target_commitish

    url = f"{api_base}/releases"
    status, body = _request(
        "POST",
        url,
        token,
        data=json.dumps(payload).encode("utf-8"),
        content_type="application/json",
    )
    if status >= 400:
        hint = ""
        if "tag" in body.lower() or status in (404, 422):
            hint = " 请先创建或允许 CI 推送固定 tag vlatest"
        _fail(
            f"failed to create release for tag {tag}.{hint}",
            status=status,
            body=body,
        )
    return _parse_json(body)  # type: ignore[return-value]


def _list_release_assets(api_base: str, token: str, release_id: int) -> list[dict]:
    url = f"{api_base}/releases/{release_id}/assets"
    status, body = _request("GET", url, token)
    if status >= 400:
        _fail(f"failed to list assets for release {release_id}", status=status, body=body)
    assets = _parse_json(body)
    if not isinstance(assets, list):
        _fail("unexpected assets response shape", body=body)
    return assets


def _delete_asset(api_base: str, token: str, asset_id: int) -> None:
    url = f"{api_base}/releases/assets/{asset_id}"
    status, body = _request("DELETE", url, token)
    if status >= 400:
        _fail(f"failed to delete asset {asset_id}", status=status, body=body)


def _upload_asset(
    api_base: str,
    token: str,
    release_id: int,
    *,
    asset_name: str,
    file_path: Path,
) -> dict:
    boundary = "----jiangchang-media-assets-boundary"
    file_bytes = file_path.read_bytes()
    content_type = mimetypes.guess_type(asset_name)[0] or "application/octet-stream"

    parts: list[bytes] = []
    parts.append(f"--{boundary}\r\n".encode("utf-8"))
    parts.append(
        (
            f'Content-Disposition: form-data; name="attachment"; filename="{asset_name}"\r\n'
            f"Content-Type: {content_type}\r\n\r\n"
        ).encode("utf-8")
    )
    parts.append(file_bytes)
    parts.append(b"\r\n")
    parts.append(f"--{boundary}--\r\n".encode("utf-8"))
    payload = b"".join(parts)

    query = urllib.parse.urlencode({"name": asset_name})
    url = f"{api_base}/releases/{release_id}/assets?{query}"
    status, body = _request(
        "POST",
        url,
        token,
        data=payload,
        content_type=f"multipart/form-data; boundary={boundary}",
    )
    if status >= 400:
        _fail(
            f"failed to upload asset {asset_name} to release {release_id}",
            status=status,
            body=body,
        )
    return _parse_json(body)  # type: ignore[return-value]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Publish a fixed Gitea release asset.")
    parser.add_argument("--tag", required=True, help="Release tag, e.g. vlatest")
    parser.add_argument("--title", required=True, help="Release title")
    parser.add_argument("--file", required=True, help="Local asset file path")
    parser.add_argument("--asset-name", required=True, help="Asset name in release")
    args = parser.parse_args(argv)

    server_url = os.environ.get("GITEA_SERVER_URL", "").strip()
    token = os.environ.get("GITEA_TOKEN", "").strip()
    repository = os.environ.get("GITEA_REPOSITORY", "").strip()
    target_commitish = os.environ.get("GITEA_SHA", "").strip() or None

    if not server_url:
        _fail("missing environment variable GITEA_SERVER_URL")
    if not token:
        _fail("missing environment variable GITEA_TOKEN")
    if not repository or "/" not in repository:
        _fail("missing or invalid environment variable GITEA_REPOSITORY")

    file_path = Path(args.file)
    if not file_path.is_file():
        _fail(f"asset file not found: {file_path}")

    api_base = _api_base(server_url, repository)
    release = _get_release_by_tag(api_base, token, args.tag)
    if release is None:
        print(f"Release for tag {args.tag} not found; creating...")
        release = _create_release(
            api_base,
            token,
            tag=args.tag,
            title=args.title,
            target_commitish=target_commitish,
        )
    else:
        print(f"Reusing existing release for tag {args.tag} (id={release['id']})")

    release_id = int(release["id"])
    assets = _list_release_assets(api_base, token, release_id)
    for asset in assets:
        if asset.get("name") == args.asset_name:
            asset_id = int(asset["id"])
            print(f"Deleting existing asset {args.asset_name} (id={asset_id})...")
            _delete_asset(api_base, token, asset_id)

    print(f"Uploading {file_path} as {args.asset_name}...")
    uploaded = _upload_asset(
        api_base,
        token,
        release_id,
        asset_name=args.asset_name,
        file_path=file_path,
    )

    download_url = uploaded.get("browser_download_url") or uploaded.get("download_url")
    if not download_url:
        owner, repo = repository.split("/", 1)
        download_url = (
            f"{server_url.rstrip('/')}/{owner}/{repo}/releases/download/"
            f"{args.tag}/{urllib.parse.quote(args.asset_name)}"
        )

    print(f"OK: uploaded release asset {args.asset_name}")
    print(f"Download URL: {download_url}")
    return 0


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