Simple Python Script to Create An Offline Conda Channel

conda_offline_channel_creator.py

"""This script copies the conda packages into a directory structure suitable
for use as an custom offline channel.

Usage:
    conda_offline_channel_creator.py <PATH_TO_CONDA_PKG_FOLDER>
    <PATH_TO_OUTPUT_FOLDER>

Example usage:
    conda_offline_channel_creator.py D:/Miniconda3/pkgs D:/offline_conda-channel

Note:
    After running this script, the following command might be required: conda
    index <PATH_TO_OUTPUT_FOLDER>
"""


import multiprocessing.dummy
import os
import shutil
import sys
import urllib.request


def _get_arch_and_package_name(package_url):

    package_url = package_url.rstrip('\n')
    _, arch, package_name = package_url.rsplit('/', maxsplit=2)
    return (arch, package_name, package_url)


def _read_urls_txt(path_to_url_txt):

    with open(path_to_url_txt) as in_file:
        package_urls = in_file.readlines()
    packages = [_get_arch_and_package_name(u) for u in package_urls]
    return packages

def _download_package(package_url, output_path):

    urllib.request.urlretrieve(package_url, output_path)

def _download_missing_packages(packages, pkg_folder_path):

    available_packages = os.listdir(pkg_folder_path)
    missing_packages = []
    for _, package_name, package_url in packages:
        if package_name not in available_packages:
            destination_path = os.path.join(pkg_folder_path, package_name)
            missing_packages.append((package_url, destination_path))

    print('[==>] Downloading %d missing packages...' % len(missing_packages),
          end='')
    with multiprocessing.dummy.Pool() as pool:
        pool.starmap(_download_package, missing_packages)
    print('Done!')


def _copy_file(src_path, dest_path):

    try:
        shutil.copy(src_path, dest_path)
    except FileNotFoundError:
        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
        shutil.copy(src_path, dest_path)


def _copy_packages(packages, pkg_folder_path, output_path):

    copy_paths = []
    for arch, package_name, _ in packages:
        src_path = os.path.join(pkg_folder_path, package_name)
        dest_path = os.path.join(output_path, arch, package_name)
        copy_paths.append((src_path, dest_path))

    print('[*] Copying %d packages...' % len(packages), end='')
    with multiprocessing.dummy.Pool() as pool:
        pool.starmap(_copy_file, copy_paths)
    print('Done!')


def main(pkg_folder_path, output_path):

    packages = _read_urls_txt(os.path.join(pkg_folder_path, 'urls.txt'))
    _download_missing_packages(packages, pkg_folder_path)
    _copy_packages(packages, pkg_folder_path, output_path)


if __name__ == '__main__':

    if len(sys.argv) != 3:
        print('Usage: %s <PATH_TO_CONDA_PKG_FOLDER> <PATH_TO_OUTPUT_FOLDER>'
              % __file__)
        exit()

    main(sys.argv[1], sys.argv[2])