matemat/package/release.py

#!/usr/bin/env python3

from typing import Any, Dict, Optional, Tuple

import os
import sys
import json
import urllib.request
import http.client
from urllib.error import HTTPError


def parse_changelog(tag: str) -> Optional[str]:
    release_changelog: str = ''
    with open('CHANGELOG.md', 'r') as f:
        in_target: bool = False
        done: bool = False
        for line in f.readlines():
            if in_target:
                if f'<!-- END RELEASE {tag} -->' in line:
                    done = True
                    break
                release_changelog += line
            elif f'<!-- BEGIN RELEASE {tag} -->' in line:
                in_target = True
                continue
    if not done:
        return None
    return release_changelog


def fetch_single_shafile(url: str) -> str:
    req = urllib.request.Request(url)
    try:
        resp: http.client.HTTPResponse = urllib.request.urlopen(req)
    except HTTPError as e:
        print(e.read().decode())
        sys.exit(1)
    resp_data: bytes = resp.readline()
    shafile: str = resp_data.decode()
    filename: str = shafile.strip().split(' ')[-1].strip()
    return filename


def fetch_wheel_url(base_url: str) -> Optional[Tuple[str, str]]:
    wheel_sha_url: str = f'{base_url}/dist/SHA256SUMS?job=build_wheel'
    wheel_filename: str = fetch_single_shafile(wheel_sha_url)
    wheel_url: str = f'{base_url}/dist/{wheel_filename}?job=build_wheel'
    return wheel_url, wheel_sha_url


def fetch_debian_url(base_url: str) -> Optional[Tuple[str, str]]:
    debian_sha_url: str = f'{base_url}/package/debian/SHA256SUMS?job=build_debian'
    debian_filename: str = fetch_single_shafile(debian_sha_url)
    debian_url: str = f'{base_url}/package/debian/{debian_filename}?job=build_debian'
    return debian_url, debian_sha_url


def fetch_arch_url(base_url: str) -> Optional[Tuple[str, str]]:
    arch_sha_url: str = f'{base_url}/package/archlinux/SHA256SUMS?job=build_archlinux'
    arch_filename: str = fetch_single_shafile(arch_sha_url)
    arch_url: str = f'{base_url}/package/archlinux/{arch_filename}?job=build_archlinux'
    return arch_url, arch_sha_url


def main():
    api_token: Optional[str] = os.getenv('GITLAB_API_TOKEN')
    release_tag: Optional[str] = os.getenv('CI_COMMIT_TAG')
    project_name: Optional[str] = os.getenv('CI_PROJECT_PATH')
    project_id: Optional[str] = os.getenv('CI_PROJECT_ID')
    if api_token is None:
        print('GITLAB_API_TOKEN is not set.', file=sys.stderr)
        sys.exit(1)
    if release_tag is None:
        print('CI_COMMIT_TAG is not set.', file=sys.stderr)
        sys.exit(1)
    if project_name is None:
        print('CI_PROJECT_PATH is not set.', file=sys.stderr)
        sys.exit(1)
    if project_id is None:
        print('CI_PROJECT_ID is not set.', file=sys.stderr)
        sys.exit(1)

    changelog: Optional[str] = parse_changelog(release_tag)
    if changelog is None:
        print('Changelog could not be parsed.', file=sys.stderr)
        sys.exit(1)

    base_url: str = f'https://gitlab.com/{project_name}/-/jobs/artifacts/{release_tag}/raw'

    wheel_url, wheel_sha_url = fetch_wheel_url(base_url)
    debian_url, debian_sha_url = fetch_wheel_url(base_url)
    arch_url, arch_sha_url = fetch_wheel_url(base_url)

    augmented_changelog = f'''{changelog.strip()}

### Download

- [Python Wheel]({wheel_url}) ([sha256]({wheel_sha_url}))
- [Debian Package]({debian_url}) ([sha256]({debian_sha_url}))
- [Arch Linux Package]({arch_url}) ([sha256]({arch_sha_url}))'''

    encoder = json.encoder.JSONEncoder()
    decoder = json.decoder.JSONDecoder()
    post_body: str = encoder.encode({'description': augmented_changelog})

    gitlab_release_api_url: str = \
        f'https://gitlab.com/api/v4/projects/{project_id}/repository/tags/{release_tag}/release'
    headers: Dict[str, str] = {
        'Private-Token': api_token
    }

    request = urllib.request.Request(
        gitlab_release_api_url,
        post_body.encode('utf-8'),
        headers=headers,
        method='POST'
    )
    try:
        response: http.client.HTTPResponse = urllib.request.urlopen(request)
    except HTTPError as e:
        print(e.read().decode())
        sys.exit(1)
    response_bytes: bytes = response.read()
    response_str: str = response_bytes.decode()
    response_data: Dict[str, Any] = decoder.decode(response_str)

    if response_data['tag_name'] != release_tag:
        print('Something went wrong...', file=sys.stderr)
        print(response_str, file=sys.stderr)
        sys.exit(1)

    print(response_data['description'])


if __name__ == '__main__':
    main()