easywks/package/release.py

from typing import Any, Dict, List, 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_job_ids(project_id: str, pipeline_id: str, api_token: str) -> Dict[str, str]:
    url: str = f'https://gitlab.com/api/v4/projects/{project_id}/pipelines/{pipeline_id}/jobs'
    headers: Dict[str, str] = {
        'Private-Token': api_token,
        'User-Agent': 'curl/7.70.0'
    }
    req = urllib.request.Request(url, headers=headers)
    try:
        resp: http.client.HTTPResponse = urllib.request.urlopen(req)
    except HTTPError as e:
        print(e.read().decode())
        sys.exit(1)
    resp_data: bytes = resp.read()
    joblist: List[Dict[str, Any]] = json.loads(resp_data.decode())

    jobidmap: Dict[str, str] = {}
    for job in joblist:
        name: str = job['name']
        job_id: str = job['id']
        jobidmap[name] = job_id
    return jobidmap


def fetch_single_shafile(url: str, api_token: str) -> str:
    headers: Dict[str, str] = {
        'User-Agent': 'curl/7.70.0',
        'Private-Token': api_token
    }
    req = urllib.request.Request(url, headers=headers)
    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, project_id: str, job_ids: Dict[str, str], api_token: str) -> Optional[Tuple[str, str]]:
    mybase: str = f'{base_url}/jobs/{job_ids["build_wheel"]}/artifacts/raw'
    wheel_sha_url: str = f'https://gitlab.com/api/v4/projects/{project_id}/jobs/{job_ids["build_wheel"]}'\
                         '/artifacts/dist/SHA256SUMS'
    wheel_filename: str = fetch_single_shafile(wheel_sha_url, api_token)
    wheel_url: str = f'{mybase}/dist/{wheel_filename}'
    return wheel_url, wheel_sha_url


def fetch_debian_url(base_url: str, project_id: str, job_ids: Dict[str, str], api_token: str) -> Optional[Tuple[str, str]]:
    mybase: str = f'{base_url}/jobs/{job_ids["build_debian"]}/artifacts/raw'
    debian_sha_url: str = f'https://gitlab.com/api/v4/projects/{project_id}/jobs/{job_ids["build_debian"]}'\
                          '/artifacts/package/debian/SHA256SUMS'
    debian_filename: str = fetch_single_shafile(debian_sha_url, api_token)
    debian_url: str = f'{mybase}/package/debian/{debian_filename}'
    return debian_url, debian_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')
    pipeline_id: Optional[str] = os.getenv('CI_PIPELINE_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)
    if pipeline_id is None:
        print('CI_PIPELINE_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)

    job_ids: Dict[str, str] = fetch_job_ids(project_id, pipeline_id, api_token)

    base_url: str = f'https://gitlab.com/{project_name}/-'

    wheel_url, wheel_sha_url = fetch_wheel_url(base_url, project_id, job_ids, api_token)
    debian_url, debian_sha_url = fetch_debian_url(base_url, project_id, job_ids, api_token)

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

### Download

- [Python Wheel]({wheel_url}) ([sha256]({wheel_sha_url}))
- [Debian Package]({debian_url}) ([sha256]({debian_sha_url}))'''
    # Docker currently not working
    # - Docker image: registry.gitlab.com/{project_name}:{release_tag}

    post_body: str = json.dumps({
        'tag_name': release_tag,
        'description': augmented_changelog,
        'assets': {
            'links': [
                {
                    'name': 'Python Wheel',
                    'url': wheel_url,
                    'link_type': 'package'
                },
                {
                    'name': 'Debian Package',
                    'url': debian_url,
                    'link_type': 'package'
                }
            ]
        }
    })

    gitlab_release_api_url: str = \
        f'https://gitlab.com/api/v4/projects/{project_id}/releases'
    headers: Dict[str, str] = {
        'Private-Token': api_token,
        'Content-Type': 'application/json; charset=utf-8',
        'User-Agent': 'curl/7.70.0'
    }

    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] = json.loads(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()