durin/tools/gitea_release_upload.py

#!/usr/bin/env python2
# -*- coding: utf-8 -*-

import json
import mimetools
import mimetypes
import os
import sys
import urllib2


def fail(msg, code=1):
    sys.stderr.write("ERROR: %s\n" % msg)
    sys.exit(code)


def api_request(url, token, method="GET", json_data=None, content_type=None):
    data = None
    headers = {
        "Authorization": "token %s" % token,
    }

    if json_data is not None:
        data = json.dumps(json_data)
        headers["Content-Type"] = "application/json"
    elif content_type is not None:
        headers["Content-Type"] = content_type

    request = urllib2.Request(url, data=data, headers=headers)
    request.get_method = lambda: method

    try:
        response = urllib2.urlopen(request)
        body = response.read()
        status = response.getcode()
        return status, body
    except urllib2.HTTPError as e:
        return e.code, e.read()


def encode_multipart_formdata(fields, files):
    boundary = mimetools.choose_boundary()
    crlf = "\r\n"
    lines = []

    for key, value in fields:
        lines.append("--" + boundary)
        lines.append('Content-Disposition: form-data; name="%s"' % key)
        lines.append("")
        lines.append(value)

    for key, filename, content, content_type in files:
        lines.append("--" + boundary)
        lines.append(
            'Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename)
        )
        lines.append("Content-Type: %s" % content_type)
        lines.append("")
        lines.append(content)

    lines.append("--" + boundary + "--")
    lines.append("")
    body = crlf.join(lines)
    content_type = "multipart/form-data; boundary=%s" % boundary
    return content_type, body


def multipart_request(url, token, fields, files):
    content_type, body = encode_multipart_formdata(fields, files)
    headers = {
        "Authorization": "token %s" % token,
        "Content-Type": content_type,
    }

    request = urllib2.Request(url, data=body, headers=headers)
    request.get_method = lambda: "POST"

    try:
        response = urllib2.urlopen(request)
        return response.getcode(), response.read()
    except urllib2.HTTPError as e:
        return e.code, e.read()


def get_release_by_tag(api_base, token, tag):
    url = "%s/releases/tags/%s" % (api_base, tag)
    status, body = api_request(url, token, method="GET")
    if status == 200:
        return json.loads(body)
    if status == 404:
        return None
    fail("failed to fetch release for tag %s: HTTP %s\n%s" % (tag, status, body))


def create_release(api_base, token, tag):
    url = "%s/releases" % api_base
    payload = {
        "tag_name": tag,
        "name": tag,
        "draft": False,
        "prerelease": False,
    }
    status, body = api_request(url, token, method="POST", json_data=payload)
    if status not in (200, 201):
        fail("failed to create release for tag %s: HTTP %s\n%s" % (tag, status, body))
    return json.loads(body)


def ensure_release(api_base, token, tag):
    release = get_release_by_tag(api_base, token, tag)
    if release is not None:
        print("Release for tag %s already exists (id=%s)" % (tag, release.get("id")))
        return release

    print("Release for tag %s does not exist, creating it" % tag)
    release = create_release(api_base, token, tag)
    print("Created release id=%s" % release.get("id"))
    return release


def upload_asset(api_base, token, release_id, file_path):
    if not os.path.isfile(file_path):
        fail("file not found: %s" % file_path)

    asset_name = os.path.basename(file_path)
    mime_type = mimetypes.guess_type(asset_name)[0] or "application/octet-stream"

    with open(file_path, "rb") as f:
        content = f.read()

    url = "%s/releases/%s/assets" % (api_base, release_id)
    status, body = multipart_request(
        url,
        token,
        fields=[("name", asset_name)],
        files=[("attachment", asset_name, content, mime_type)],
    )

    if status not in (200, 201):
        fail("failed to upload asset %s: HTTP %s\n%s" % (asset_name, status, body))

    print("Uploaded asset: %s" % asset_name)


def find_assets(build_dir):
    names = []
    for name in sorted(os.listdir(build_dir)):
        if name.startswith("libdurin-plugin.so"):
            full = os.path.join(build_dir, name)
            if os.path.isfile(full):
                names.append(full)
    return names


def main():
    if len(sys.argv) != 4:
        sys.stderr.write(
            "Usage: %s <gitea-server> <owner/repo> <tag>\n" % sys.argv[0]
        )
        sys.stderr.write(
            "Example: %s https://gitea.psi.ch mx/durin 1.0.0\n" % sys.argv[0]
        )
        sys.exit(1)

    server = sys.argv[1].rstrip("/")
    repo = sys.argv[2]
    tag = sys.argv[3]

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

    build_dir = "build"
    if not os.path.isdir(build_dir):
        fail("build directory not found: %s" % build_dir)

    assets = find_assets(build_dir)
    if not assets:
        fail("no libdurin-plugin.so* files found in %s" % build_dir)

    api_base = "%s/api/v1/repos/%s" % (server, repo)

    release = ensure_release(api_base, token, tag)
    release_id = release.get("id")
    if not release_id:
        fail("release id missing in API response")

    for asset in assets:
        upload_asset(api_base, token, release_id, asset)

    print("Done.")


if __name__ == "__main__":
    main()