get-pr-packages

#!/usr/bin/python3

import argparse
import requests
import json
import subprocess
import os
import tempfile
import zipfile

is_dom0 = os.path.exists('/etc/qubes-release')
netvm = 'sys-net'

GITHUB_API = "https://api.github.com"
GITLAB_API = "https://gitlab.com/api/v4"

# FIXME: add option writing to file directly, otherwise artifacts are fetched
# to memory first
def get_url(url, output=None):
    if is_dom0:
        stdout = output if output else subprocess.PIPE
        data = subprocess.run(['qvm-run', '-p', '--nogui', netvm, 'curl', '-L', url],
                              stdout=stdout, check=True).stdout
    else:
        r = requests.get(url, stream=bool(output))
        r.raise_for_status()
        if output:
            for chunk in r.iter_content(chunk_size=8192):
                output.write(chunk)
            output.flush()
            data = None
        else:
            data = r.content
    return data

def get_job_url(pr_url):
    pr_details = json.loads(get_url(pr_url).decode())
    statuses_url = pr_details['statuses_url']

    job_url = None
    statuses = get_url(statuses_url)
    for status in json.loads(statuses.decode()):
        if status['context'] != "continuous-integration/pullrequest":
            continue
        repo, _, pipeline = status['target_url'] \
                     .replace('https://gitlab.com/', '') \
                     .partition('/-/pipelines/')
        r = get_url(f"{GITLAB_API}/projects/{repo.replace('/', '%2F')}/pipelines/{pipeline}/jobs")
        for job in json.loads(r.decode()):
            if job['status'] != 'success':
                continue
            if 'publish:repo' not in job['name']:
                continue
            return job['web_url']

    raise Exception("Successful 'publish:repo' job not found")

def download_repo(job_url, output_dir):
    repo_url = job_url + '/artifacts/raw/repo'
    
    with tempfile.TemporaryFile() as f:
        repo_zip_data = get_url(job_url + '/artifacts/download', output=f)
        f.seek(0)
        os.mkdir(output_dir)
        with zipfile.ZipFile(f, 'r') as repo_zip:
            repo_zip.extractall(output_dir)
        # get rid of 'repo' dir nesting
        for subdir in os.listdir(output_dir + '/repo'):
            os.rename(output_dir + '/repo/' + subdir, output_dir + '/' + subdir)

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Output dir, by default repo-pr-number, must not exists yet")
    parser.add_argument("pr", help="PR in form of owner/repo/pr-number")
    args = parser.parse_args()

    owner, repo, pr_number = args.pr.split('/')
    if args.output is None:
        args.output = f"{repo}-pr-{pr_number}"
    if os.path.exists(args.output):
        raise FileExistsError(args.output)

    pr_url = f"{GITHUB_API}/repos/{owner}/{repo}/pulls/{pr_number}"

    job_url = get_job_url(pr_url)

    download_repo(job_url, args.output)

if __name__ == '__main__':
    main()