ecoarkpywhl/scripts/upload_release_assets.py

#!/usr/bin/env python3
"""
Upload wheel / zip assets to a Gitea Release.

Environment variables (set by Gitea runner or user):
  ENABLE_RELEASE_UPLOAD  — set to "true" to activate (default: skip)
  GITEA_TOKEN            — Gitea access token with `repo` scope
  RELEASE_TOKEN          — fallback if GITEA_TOKEN is unset
  GITHUB_SERVER_URL      — Gitea server URL (runner-provided)
  GITHUB_REPOSITORY      — owner/repo (runner-provided)
  GITHUB_REF_NAME        — tag name when pushing a tag (runner-provided)
  GITHUB_SHA             — commit SHA (runner-provided)

Usage:
  python3 scripts/upload_release_assets.py <assets-dir>
"""
import glob
import json
import os
import secrets
import sys
import urllib.error
import urllib.request


def env(key, default=""):
    return os.environ.get(key, default)


def _multipart_formdata(field, filename, filepath):
    boundary = "----GiteaReleaseBoundary{:x}".format(secrets.randbits(64))
    with open(filepath, "rb") as f:
        payload = f.read()
    body = (
        "--{}\r\n"
        'Content-Disposition: form-data; name="{}"; filename="{}"\r\n'
        "Content-Type: application/octet-stream\r\n\r\n"
    ).format(boundary, field, filename).encode() + payload + (
        "\r\n--{}--\r\n".format(boundary)
    ).encode()
    return boundary, body


def _api(method, url, headers, data=None, filepath=None):
    req_headers = dict(headers)
    body = None
    if filepath:
        boundary, body = _multipart_formdata("attachment", os.path.basename(filepath), filepath)
        req_headers["Content-Type"] = "multipart/form-data; boundary={}".format(boundary)
    elif data is not None:
        body = json.dumps(data).encode()
        req_headers.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url, data=body, headers=req_headers, method=method)
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        snippet = e.read().decode()[:200] if e.fp else ""
        print("  [release] HTTP {} on {} {}: {}".format(e.code, method, url, snippet))
        return e.code, None


def main():
    # ── Guard ──
    if env("ENABLE_RELEASE_UPLOAD").lower() not in ("true", "1", "yes"):
        print("[release] ENABLE_RELEASE_UPLOAD not active — skipping.")
        return

    token = env("GITEA_TOKEN") or env("RELEASE_TOKEN")
    if not token:
        print("[release] No token (GITEA_TOKEN / RELEASE_TOKEN) — skipping.")
        return

    server_url = env("GITHUB_SERVER_URL").rstrip("/")
    repo = env("GITHUB_REPOSITORY")
    ref_name = env("GITHUB_REF_NAME")
    sha = env("GITHUB_SHA")

    if not repo:
        print("[release] GITHUB_REPOSITORY unset — skipping.")
        return

    # ── Scan assets ──
    assets_dir = sys.argv[1] if len(sys.argv) > 1 else "output"
    whl_files = sorted(
        glob.glob(os.path.join(assets_dir, "*.whl"))
        + glob.glob(os.path.join(assets_dir, "wheels", "*.whl"))
    )
    zip_files = sorted(glob.glob(os.path.join(assets_dir, "*.zip")))
    all_files = whl_files + zip_files

    if not all_files:
        print("[release] No .whl or .zip found under {}/ — nothing to upload.".format(assets_dir))
        return

    tag = ref_name or ("build-{}".format(sha[:7]) if sha else "latest")

    # ── Gitea API ──
    api_base = "{}/api/v1/repos/{}".format(server_url, repo)
    headers = {"Authorization": "token {}".format(token), "Accept": "application/json"}

    # 1) Get or create release
    print("[release] Target tag: {}".format(tag))
    status, release = _api("GET", "{}/releases/tags/{}".format(api_base, tag), headers)
    if status == 200 and release:
        release_id = release["id"]
        print("[release] Found existing release #{} ({})".format(release_id, tag))
    else:
        print("[release] Creating release for tag {} ...".format(tag))
        status, release = _api("POST", "{}/releases".format(api_base), headers, {
            "tag_name": tag,
            "target_commitish": sha,
            "name": "EcoArk Wheels {}".format(tag),
            "body": "Automated wheel build from commit {}.".format(sha[:7] if sha else "unknown"),
            "draft": False,
            "prerelease": True,
        })
        if status not in (200, 201):
            print("[release] Failed to create release — skipping asset upload.")
            return
        release_id = release["id"]
        print("[release] Created release #{} ({})".format(release_id, tag))

    # 2) Upload each asset as an attachment
    for fpath in all_files:
        fname = os.path.basename(fpath)
        print("[release] Uploading {} ...".format(fname))
        _api("POST", "{}/releases/{}/assets".format(api_base, release_id), headers, filepath=fpath)

    print('[release] Done — {} asset(s) uploaded to "{}".'.format(len(all_files), tag))


if __name__ == "__main__":
    main()