Jungfraujoch/gitea_upload_file.py

#!/usr/bin/env python3

# Uploads a file as an asset to the Gitea release for the tag in the VERSION file.
# Uses only the standard library (no 'requests'), so it runs with a bare Python
# interpreter on the Linux package runners and on the Windows viewer runner alike.

import json
import os
import sys
import uuid
import urllib.request
import urllib.error
from typing import Optional

gitea_api_url = "https://gitea.psi.ch/api/v1"
repo_owner = "mx"
repo_name = "jungfraujoch"

gitea_token = os.getenv('TOKEN')
if gitea_token is None:
    print("ERROR: Required environment variables not set")
    sys.exit(1)


def read_version():
    try:
        with open('VERSION', 'r') as f:
            return f.readline().strip()
    except FileNotFoundError:
        print("ERROR: VERSION file not found")
        sys.exit(1)


def _api_request(method: str, url: str, headers: dict, data: Optional[bytes] = None):
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, resp.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()


def get_release_by_tag(version: str):
    """Fetch release information by tag name; exits on error."""
    url = f'{gitea_api_url}/repos/{repo_owner}/{repo_name}/releases/tags/{version}'
    headers = {'Authorization': f'token {gitea_token}'}
    status, body = _api_request('GET', url, headers)
    if status != 200:
        print(f"ERROR: Failed to fetch release {version}. Status code: {status}")
        print(body.decode(errors='replace'))
        sys.exit(1)
    return json.loads(body)


def _multipart_body(fields: dict, file_field: str, file_name: str, file_bytes: bytes):
    """Encode a multipart/form-data body; returns (body, content_type)."""
    boundary = uuid.uuid4().hex
    sep = f'--{boundary}'.encode()
    parts = []
    for name, value in fields.items():
        parts += [sep, f'Content-Disposition: form-data; name="{name}"'.encode(),
                  b'', value.encode()]
    parts += [sep,
              f'Content-Disposition: form-data; name="{file_field}"; filename="{file_name}"'.encode(),
              b'Content-Type: application/octet-stream', b'', file_bytes]
    parts += [f'--{boundary}--'.encode(), b'']
    return b'\r\n'.join(parts), f'multipart/form-data; boundary={boundary}'


def upload_file_to_release(version: str, file_path: str, name: Optional[str] = None, label: Optional[str] = None):
    """Upload a file as an asset to the release identified by the given version tag."""
    if not os.path.isfile(file_path):
        print(f"ERROR: File not found: {file_path}")
        sys.exit(1)

    asset_name = name if name else os.path.basename(file_path)
    print(f"Uploading file '{file_path}' to release {version} as {asset_name}")

    release_id = get_release_by_tag(version).get('id')
    if release_id is None:
        print("ERROR: Release ID not found in response")
        sys.exit(1)

    fields = {'name': asset_name}
    if label:
        fields['label'] = label

    with open(file_path, 'rb') as f:
        body, content_type = _multipart_body(fields, 'attachment', asset_name, f.read())

    url = f'{gitea_api_url}/repos/{repo_owner}/{repo_name}/releases/{release_id}/assets'
    headers = {'Authorization': f'token {gitea_token}', 'Content-Type': content_type}
    status, resp_body = _api_request('POST', url, headers, body)
    if status not in (200, 201):
        print(f"ERROR: Failed to upload asset. Status code: {status}")
        print(resp_body.decode(errors='replace'))
        sys.exit(1)

    asset = json.loads(resp_body)
    download_url = asset.get('browser_download_url') or asset.get('url')
    print(f"Uploaded asset '{asset_name}' to release {version}: {download_url}")


if __name__ == '__main__':
    if len(sys.argv) != 2:
        exit(1)
    upload_file_to_release(read_version(), sys.argv[1])