artifact-downloader.py

#!/usr/bin/env python3
"""Nexus Artifact Downloader

Search and download nexus artifacts.

Usage:
  ./artifact-downloader.py download [options]
  ./artifact-downloader.py search [options]
  ./artifact-downloader.py (-h | --help)
  ./artifact-downloader.py --version

Options:
  -h --help                  Show this screen.
  -v --version               Show version.
  --nexus-domain DOMAIN      Nexus artifact
  --nexus-port PORT          Nexus port
  --artifact-id ID           Artifact id
  --artifact-group GROUP     Artifact group
  --desired-version VERSION  Artifact desired version
  --packaging PKG            Artifact packaging [default: tar.gz]
  --repository REPOSITORY    Repository
  --quiet                    Do not display progress bar

"""

import os
import re
import requests

from tqdm import tqdm
from xml.etree import ElementTree
from envopt import envopt

artifact_id_key = '--artifact-id'
artifact_group_key = '--artifact-group'
desired_version_key = '--desired-version'
repository_key = '--repository'
nexus_domain_key = '--nexus-domain'
nexus_port_key = '--nexus-port'
packaging_key = '--packaging'

def download_nexus_artifact(arguments):
    download_url = 'http://{}:{}/service/local/artifact/maven/redirect?' \
                   'r={}&g={}&a={}&p={}&v={}'
    download_url = download_url.format(arguments[nexus_domain_key],
                                        arguments[nexus_port_key],
                                        arguments[repository_key],
                                        arguments[artifact_group_key],
                                        arguments[artifact_id_key],
                                        arguments[packaging_key],
                                        arguments[desired_version_key])

    print("HTTP GET %s" % download_url)

    response = requests.get(download_url, stream=True)

    if response == 404:
        raise RuntimeError(
            "Failed downloading file due to non 404 return code. "
            "File not found"
        )

    if response.status_code != 200:
        raise RuntimeError(
            "Failed downloading file due to non 200 return code. "
            "Return code was " + str(response.status_code)
        )

    filename = response.url.rsplit('/', 1)[-1]
    total_size = int(response.headers.get("content-length", 0))

    with tqdm(desc=filename, total=total_size, unit='B',
              unit_scale=True, miniters=1) as progress:
        with open(filename, "wb") as handle:
            for chunk in response.iter_content(32*1024):
                progress.update(len(chunk))
                handle.write(chunk)

def search_nexus_artifact(arguments):
    repo = '&repositoryId={}'.format(arguments[repository_key]) if arguments[repository_key] else ''
    group = '&g={}'.format(arguments[artifact_group_key]) if arguments[artifact_group_key] else ''

    search_url = 'http://{}:{}/service/local/lucene/search?&a={}{}{}'
    search_url = search_url.format(arguments[nexus_domain_key],
                                    arguments[nexus_port_key],
                                    arguments[artifact_id_key],
                                    arguments[artifact_group_key],
                                    group,
                                    repo)

    print("HTTP GET %s" % search_url)

    response = requests.get(search_url, stream=True)
    tree = ElementTree.fromstring(response.content)
    return tree.findall(".//artifact")

def select_artifact_version(desired_version, available_artifacts):
    filtered_artifacts = [artifact for artifact
                          in available_artifacts
                          if contains_version(artifact, desired_version)]

    print("Possible versions found:")
    print([artifact.find('.//version').text for artifact in filtered_artifacts])
    selected_artifact = False

    for artifact in filtered_artifacts:
        artifact_version = artifact.find('.//version').text

        if ver_cmp(desired_version, normalize_version(artifact_version)) < 0:
            desired_version = artifact_version
            selected_artifact = artifact
    return selected_artifact

def normalize_version(desired_version):
    if not re.match("^v?([0-9]+)(\\.[0-9]+)?(\\.[0-9]+)?(\\.[0-9]+)?$", desired_version):
        raise RuntimeError('Invalid Version')

    arr_artifact_version = desired_version.split('.')

    if len(arr_artifact_version) <= 3:
        while len(arr_artifact_version) < 3:
            arr_artifact_version.append('0')
    else:
        arr_artifact_version = arr_artifact_version[slice(3)]

    return '.'.join(arr_artifact_version)

def ver_tuple(z):
    return tuple([int(x) for x in z.split('.') if x.isdigit()])

def ver_cmp(a, b):
    return cmp(ver_tuple(a), ver_tuple(b))

def cmp(elm1, elm2):
    return ((elm1 > elm2) - (elm1 < elm2))

def contains_version(artifact, version):
    return artifact.find('.//version').text.find(version) == 0

def print_if_present(tree_element):
    print(tree_element)
    print(tree_element.text if tree_element else '')

def main(arguments):
    if not arguments[artifact_id_key]:
        raise RuntimeError('Must provide ' + artifact_id_key)

    all_available_artifacts = search_nexus_artifact(arguments)
    if arguments['search']:
        for artifact in all_available_artifacts:
            print(ElementTree.tostring(artifact, encoding='utf8', method='xml').decode("utf-8"))
            print_if_present(artifact.find('.//groupId'))
            print_if_present(artifact.find('.//artifactId'))
            print_if_present(artifact.find('.//version'))
            print_if_present(artifact.find('.//repositoryId'))
            print_if_present(artifact.find('.//extension'))
            print('---')

    if arguments['download']:
        if not arguments[desired_version_key]:
            raise RuntimeError('Must provide ' + desired_version_key)

        query_version = re.sub('^[0-9]|^.', '', arguments[desired_version_key])
        print("Searching for versions matching", query_version)

        selected_artifact = select_artifact_version(query_version, all_available_artifacts)
        if selected_artifact:
            repository = selected_artifact.find('.//repositoryId').text
            version = selected_artifact.find('.//version').text
            print("Most recent is", version)
            download_nexus_artifact(arguments)
        else:
            raise RuntimeError('Desired version matching artifact not found')

if __name__ == "__main__":
    arguments = envopt(__doc__, prefix='NXAD_', version='download nexus artifact')
    print(arguments)
    main(arguments)