Launcher/core_lib/game/version_info.py

import json
import logging
import re
import warnings
from collections.abc import Callable
from pathlib import Path

import requests

from ..common.common import FileObject, download_file
from ..common.exception import VersionManifestFetchException, NoVersionFoundException, \
    NoSpecifiedVersionKeyException, VersionDataKeyNotFoundException, VersionManifestSaveException, HashMismatchException

MOJANG_VERSION_MANIFEST_V2 = "https://piston-meta.mojang.com/mc/game/version_manifest_v2.json"
MOJANG_VERSION_MANIFEST = "https://piston-meta.mojang.com/mc/game/version_manifest.json"
MOJANG_LIBRARIES_ENDPOINT = "https://libraries.minecraft.net/"

NATIVE_OS_KEY_MAP = {
    "darwin": "macos",
    "macos": "macos",
    "osx": "macos",
    "linux": "linux",
    "windows": "windows",
}

NATIVE_OS_RULE_MAP = {
    "darwin": "osx",
    "macos": "osx",
    "osx": "osx",
    "linux": "linux",
    "windows": "windows",
}

NATIVE_ARCH_KEY_MAP = {
    "x86": "x86",
    "i386": "x86",
    "i486": "x86",
    "i586": "x86",
    "i686": "x86",

    "x86_64": "x86_64",
    "amd64": "x86_64",

    "aarch64": "arm64",
    "arm64": "arm64",
    "aarch_64": "arm64",
}

NATIVE_OLD_TO_NEW_ARCH = {
    "32": "x86",
    "64": "x86_64",

    "x86": "x86",
    "x86_64": "x86_64",
    "amd64": "x86_64",

    "aarch64": "arm64",
    "arm64": "arm64",
}

NATIVE_NEW_TO_OLD_ARCH = {
    "x86": "32",
    "x86_64": "64",
}

RUNTIME_PLATFORM_MAP = {
    "linux": "linux",
    "darwin": "mac-os",
    "macos": "mac-os",
    "osx": "mac-os",
    "windows": "windows",
    "win32": "windows",
    "cygwin": "windows",
    "msys": "windows",
}

RUNTIME_ARCH_EMPTY = "$EMPTY"

RUNTIME_ARCH_MAP = {
    "linux": {
        "i386": "i386",
        "i686": "i386",
        "x86": "i386",
        "amd64": RUNTIME_ARCH_EMPTY,
        "x86_64": RUNTIME_ARCH_EMPTY,
    },
    "windows": {
        "i386": "x86",
        "i686": "x86",
        "x86": "x86",
        "amd64": "x64",
        "x86_64": "x64",
        "aarch64": "arm64",
        "arm64": "arm64",
    },
    "mac-os": {
        "amd64": RUNTIME_ARCH_EMPTY,
        "x86_64": RUNTIME_ARCH_EMPTY,
        "aarch64": "arm64",
        "arm64": "arm64",
    },
}

logger = logging.getLogger("Launcher.CoreLib")

def get_version_manifest(manifest_url=MOJANG_VERSION_MANIFEST_V2, timeout=10) -> dict | None:
    """
    Get version manifest.
    :param manifest_url: Source of the version manifest. (Default use MOJANG_VERSION_MANIFEST_V2)
    :param timeout: Request timeout
    :return:
        version_manifest: dict
    """
    logger.debug("Getting version manifest from {}".format(manifest_url))

    try:
        r = requests.get(manifest_url, timeout=timeout)
        r.raise_for_status()
    except requests.exceptions.RequestException as e:
        raise VersionManifestFetchException(
            "Error getting version manifest from {}: {}".format(manifest_url, e)
        )

    try:
        data = r.json()
        logger.debug("VersionManifest: {}".format(json.dumps(data, indent=4)))
        return data
    except ValueError as e:
        raise VersionManifestFetchException(
            "Unable to deserialize response data into dictionary: {}".format(e)
        )

def get_latest_version(version_manifest: dict, is_snapshot=False) -> None | str:
    """
    Get latest version (id) from version manifest.
    :param version_manifest: Version manifest.
    :param is_snapshot: Return latest snapshot version.
    :return:
        latest_version: str
    """
    latest = version_manifest.get("latest", {})

    if len(latest) == 0:
        raise NoVersionFoundException(
            "No latest version section available in version manifest."
        )

    latest_ver = latest.get("release" if not is_snapshot else "snapshot", None)

    if latest_ver is None:
        raise NoSpecifiedVersionKeyException(
            "No latest version available."
        )
    else:
        logger.debug("Latest version: {}".format(latest_ver))

    return latest_ver

def get_all_versions(version_manifest: dict, is_snapshot=False, include_all_type=False, specified_type=None) -> list:
    """
    Get all versions (id) from version manifest.
    :param version_manifest:
    :param is_snapshot: Return all versions that type is snapshot.
    :param include_all_type: Return all versions (all types included)
    :param specified_type: Specific version type
    :return:
        filtered_versions: list
    """
    all_vers = version_manifest.get("versions", [])

    if len(all_vers) == 0:
        raise NoVersionFoundException(
            "No version section available in version manifest."
        )

    if not specified_type:
        target_type = "release" if not is_snapshot else "snapshot"
    else:
        target_type = specified_type

    filtered = []

    for ver in all_vers:
        ver_id = ver.get("id", None)
        ver_type = ver.get("type")

        logger.debug("Ver: {}, Type: {}".format(ver_id, ver_type))

        if ver_type == target_type or include_all_type:
            filtered.append(ver)

    if len(filtered) == 0:
        raise NoSpecifiedVersionKeyException("No version matches specified type: {}".format(target_type))

    return filtered

def get_available_version_types(version_manifest: dict) -> list:
    """
    Get available version types from version manifest.
    :param version_manifest:
    :return:
        available_types: list
    """
    all_vers = version_manifest.get("versions", [])

    if len(all_vers) == 0:
        raise NoVersionFoundException(
            "No version section available in version manifest."
        )

    types = []

    for ver in all_vers:
        ver_type = ver.get("type")

        if ver_type not in types:
            types.append(ver_type)

    return types

def get_specific_version_manifest_url(version_manifest: dict, version_id: str) -> str | None:
    """
    Get specific version's manifest url from version manifest.
    :param version_manifest:
    :param version_id:
    :return:
        specific_version_manifest_url: str
    """
    versions = version_manifest.get("versions", [])

    if len(versions) == 0:
        raise NoVersionFoundException(
            "No version section available in version manifest."
        )

    for version in versions:
        if version.get("id") == version_id:
            specific_ver_manifest_url = version.get("url", None)

            if specific_ver_manifest_url is None:
                raise VersionManifestFetchException(
                    "Specific version manifest URL not available in version manifest."
                )

            return specific_ver_manifest_url

    raise NoSpecifiedVersionKeyException(
        "Specified version {} manifest URL not available in version manifest.".format(version_id)
    )

def get_specific_version_manifest_url_and_hash(version_manifest: dict, version_id: str) \
        -> tuple[str,str] | tuple[None, None]:
    """
    Get specific version's manifest url and hash from version manifest.
    :param version_manifest:
    :param version_id:
    :return:
    """
    versions = version_manifest.get("versions", [])

    if len(versions) == 0:
        raise NoVersionFoundException(
            "No version section available in version manifest."
        )

    for version in versions:
        if version.get("id") == version_id:
            specific_ver_manifest_url = version.get("url", None)
            specific_ver_manifest_hash = version.get("sha1", None)

            if specific_ver_manifest_url is None:
                raise VersionManifestFetchException(
                    "Specific version manifest URL is None. Is the manifest URL or source correct?"
                )

            if specific_ver_manifest_hash is None:
                logger.debug("Specific version manifest hash not available in version manifest.")

            return specific_ver_manifest_url, specific_ver_manifest_hash

    raise NoSpecifiedVersionKeyException(
        "Specified version {} manifest URL not available in version manifest.".format(version_id)
    )


def fetch_specific_version_manifest(version_manifest: dict, spec_version: str) -> tuple[bytes, str | None]:
    """
    Get specific version's manifest data and hash from version manifest.
    :param version_manifest:
    :param spec_version:
    :return:
        specific_version_manifest_data: bytes
        hash: str (sha1)
    """
    ver_url, sha1 = get_specific_version_manifest_url_and_hash(version_manifest, spec_version)

    try:
        r = requests.get(ver_url, timeout=10)
        r.raise_for_status()
    except requests.exceptions.RequestException as e:
        raise VersionManifestFetchException(
            "Failed to fetch version manifest URL: {}\n{}".format(ver_url, e)
        )

    return r.content, sha1

# mapping
get_specific_version_manifest = fetch_specific_version_manifest

def find_useful_part_in_specific_version_manifest(spec_version_manifest: dict)\
        -> tuple[dict | None, dict | None, dict | None, dict | None, list | None, str | None, str | None]:
    """
    Return useful part from version manifest.
    :param spec_version_manifest: Specific version's data
    :return:
        arguments: dict (contains jvm and game args, for newer version of Minecraft)
        asset_index: dict
        downloads: dict (Game core files are store in here)
        java_version: dict
        libraries: dict
        main_class: str
        legacy_game_args: str (aka "minecraftArguments" in old Minecraft version data)
    """
    arguments = spec_version_manifest.get("arguments", {})
    asset_index = spec_version_manifest.get("assetIndex", {})
    downloads = spec_version_manifest.get("downloads", {})
    java_version = spec_version_manifest.get("javaVersion", {})
    libraries = spec_version_manifest.get("libraries", [])
    main_class = spec_version_manifest.get("mainClass", None)
    legacy_game_args = spec_version_manifest.get("minecraftArguments", None)

    if not arguments and not legacy_game_args:
        raise VersionDataKeyNotFoundException(
            "No version manifest available in version manifest. (\"arguments\" or \"minecraftArguments\")"
        )

    if len(asset_index) == 0:
        raise VersionDataKeyNotFoundException(
            "No asset index section available in version manifest. (\"assetIndex\")"
        )

    if len(downloads) == 0:
        raise VersionDataKeyNotFoundException(
            "No downloads section available in version manifest. (\"downloads\")"
        )

    if len(java_version) == 0:
        raise VersionDataKeyNotFoundException(
            "No java version section available in version manifest. (\"javaVersion\")"
        )

    if len(libraries) == 0:
        raise VersionDataKeyNotFoundException(
            "No libraries section available in version manifest. (\"libraries\")"
        )

    if main_class is None:
        raise VersionDataKeyNotFoundException(
            "No main class section available in version manifest. (\"mainClass\")"
        )

    return arguments, asset_index, downloads, java_version, libraries, main_class, legacy_game_args

def get_core_file_download_url_and_hash(version_manifest: dict, is_server: bool=False) -> tuple[str, str] | tuple[None, None]:
    """
    Get core file download url and hash from version manifest.
    :param version_manifest:
    :param is_server: Download server version core file (aka "server.jar") if is enabled. Default is False and ony
     download client version core file ("client.jar")
    :return:
        core_file_download_url: str
        hash: str (sha1)
    """
    core_downloads = version_manifest.get("downloads", {})

    if len(core_downloads) == 0:
        raise VersionDataKeyNotFoundException(
            "No downloads section available in version manifest. (\"downloads\")"
        )

    core_name = "client" if not is_server else "server"
    core_section = core_downloads.get(core_name, {})

    if len(core_section) == 0:
        raise VersionDataKeyNotFoundException(
            logger.debug("No target core section available in version manifest. Wants: {}".format(core_name))
        )

    download_url = core_section.get("url", None)
    sha1 = core_section.get("sha1", None)

    if download_url is None:
        raise VersionDataKeyNotFoundException(
            "Core file url is None.\n"
        )

    if sha1 is None:
        warnings.warn(
            "Core file sha1 is None. This may cause security issues.\n"
        )

    return download_url, sha1


def fetch_and_save_version_manifest(dest: Path, target_version: str=None, root_manifest=None) -> dict | None:
    """
    Fetch and save version manifest. (Two types. One is root version manifest, another is specific version manifest)
    :param dest: Destination of the version manifest.
    :param target_version: Target version to fetch manifest from. Set this to a Minecraft version if you want to save
    its version data.
    :param root_manifest: Root manifest to fetch specific version manifest from. If not set the function will auto
     fetch it.
    :return:
        version_manifest: dict
    """
    if root_manifest is None:
        root_manifest = get_version_manifest()

    if target_version:
        # For specified ver
        data, sha1 = fetch_specific_version_manifest(root_manifest, target_version)
    else:
        # For root version manifest (contains all versions)
        data = root_manifest
        sha1 = None

    file = FileObject(
        path=dest,
        sha1=sha1,
    )

    dest.parent.mkdir(parents=True, exist_ok=True)

    try:
        method = "wb" if isinstance(data, bytes) else "w"
        with dest.open(method) as f:
            if isinstance(data, bytes):
                f.write(data)
            elif isinstance(data, dict):
                json.dump(data, f)

        if sha1 is not None and not file.verify_sha1(sha1):
            raise HashMismatchException(
                "sha1",
                file.sha1,
                sha1,
                file.posix_path
            )

        if isinstance(data, bytes):
            # Convert the data back to dict
            return json.loads(data.decode("utf-8"))
        else:
            return data
    except Exception as e:
        logger.exception(e)
        raise VersionManifestSaveException(
            "Error saving version manifest:\n{}".format(e)
        )

def download_core_file(manifest: dict, destination: Path, is_server=False, raise_for_null_hash=False,
                       progress_callback: Callable[str]=None) -> FileObject:
    """
    Download core file from version manifest.
    :param manifest:
    :param destination:
    :param is_server:
    :param raise_for_null_hash:
    :param progress_callback:
    :return:
        client_file: FileObject
    """
    url, sha1 = get_core_file_download_url_and_hash(manifest, is_server=is_server)

    destination.parent.mkdir(parents=True, exist_ok=True)
    file = FileObject(destination, url=url)

    if sha1 is None and raise_for_null_hash:
        raise VersionDataKeyNotFoundException(
            "Core file hash is missing. Enable \"raise_for_null_hash\" flag to ignore this exception.\n"
        )
    elif sha1 is None:
        logger.warning("Client has no sha1.")

    if file.exists and file.verify("sha1", sha1):
        logger.info(f"Client existed and sha1 matches.")
    elif url is not None:
        download_file(file, progress_callback=progress_callback)
    elif not url:
        raise VersionDataKeyNotFoundException(
            "Core file url is None. Unable to download it."
        )

    return file