download_minecraft_assets.py

import json
import requests
import argparse
from pathlib import Path
from multiprocessing.pool import ThreadPool
import threading
import sys

VERSIONS_URL = 'https://launchermeta.mojang.com/mc/game/version_manifest.json'
OBJECTS_URL = 'http://resources.download.minecraft.net/{hash[0]}{hash[1]}/{hash}'

class AtomicCounter:
    def __init__(self, initial=0):
        self.value = initial
        self._lock = threading.Lock()

    def increment(self, num=1):
        with self._lock:
            self.value += num
            return self.value

NUM_FILES_DOWNLOADED = AtomicCounter()


def download_object(path_str, hash, version, total):
    i = NUM_FILES_DOWNLOADED.increment()
    print(f'Downloading file {i} / {total}: {path_str}')
    path = Path.cwd() / 'assets' / version / Path(path_str)
    Path.mkdir(path.parent, parents=True, exist_ok=True)
    object_url = OBJECTS_URL.format(hash=hash)
    with open(path, 'wb') as download_file:
        req = requests.get(object_url)
        download_file.write(req.content)

def main():
    parser = argparse.ArgumentParser(
        description='Download default Minecraft assets'
    )

    action = parser.add_mutually_exclusive_group(
        required=True
    )
    action.add_argument(
        '-d', '--download', 
        type=str,
        action='store',
        dest='version',
        help='Download a given version of the Minecraft assets'
    )
    action.add_argument(
        '-l', '--list',
        help='List all available versions',
        action='store_true',
    )
    parser.add_argument(
        '-t', '--types',
        help='Which types to include, defaults to only releases',
        nargs='+',
        choices=[
            'snapshot',
            'release',
            'old_beta',
            'old_alpha',
        ],
        default=['release'],
    )

    args = parser.parse_args()

    versions = requests.get(VERSIONS_URL).json()['versions']
    if args.list:
        for version in versions:
            if version['type'] in args.types:
                print(version['id'])
    else:
        try:
            requested_version = next(
                    x for x in versions if x['id'] == args.version
            )
        except StopIteration:
            sys.exit(f'Invalid version: {args.version}')
        version_metadata = requests.get(requested_version['url']).json()
        objects = requests.get(version_metadata['assetIndex']['url']).json()['objects']

        pool = ThreadPool(5)
        pool.starmap(
            download_object,
            ((
                path,
                obj['hash'],
                args.version,
                len(objects)
            ) for path, obj in objects.items())
        )
        pool.close()
        pool.join()
            

if __name__ == '__main__':
    main()