download-latest-firmware-build.py

#!/usr/bin/python3

from pprint import pprint
import requests
import zipfile
import tempfile
from io import BytesIO
import argparse
import os.path
import re
import tempfile
import sys
import logging
import subprocess

GITLAB_API_BASE = "https://gitlab.freifunk-stuttgart.de/api/v4"
PROJECT_ID = 1

ap = argparse.ArgumentParser()
ap.add_argument("--pipeline-id", help="Pipeline ID to download. If omitted, download latest successfull.", default=None)
ap.add_argument("--pipeline-id-file", help="Store downloaded Pipeline ID in this file and only download if latest pipeline ID doesn't match file contents")
ap.add_argument("--create-symlink", help="Create symlink to downloaded firmware at the specified path")
ap.add_argument("--debug", action="store_true", help="Produce lots of debug output")
ap.add_argument("--branch", help="Download only builds of this branch")
args = ap.parse_args()

def find_version_from_archive(archive_file_list):
    for file in archive_file_list:
        if file.filename.startswith("gluon/output/images"):
            filename = os.path.basename(file.filename)
            version_regex = re.compile(r'gluon-ffs-(((experimental|[0-9]+\.[0-9])+[+][0-9]{4}-[0-9]{2}-[0-9]{2})-g\.[a-f0-9]+-s\.[a-f0-9]+-)')
            version_matches = version_regex.match(filename)
            if version_matches:
                logging.debug("Found version number {}".format(version_matches.group(2)))
                return version_matches.group(2)
    raise ValueError("Could not determine version from ZIP file")

def extract_zip(artifact_zipfile):
    logging.debug("Extracting ZIP from FD {}".format(artifact_zipfile))
    with zipfile.ZipFile(artifact_zipfile) as artifact_zip:
        version = find_version_from_archive(artifact_zip.infolist())
        with tempfile.TemporaryDirectory(dir=os.getcwd()) as tempdir:
            # Python ZipFile doesn't support symlinks
            # https://bugs.python.org/issue27318
            logging.debug("Running 'unzip'")
            subprocess.check_call(['unzip', artifact_zipfile.name, '-d{}'.format(tempdir)], stdin=artifact_zipfile)
            outputdir = os.path.join(tempdir, "gluon", "output")
            os.rename(outputdir, version)
        return version

def find_latest_pipeline_id(branch):
    logging.debug("Finding pipeline ID")
    pipelines_request = requests.get("{}/projects/{}/pipelines".format(GITLAB_API_BASE, PROJECT_ID))
    pipelines_request.raise_for_status()

    pipelines = pipelines_request.json()
    for pipeline in pipelines:
        if pipeline["status"] == "success" and pipeline["ref"].startswith(branch):
            logging.debug("Found Pipeline ID: {}".format(pipeline["id"]))
            return pipeline["id"]
    return None

if args.debug:
    logging.basicConfig(level=logging.DEBUG)

if args.pipeline_id is None:
    pipeline_id = int(find_latest_pipeline_id(args.branch))

    if args.pipeline_id_file:
        try:
            with open(args.pipeline_id_file, "r") as pipeline_id_file:
                pipeline_id_from_file = int(pipeline_id_file.read())
                
                if pipeline_id == pipeline_id_from_file:
                    print("Pipeline up to date")
                    sys.exit(1)
        except FileNotFoundError:
            pass
else:
    pipeline_id = args.pipeline_id
   
pipeline_jobs_request = requests.get("{}/projects/{}/pipelines/{}/jobs".format(GITLAB_API_BASE, PROJECT_ID, pipeline_id))
pipeline_jobs_request.raise_for_status()

pipeline_jobs = pipeline_jobs_request.json()
for job in pipeline_jobs:
    if job["name"] == "package":
        with tempfile.NamedTemporaryFile() as artifact_temp:
            artifact_url = "{}/projects/{}/jobs/{}/artifacts".format(GITLAB_API_BASE, PROJECT_ID, job["id"])
            logging.debug("Starting download via wget of URL '{}'".format(artifact_url))
            wget_process = subprocess.run(["wget", "-O" + artifact_temp.name, artifact_url], encoding='utf-8')
            if wget_process.returncode != 0:
                print(wget_process.stdout)

            downloaded_version = extract_zip(artifact_temp)

            if args.create_symlink:
                if os.path.islink(args.create_symlink):
                    os.remove(args.create_symlink)
                images_dir = os.path.abspath(os.path.join(downloaded_version, "images"))
                os.symlink(images_dir, args.create_symlink)

            if args.pipeline_id_file:
                with open(args.pipeline_id_file, "w") as pipeline_id_file:
                    pipeline_id_file.write(str(pipeline_id))