download_artifact.py

#!/usr/bin/env python3
# Eclipse SUMO, Simulation of Urban MObility; see https://eclipse.dev/sumo
# Copyright (C) 2008-2026 German Aerospace Center (DLR) and others.
# This program and the accompanying materials are made available under the
# terms of the Eclipse Public License 2.0 which is available at
# https://www.eclipse.org/legal/epl-2.0/
# This Source Code may also be made available under the following Secondary
# Licenses when the conditions for such availability set forth in the Eclipse
# Public License 2.0 are satisfied: GNU General Public License, version 2
# or later which is available at
# https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later

# @file    download_artifact.py
# @author  Michael Behrisch
# @date    2024-05-03

import argparse
import io
import os
import sys
import zipfile

import requests


def request(url, token):
    if token:
        return requests.get(url, headers={"Authorization": "Bearer " + token})
    return requests.get(url)


def get_latest_artifact_url(options):
    prefix = "%s/repos/%s/%s/actions/" % (options.api_url, options.owner, options.repository)
    workflow_id = 0
    response = request(prefix + "workflows", options.token)
    for workflow in response.json()['workflows']:
        # for several workflows with the same name we take the one most recently created (highest id)
        if workflow['name'] == options.workflow and workflow['id'] > workflow_id:
            workflow_id = workflow['id']
    if workflow_id == 0:
        raise RuntimeError("Workflow '%s' not found." % options.workflow)

    workflow_run_ids = []
    response = request("%sworkflows/%s/runs" % (prefix, workflow_id), options.token)
    for workflow_run in response.json()['workflow_runs']:
        if options.branch == "main" and workflow_run['head_branch'][:1] == "v":
            # there seems to be no easy way to identify a tag so we take the first letter
            workflow_run_ids.append(workflow_run['id'])
    for workflow_run in response.json()['workflow_runs']:
        if (workflow_run['status'] == "completed"
            and (options.allow_failed or workflow_run['conclusion'] == "success")
                and workflow_run['head_branch'] == options.branch):
            workflow_run_ids.append(workflow_run['id'])
            break
    if not workflow_run_ids:
        raise RuntimeError("No successful workflow run found in branch '%s'." % options.branch)

    for workflow_run_id in workflow_run_ids:
        response = request("%sruns/%s/artifacts" % (prefix, workflow_run_id), options.token)
        for artifact in response.json()['artifacts']:
            if artifact['name'].startswith(options.prefix):
                yield "%sartifacts/%s/zip" % (prefix, artifact['id'])


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--api-url", default="https://api.github.com")
    ap.add_argument("--owner", default="eclipse-sumo")
    ap.add_argument("--repository", default="sumo")
    ap.add_argument("--workflow", default="cibuildwheel")
    ap.add_argument("--branch", default="main")
    ap.add_argument("--token", help="GitHub authentication token")
    ap.add_argument("--directory", help="output directory")
    ap.add_argument("--prefix", default="cibw", help="prefix of the artifact zip file")
    ap.add_argument("--allow-failed", action="store_true", default=False, help="download even if the build failed")
    ap.add_argument("-v", "--verbose", action="store_true", default=False, help="tell me more")
    options = ap.parse_args()

    if not options.token:
        for cred_path in (".", os.path.dirname(__file__), os.path.expanduser("~")):
            if os.path.exists(os.path.join(cred_path, ".git-credentials")):
                with open(os.path.join(cred_path, ".git-credentials")) as f:
                    options.token = f.read().split(":")[-1].split("@")[0]
                    break
    if not options.token:
        sys.exit("no authentication token found, please use the option --token or provide a .git-credentials file")
    for artifact_url in get_latest_artifact_url(options):
        response = request(artifact_url, options.token)
        if response.status_code == 200:
            with zipfile.ZipFile(io.BytesIO(response.content)) as zip:
                zip.extractall(options.directory)
        if options.verbose:
            print(artifact_url, response)