From 864ce33db18e3929d4a12d5b07b94a7509ced10a Mon Sep 17 00:00:00 2001 From: Wei Hong Date: Sun, 12 Jul 2026 22:43:34 +0800 Subject: [PATCH] All the library (included natives) process are test finished. --- __init__.py | 1 + core_lib/common/common.py | 79 ++-- core_lib/common/exception.py | 48 ++- core_lib/game/asset.py | 17 + core_lib/game/library.py | 384 ++++++++++++++++-- core_lib/game/version_info.py | 285 ++++++++++--- core_lib/qt/hack.py | 5 + dialog/test.py | 744 +++++++++++++++++++--------------- 8 files changed, 1094 insertions(+), 469 deletions(-) create mode 100644 core_lib/game/asset.py diff --git a/__init__.py b/__init__.py index 2820700..dcdf65e 100644 --- a/__init__.py +++ b/__init__.py @@ -14,5 +14,6 @@ This launcher should adheres to the following rules: 3: Be cool and fun 4: No AI slop (Nobody wants to read spaghetti code that made by AI) 5: Freedom just like the bird +6: Don't repeat yourself (originated from Django) """ __website__ = "https://repo.weispace.net/wei/Launcher" \ No newline at end of file diff --git a/core_lib/common/common.py b/core_lib/common/common.py index 319a5f4..0381389 100644 --- a/core_lib/common/common.py +++ b/core_lib/common/common.py @@ -10,7 +10,7 @@ from typing import List import requests -from .exception import HashMismatchException +from .exception import HashMismatchException, DownloadException DEFAULT_HASH_CHUNK_SIZE = 4096 DEFAULT_DOWNLOAD_CHUNK_SIZE = 4096 @@ -149,30 +149,35 @@ def download_file(file: FileObject, overwrite=True, progress_callback: Callable[ file.path.parent.mkdir(parents=True, exist_ok=True) if file.exists and not overwrite: - raise Exception("File {} already exists. Use overwrite=True to overwrite.".format(file)) + raise FileExistsError("File {} already exists. Use overwrite=True to overwrite.".format(file)) if file.url is None: - raise Exception("Target file {} URL cannot be None.".format(file)) + raise ValueError("Target file {} URL cannot be None.".format(file)) logger.debug("Downloading file {} to {}".format(file.url, file)) - with requests.get(file.url, stream=True) as r: - current_chunk = 0 - total_bytes = r.headers.get("content-length", None) - logger.debug("URL: {}".format(file.url)) - logger.debug("Response code: {}".format(r.status_code)) + try: + with requests.get(file.url, stream=True) as r: + current_chunk = 0 + total_bytes = r.headers.get("content-length", None) + logger.debug("URL: {}".format(file.url)) + logger.debug("Response code: {}".format(r.status_code)) - with file.path.open("wb") as f: - for chunk in r.iter_content(chunk_size=DEFAULT_DOWNLOAD_CHUNK_SIZE): - if not chunk: - break + with file.path.open("wb") as f: + for chunk in r.iter_content(chunk_size=DEFAULT_DOWNLOAD_CHUNK_SIZE): + if not chunk: + break - f.write(chunk) - current_chunk += 1 + f.write(chunk) + current_chunk += 1 - if callable(progress_callback) and total_bytes is not None: - current_progress = ((current_chunk * DEFAULT_DOWNLOAD_CHUNK_SIZE) / int(total_bytes) ) * 100 - progress_callback(file.name, current_progress) + if callable(progress_callback) and total_bytes is not None: + current_progress = ((current_chunk * DEFAULT_DOWNLOAD_CHUNK_SIZE) / int(total_bytes)) * 100 + progress_callback(file.name, current_progress) + except Exception as e: + raise DownloadException( + "An error occurred while downloading file {}. {}".format(file.url, e) + ) if file.expected_sha256 and not file.verify_sha256(): raise HashMismatchException("sha256", file.sha256, file.expected_sha256, file.posix_path) @@ -190,17 +195,20 @@ def download_file(file: FileObject, overwrite=True, progress_callback: Callable[ def download_multiple_files(files: List[FileObject], download_callback: Callable[str]=None, max_workers=THREADED_DOWNLOAD_MAX_WORKERS, progress_callback: Callable[str]=None) \ - -> List[FileObject]: + -> tuple[list[FileObject], list[FileObject]]: """ Download multiple files :param download_callback: :param files: :param max_workers: :param progress_callback: - :return: List of file objects that download failed. + :return: + fails: List of file objects that download failed. + successes: List of file objects that download succeeded. """ fails = [] + successes = [] files_count = 0 with ThreadPoolExecutor(max_workers=max_workers) as executor: @@ -212,6 +220,7 @@ def download_multiple_files(files: List[FileObject], download_callback: Callable try: files_count += 1 future.result() + successes.append(file) if callable(download_callback): download_callback(file.posix_path) @@ -222,21 +231,33 @@ def download_multiple_files(files: List[FileObject], download_callback: Callable fails.append(file) - return fails + return fails, successes -def unzip(file: FileObject, destination: Path, target_extract_dir: Path | None=None, exclude: list[str] | None = None): - with zipfile.ZipFile(file.posix_path, "r") as archive: +def unzip(filepath: Path, destination: Path, exclude: list[str] | None = None, flatten: bool=False): + exclude = exclude or [] + + with zipfile.ZipFile(filepath.as_posix(), "r") as archive: + for member in archive.namelist(): + # Ignore exclude file. + if any(member.startswith(pattern) for pattern in exclude): + continue + + if flatten: + member_path = Path(member) + + if member.endswith("/") or not member_path.name: + continue - if target_extract_dir is None and exclude is None: - archive.extractall(path=destination) - else: - for member in archive.namelist(): - # Ignore exclude file. if any(member.startswith(pattern) for pattern in exclude): continue - if member.startswith(target_extract_dir.as_posix()): - archive.extract(member, destination) + target = destination / member_path.name + + with archive.open(member) as source: + with target.open("wb") as output: + output.write(source.read()) + else: + archive.extract(member, path=destination) def get_platform_info(): """ diff --git a/core_lib/common/exception.py b/core_lib/common/exception.py index 2275696..b4f4838 100644 --- a/core_lib/common/exception.py +++ b/core_lib/common/exception.py @@ -2,14 +2,54 @@ from pathlib import Path class LauncherException(Exception): - def __init__(self, message): - self.message = message + user_message = "Launcher error occurred." - def __str__(self): - return self.message + def __init__(self, message=None, *, user_message=None): + super().__init__(message or user_message or self.user_message) + if user_message is not None: + self.user_message = user_message class HashMismatchException(Exception): def __init__(self, hash_name, got_hash, expected_hash, filepath: str | Path): if type(filepath) == Path: filepath = Path(filepath).absolute().as_posix() self.message = "File {} hash mismatch. Expected hash {}. Got hash {} ({})".format(filepath, got_hash, expected_hash, hash_name) + super().__init__(self.message) + +class UnsupportedPlatformException(LauncherException): + user_message = "Current platform or architecture is not supported." + +class VersionNotSelectedException(LauncherException): + user_message = "Select a version before continuing." + +class VersionManifestFetchException(LauncherException): + user_message = "Unable to fetch version manifest." + +class VersionManifestSaveException(LauncherException): + user_message = "Unable to save version manifest." + +class NoVersionFoundException(LauncherException): + user_message = "No version found. (This should never happen)" + +class NoSpecifiedVersionKeyException(LauncherException): + user_message = "Specified version or type not found." + +class VersionDataKeyNotFoundException(LauncherException): + user_message = "Specified version data section (key) not found." + +class VersionManifestException(LauncherException): + user_message = "Unable to load version manifest." + +class DependencyException(LauncherException): + user_message = "Unable to find dependency." + +class NativeResolveException(LauncherException): + user_message = "Unable to resolve native libraries." + + +class DownloadException(LauncherException): + user_message = "Some files failed to download." + +class ExtractException(LauncherException): + user_message = "Unable to extract native libraries." + diff --git a/core_lib/game/asset.py b/core_lib/game/asset.py new file mode 100644 index 0000000..dbd7e9b --- /dev/null +++ b/core_lib/game/asset.py @@ -0,0 +1,17 @@ + + +MOJANG_RESOURCE_ENDPOINT = "https://resources.download.minecraft.net/{}/{}" + + +def fetch_asset_manifest(target_version_manifest: dict) -> dict: + """ + Fetch the asset manifest from target version's manifest. + :param target_version_manifest: + :return: + asset_manifest: dict + """ + + assetIndex = target_version_manifest.get("assetIndex") + + if assetIndex is None: + pass \ No newline at end of file diff --git a/core_lib/game/library.py b/core_lib/game/library.py index cacc786..179a56a 100644 --- a/core_lib/game/library.py +++ b/core_lib/game/library.py @@ -1,43 +1,22 @@ import json import logging import re +from pathlib import Path +from typing import Callable -from ..common.common import get_platform_info +from ..common.common import get_platform_info, FileObject, download_multiple_files, unzip from .version_info import ( MOJANG_LIBRARIES_ENDPOINT, NATIVE_OS_KEY_MAP, NATIVE_OS_RULE_MAP, NATIVE_ARCH_KEY_MAP, NATIVE_NEW_TO_OLD_ARCH, - NATIVE_OLD_TO_NEW_ARCH, + NATIVE_OLD_TO_NEW_ARCH, find_useful_part_in_specific_version_manifest, ) +from ..common.exception import UnsupportedPlatformException, NativeResolveException, DependencyException logger = logging.getLogger("Launcher.CoreLib") -def generate_formatted_artifact_with_name(lib_name, lib_artifact, libraries_endpoint=MOJANG_LIBRARIES_ENDPOINT): - """ - Generates a formatted artifact - :param lib_name: Library name - :param lib_artifact: Type should be a dict, and it must contain the key "path" (relative path) - :param libraries_endpoint: The function will use it and relative path to generate a new URL if artifact does not contain it. - Default value is MOJANG_LIBRARIES_ENDPOINT - :return: - """ - artifact_new = dict(lib_artifact) - artifact_new["name"] = lib_name - - if not artifact_new.get("url") and artifact_new.get("path"): - artifact_new["url"] = ( - libraries_endpoint.rstrip("/") - + "/" - + artifact_new["path"].lstrip("/") - ) - elif not artifact_new.get("url") and not artifact_new.get("path"): - logger.warning(f"Artifact {lib_name} missing relative path. This may cause issues.") - return None - - return artifact_new - def is_library_allowed(rules, platform_rule_name, platform_arch_type, platform_version): """ @@ -116,14 +95,14 @@ def is_native_allowed(lib_name, platform_key_name, platform_arch_type, unspecifi native_os_name = parts[1] if unspecified_arch_policy == "universal": arch_type = platform_arch_type - logger.debug(f"This native may for platform \"{native_os_name}\" (\"May\" for all architecture type)\n") + logger.debug(f"This native may for platform \"{native_os_name}\" (\"May\" for all architecture type)") else: arch_type = unspecified_arch_policy - logger.debug(f"This native may for platform \"{native_os_name}\" and specific architecture type {unspecified_arch_policy}\n") + logger.debug(f"This native may for platform \"{native_os_name}\" and specific architecture type {unspecified_arch_policy}") elif len(parts) == 3: # With architecture type native_os_name, arch_type = parts[1], parts[2] - logger.debug(f"This native may for platform \"{native_os_name}\" and architecture type \"{arch_type}\"\n") + logger.debug(f"This native may for platform \"{native_os_name}\" and architecture type \"{arch_type}\"") else: logger.warning("Unsupported native key '{}'".format(native_key)) return False @@ -140,7 +119,20 @@ def is_native_allowed(lib_name, platform_key_name, platform_arch_type, unspecifi return False -def is_legacy_native_allowed(natives, classifiers, platform_rule_name, platform_arch_type, unspecified_arch_policy: str="x86_64") -> tuple[bool, dict | None]: +def is_legacy_native_allowed(natives, classifiers, platform_rule_name, platform_arch_type, unspecified_arch_policy: str="x86_64") -> tuple[bool, dict | None, str | None]: + valid_non_arch_values = { + "universal", + "x86", + "x86_64", + "arm64", + } + + if unspecified_arch_policy not in valid_non_arch_values: + raise ValueError( + f"Unsupported considered_non_arch: " + f"{unspecified_arch_policy!r}" + ) + classifier = natives.get(platform_rule_name) if classifier: @@ -159,7 +151,7 @@ def is_legacy_native_allowed(natives, classifiers, platform_rule_name, platform_ native_artifact = classifiers.get(classifier) if native_artifact: - return True, native_artifact + return True, native_artifact, classifier # classifiers-rule use NATIVE_OS_RULE_MAP if classifiers: @@ -181,13 +173,13 @@ def is_legacy_native_allowed(natives, classifiers, platform_rule_name, platform_ if unspecified_arch_policy == "universal": universal = True arch_type = platform_arch_type - logger.debug(f"This native may for platform \"{native_os_name}\" (\"May\" for all architecture type)\n") + logger.debug(f"This native may for platform \"{native_os_name}\" (\"May\" for all architecture type)") else: arch_type = unspecified_arch_policy - logger.debug(f"This native may for platform \"{native_os_name}\" and specific architecture type {unspecified_arch_policy}\n") + logger.debug(f"This native may for platform \"{native_os_name}\" and specific architecture type {unspecified_arch_policy}") elif len(parts) == 3: native_os_name, arch_type = parts[1].lower(), parts[2].lower() - logger.debug(f"This native may for platform \"{native_os_name}\" and architecture type \"{arch_type}\"\n") + logger.debug(f"This native may for platform \"{native_os_name}\" and architecture type \"{arch_type}\"") else: logger.warning("Unsupported legacy native key '{}'".format(native)) continue @@ -198,6 +190,326 @@ def is_legacy_native_allowed(natives, classifiers, platform_rule_name, platform_ continue if native_os_name == platform_rule_name and (arch_type_converted == platform_arch_type or universal): - return True, data_raw + return True, data_raw, native - return False, None \ No newline at end of file + return False, None, None + +def generate_relative_path_by_maven_coordinate(maven_coordinate: str, extra=None) -> str | None: + parts = maven_coordinate.split(":") + + if len(parts) not in (3, 4): + logger.warning("Unsupported maven coordinate '{}'".format(maven_coordinate)) + return None + + group, artifact, version = parts[:3] + classifier = extra or (parts[3] if len(parts) == 4 else None) + + base_path = group.replace(".", "/") + filename = f"{artifact}-{version}" + + if classifier: + filename += f"-{classifier}" + + return f"{base_path}/{artifact}/{version}/{filename}.jar" + +def remove_duplicate_artifacts(artifacts: list[dict], ignore_null=False, compare_by_name=False) -> list[dict]: + seen = set() + result = [] + + for artifact in artifacts: + path = artifact.get("path") + name = artifact.get("name") + + if name in seen and compare_by_name: + continue + elif path in seen: + continue + + if name is None or path is None: + logger.warning("Artifact '{}' has no name or path.".format(artifact)) + + if (not ignore_null or name is not None) and compare_by_name: + result.append(artifact) + seen.add(name) + elif not ignore_null or path is not None: + result.append(artifact) + seen.add(path) + + return result + +def generate_formatted_artifact_with_name(lib_name, lib_artifact, libraries_endpoint=MOJANG_LIBRARIES_ENDPOINT, + extra: dict | None=None, relative_args: dict|None=None) -> dict | None: + """ + Generates a formatted artifact + :param lib_name: Library name + :param lib_artifact: Type should be a dict, and it must contain the key "path" (relative path) + :param libraries_endpoint: The function will use it and relative path to generate a new URL if artifact does not contain it. + Default value is MOJANG_LIBRARIES_ENDPOINT + :param extra: Extra part of the artifact (Will be included if exists) + :param relative_args: Arguments of the generate_relative_path_by_maven_coordinate (Will be called when relative path + is missing) + :return: artifact_new: dict + """ + artifact_new = dict(lib_artifact) + artifact_new["name"] = lib_name + + if not artifact_new.get("path"): + logger.warning(f"Artifact {lib_name} missing relative path. Trying to generate new relative path...") + if relative_args is None: relative_args = {} + relative_path = generate_relative_path_by_maven_coordinate(lib_name, **relative_args) + if relative_path: + artifact_new["path"] = relative_path + logger.debug(f"Artifact {lib_name} relative path: {relative_path}") + else: + logger.error(f"Unable to generate new relative path for artifact {lib_name}") + return None + + if not artifact_new.get("url") and artifact_new.get("path"): + artifact_new["url"] = ( + libraries_endpoint.rstrip("/") + + "/" + + artifact_new["path"].lstrip("/") + ) + + if extra: + artifact_new.update(extra) + + return artifact_new + +def collect_library_artifacts(specific_version_manifest): + _, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(specific_version_manifest) + + items = [] + + # Same as natives + _, _, os_version = get_platform_info() + platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info() + + for library in libraries: + name = library.get("name", "") + artifact_raw = library.get("downloads", {}).get("artifact", {}) + rules = library.get("rules", []) + classifiers = library.get("downloads", {}).get("classifiers", {}) + natives = library.get("natives", {}) + + # Ignore natives + if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or + any(native.startswith("natives-") for native in classifiers)): + logger.info(f"Skipping {name} because its a native library.") + continue + + if not is_library_allowed( + rules, + platform_rule_name, + platform_arch_type, + os_version, + ): + logger.debug(f"Library {name} can't pass library rules.") + continue + + artifact = generate_formatted_artifact_with_name( + name, + artifact_raw + ) + + if artifact: + items.append(artifact) + + if not items: + raise DependencyException( + "Could not find any artifact in any of the libraries." + ) + + return items + + +def collect_native_artifacts(specific_version_manifest): + """ + Collects native artifacts that support current platform + :param specific_version_manifest: + :return: + filtered_natives_artifacts: list[dict] + """ + filtered = [] + + # Convert current platform and architecture type to same formula + _, _, os_version = get_platform_info() + platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info() + + if platform_key_name is None or platform_arch_type is None or platform_rule_name is None: + raise UnsupportedPlatformException( + f"Unsupported platform: key={platform_key_name}, " + f"rule={platform_rule_name}, arch={platform_arch_type}" + ) + + _, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(specific_version_manifest) + + for library in libraries: + name = library.get("name", "") + artifact = library.get("downloads", {}).get("artifact", {}) + # For newer Minecraft (VER>1.7.10) + rules = library.get("rules", []) + # For legacy version (VER<1.8) + classifiers = library.get("downloads", {}).get("classifiers", {}) + natives = library.get("natives", {}) + exclude = library.get("extract", {}).get("exclude", ["META-INF/"]) # For extract use + + is_natives = False + + # Check if library is a native + if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or + any(native.startswith("natives-") for native in classifiers)): + is_natives = True + + if not is_natives: + continue + + if rules and classifiers: + logger.debug(f"Found newer and legacy (classifier) native rules for {name}") + elif rules: + logger.debug(f"Found newer native rules for {name}") + elif classifiers: + logger.debug(f"Found legacy native rules (classifier) for {name}") + + # =============== New rules =============== + + # Parse action section + if not is_library_allowed( + rules, + platform_rule_name, + platform_arch_type, + os_version, + ): + logger.debug(f"Library {name} can't pass library rules.") + continue + + if is_native_allowed( + name, + platform_key_name, + platform_arch_type, + ): + result_artifact = generate_formatted_artifact_with_name(name, artifact, extra={"exclude": exclude, + "flatten": True}) + if result_artifact: + filtered.append(result_artifact) + # For some reason, we still need to continue process because some version use multiple type of + # classifier. (Such as 1.14.2) + else: + logger.debug(f"Library {name} can't pass native rules.") + + # =============== Old rules =============== + logger.debug( + "Library {} doesn't have native key. This library may use old rule.".format(name) + ) + + result, result_artifact, classifier = is_legacy_native_allowed( + natives, + classifiers, + platform_rule_name, + platform_arch_type, + ) + + if not result: + logger.debug(f"Library {name} can't pass legacy native rules.") + continue + else: + result_artifact = generate_formatted_artifact_with_name(name, + result_artifact, + extra={"exclude": exclude, "flatten": True}, + relative_args={ + "extra": classifier + }) + if result_artifact: + filtered.append(result_artifact) + + # Remove duplicate items + items = remove_duplicate_artifacts(filtered) + + if not filtered: + raise NativeResolveException( + "No compatible native artifact found.", + user_message="No compatible native library was found for this platform.", + ) + elif not items: + raise NativeResolveException( + "No compatible native artifact found. Report developer because this may be a issue" + " at function \"remove_duplicate_artifacts\"", + ) + + return items + +def download_libraries(artifacts, output_dir, progress_callback: Callable[[str], None]): + """ + Download libraries from a list that contains artifacts + :param artifacts: + :param output_dir: + :param progress_callback: + :return: + fails: list (List of artifacts that download failed) + successes: list (artifacts that were downloaded) + """ + files = [] + + for artifact in artifacts: + sha1 = artifact.get("sha1") + relative_path = artifact.get("path") + url = artifact.get("url") + + if not relative_path or not url: + logger.warning( + "Artifact {} has no path or url. Skipping.".format(artifact["name"]) + ) + continue + else: + full_path = output_dir / Path(relative_path) + + file = FileObject( + full_path, + url=url, + sha1=sha1, + ) + + if sha1 is None: + logger.warning(f"File {full_path} has no SHA1 hash.") + + if file.exists and (sha1 is None or file.verify("sha1", sha1)): + logger.debug(f"File {full_path} exists.") + continue + + files.append(file) + + fails, successes = download_multiple_files(files, progress_callback=progress_callback) + + return fails, successes + + +def unzip_natives(filtered, libraries_dir, destination: Path): + """ + Unzip native libraries from a list that contains artifacts + :param filtered: + :param libraries_dir: + :param destination: + :return: + + IMPORTANT: + Please put "exclude" key inside the artifact (from the version data). If not, some unnecessary file will be also + included when unzipping the native library. + """ + destination.mkdir(parents=True, exist_ok=True) + + for result_artifact in filtered: + exclude = result_artifact.get("exclude", []) + flatten = result_artifact.get("flatten", False) + relative_path = result_artifact.get("path") + + if not relative_path: + logger.warning( + "Artifact {} has no relative path. Skipping.".format(result_artifact["name"]) + ) + continue + else: + full_path = libraries_dir / Path(relative_path) + + unzip(full_path, destination, exclude=exclude, flatten=flatten) + + logger.debug(f"Unzipped {full_path}.") diff --git a/core_lib/game/version_info.py b/core_lib/game/version_info.py index ec1c34b..7f77417 100644 --- a/core_lib/game/version_info.py +++ b/core_lib/game/version_info.py @@ -1,8 +1,16 @@ import json import logging +import warnings +from pathlib import Path + import requests +from ..common.common import FileObject +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 = { @@ -57,54 +65,79 @@ NATIVE_NEW_TO_OLD_ARCH = { 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: - logger.error("Error getting version manifest from {}".format(manifest_url)) - logger.error("Exception: {}".format(e)) - return None + 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: - logger.error("Unable to deserialize response data into dictionary.") - logger.error("URL: {}".format(manifest_url)) - logger.error("Exception: {}".format(e)) - return None - + 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: - logger.debug("No latest version section available in version manifest.") - return None + 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: - logger.debug("No latest version available.") + 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, specific_type=None) -> list: +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: - logger.debug("No version section available in version manifest.") - return [] + raise NoVersionFoundException( + "No version section available in version manifest." + ) - if not specific_type: + if not specified_type: target_type = "release" if not is_snapshot else "snapshot" else: - target_type = specific_type + target_type = specified_type filtered = [] @@ -118,16 +151,23 @@ def get_all_versions(version_manifest: dict, is_snapshot=False, filtered.append(ver) if len(filtered) == 0: - logger.debug("No version matches specified type: {}".format(target_type)) + 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: - logger.debug("No version section available in version manifest.") - return [] + raise NoVersionFoundException( + "No version section available in version manifest." + ) types = [] @@ -139,35 +179,50 @@ def get_available_version_types(version_manifest: dict) -> list: return types -def get_specific_version_manifest_url(version_manifest: dict, - version_id: str) -> str | None: +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: - logger.debug("No version section available in version manifest.") - return None + 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: - logger.debug("Specific version manifest URL not available in version manifest.") - return None + raise VersionManifestFetchException( + "Specific version manifest URL not available in version manifest." + ) return specific_ver_manifest_url - logger.debug("No version section available in version manifest.") + raise NoSpecifiedVersionKeyException( + "Specified version {} manifest URL not available in version manifest.".format(version_id) + ) - return None - -def get_specific_version_manifest_url_and_hash(version_manifest: dict, - version_id: str) -> tuple[str,str] | tuple[None, None]: +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: - logger.debug("No version section available in version manifest.") - return None, None + raise NoVersionFoundException( + "No version section available in version manifest." + ) for version in versions: if version.get("id") == version_id: @@ -175,91 +230,189 @@ def get_specific_version_manifest_url_and_hash(version_manifest: dict, specific_ver_manifest_hash = version.get("hash", None) if specific_ver_manifest_url is None: - logger.debug("Specific version manifest URL not available in version manifest.") + 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 - logger.debug("No version section available in version manifest.") + raise NoSpecifiedVersionKeyException( + "Specified version {} manifest URL not available in version manifest.".format(version_id) + ) - return None, None -def fetch_specific_version_manifest(version_manifest: dict, spec_version: str) -> dict | None: - ver_url = get_specific_version_manifest_url(version_manifest, spec_version) - - if ver_url is None: - logger.debug("Unable to get specific version manifest url from version manifest.\n" - "Version: {}".format(spec_version)) - return None +def fetch_specific_version_manifest(version_manifest: dict, spec_version: str) -> tuple[dict, str | None]: + """ + Get specific version's manifest data and hash from version manifest. + :param version_manifest: + :param spec_version: + :return: + specific_version_manifest_data: str + 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: - logger.error("Error getting version manifest from {}".format(ver_url)) - logger.error("Exception: {}".format(e)) - return None + raise VersionManifestFetchException( + "Failed to fetch version manifest URL: {}\n{}".format(ver_url, e) + ) try: data = r.json() logger.debug("SpecificVersionManifest: {}".format(json.dumps(data, indent=4))) - return data + return data, sha1 except ValueError as e: - logger.error("Unable to deserialize response data into dictionary.") + raise VersionManifestFetchException( + "Unable to deserialize response data into dictionary: {} (TargetVer: {})".format(e, ver_url) + ) +# 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]: + -> 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 len(arguments) == 0: - logger.debug("No arguments section available in version manifest.") + 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: - logger.debug("No asset index section available in version manifest.") + raise VersionDataKeyNotFoundException( + "No asset index section available in version manifest. (\"assetIndex\")" + ) if len(downloads) == 0: - logger.debug("No downloads section available in version manifest.") + raise VersionDataKeyNotFoundException( + "No downloads section available in version manifest. (\"downloads\")" + ) if len(java_version) == 0: - logger.debug("No java version section available in version manifest.") + raise VersionDataKeyNotFoundException( + "No java version section available in version manifest. (\"javaVersion\")" + ) if len(libraries) == 0: - logger.debug("No libraries section available in version manifest.") + raise VersionDataKeyNotFoundException( + "No libraries section available in version manifest. (\"libraries\")" + ) if main_class is None: - logger.debug("No main class section available in version manifest.") + raise VersionDataKeyNotFoundException( + "No main class section available in version manifest. (\"mainClass\")" + ) - return arguments, asset_index, downloads, java_version, libraries, main_class + 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=False) -> tuple[str, str] | tuple[None, None]: +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: - logger.debug("No downloads section available in version manifest.") - return None, None + 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: - logger.debug("No target core section available in version manifest.\n" - "Wants: {}".format(core_name)) - return None, None + 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: - logger.debug("Core file url is None.\n") + raise VersionDataKeyNotFoundException( + "Core file url is None.\n" + ) if sha1 is None: - logger.debug("Core file sha1 is None.\n") + warnings.warn( + "Core file sha1 is None. This may cause security issues.\n" + ) - return download_url, sha1 \ No newline at end of file + 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: + with dest.open("w") as f: + json.dump(data, f, indent=4) + + if sha1 is not None and not file.verify_sha1(sha1): + raise HashMismatchException( + "sha1", + file.sha1, + sha1, + file.posix_path + ) + + return data + except Exception as e: + raise VersionManifestSaveException( + "Error saving version manifest:\n{}".format(e) + ) \ No newline at end of file diff --git a/core_lib/qt/hack.py b/core_lib/qt/hack.py index 0afd46a..1e2fb90 100644 --- a/core_lib/qt/hack.py +++ b/core_lib/qt/hack.py @@ -1,3 +1,4 @@ +from PySide6.QtCore import QCoreApplication, QThread from PySide6.QtGui import QScreen from PySide6.QtWidgets import QMainWindow, QApplication @@ -8,3 +9,7 @@ def move_window_to_center(window: QMainWindow): geometry = window.frameGeometry() geometry.moveCenter(center) window.move(geometry.topLeft()) + +def is_main_thread() -> bool: + app = QCoreApplication.instance() + return app is not None and QThread.currentThread() == app.thread() \ No newline at end of file diff --git a/dialog/test.py b/dialog/test.py index a91a8e9..be9affc 100644 --- a/dialog/test.py +++ b/dialog/test.py @@ -1,19 +1,27 @@ -import re +import json import textwrap import threading +import time from pathlib import Path -from PySide6.QtCore import Qt, QObject, Signal +from PySide6.QtCore import Qt, QObject, Signal, Slot from PySide6.QtWidgets import QPushButton, QVBoxLayout, QDialog, QPlainTextEdit, \ - QHBoxLayout, QComboBox + QHBoxLayout, QComboBox, QMessageBox -from core_lib.common.common import download_file, FileObject, download_multiple_files, get_platform_info -from core_lib.game.library import generate_formatted_artifact_with_name, get_natives_platform_info, is_library_allowed, \ - is_native_allowed, is_legacy_native_allowed -from core_lib.game.version_info import get_version_manifest, \ - fetch_specific_version_manifest, find_useful_part_in_specific_version_manifest, get_core_file_download_url_and_hash, \ - get_all_versions, get_available_version_types, get_specific_version_manifest_url_and_hash +from core_lib.common.common import download_file, FileObject, download_multiple_files +from core_lib.common.exception import VersionNotSelectedException, UnsupportedPlatformException, \ + NativeResolveException, DownloadException, DependencyException, VersionManifestFetchException, \ + VersionManifestSaveException, NoSpecifiedVersionKeyException +from core_lib.game.library import collect_native_artifacts, download_libraries, unzip_natives, collect_library_artifacts + +from core_lib.game.version_info import (find_useful_part_in_specific_version_manifest, + get_core_file_download_url_and_hash, \ + get_all_versions, get_available_version_types, \ + fetch_and_save_version_manifest, get_specific_version_manifest_url_and_hash) from core_lib.qt import messagebox +from core_lib.qt.hack import is_main_thread + +MAX_DOWNLOAD_ATTEMPTS = 3 class TestDialog(QDialog): @@ -28,10 +36,6 @@ class TestDialog(QDialog): self.button.setObjectName("ActionButton") self.button.clicked.connect(self.run_test) - self.button_6 = QPushButton("Download Version Data") - self.button_6.setObjectName("ActionButton") - self.button_6.clicked.connect(self.run_download_specific_version_manifest) - self.button_2 = QPushButton("Download client") self.button_2.setObjectName("ActionButton") self.button_2.clicked.connect(self.run_download_client) @@ -40,13 +44,9 @@ class TestDialog(QDialog): self.button_3.setObjectName("ActionButton") self.button_3.clicked.connect(self.run_download_libraries) - self.button_4 = QPushButton("Download and unzip natives") + self.button_4 = QPushButton("Refresh versions") self.button_4.setObjectName("ActionButton") - self.button_4.clicked.connect(lambda : self.run_download_and_unzip_natives()) - - self.button_5 = QPushButton("Refresh versions") - self.button_5.setObjectName("ActionButton") - self.button_5.clicked.connect(self.load_all_version) + self.button_4.clicked.connect(self.load_all_version) self.ver_dropdown = QComboBox() self.ver_dropdown.setObjectName("Dropdown") @@ -68,14 +68,12 @@ class TestDialog(QDialog): self.right_layout.setObjectName("RightLayout") self.right_layout.setAlignment(Qt.AlignmentFlag.AlignRight) self.right_layout.addWidget(self.button) - self.right_layout.addWidget(self.button_6) self.right_layout.addWidget(self.button_2) self.right_layout.addWidget(self.button_3) - self.right_layout.addWidget(self.button_4) self.right_layout.addStretch() self.right_layout.addWidget(self.ver_type_dropdown) self.right_layout.addWidget(self.ver_dropdown) - self.right_layout.addWidget(self.button_5) + self.right_layout.addWidget(self.button_4) self.right_layout.setContentsMargins(0, 10, 0, 0) self.layout.addLayout(self.right_layout) @@ -133,6 +131,14 @@ class TestDialog(QDialog): ) ) + self.widget_signal.show_warning_message.connect( + lambda msg: messagebox.warning( + self.parent, + "Warning", + message=msg + ) + ) + self.widget_signal.show_info_message.connect( lambda msg: messagebox.info( self.parent, @@ -149,13 +155,29 @@ class TestDialog(QDialog): lambda widget: widget.setDisabled(True) ) + self.widget_signal.askyesno.connect( + self.ask_request + ) + + self.widget_signal.get_selected_ver.connect( + self.get_selected_ver + ) + + self.widget_signal.get_selected_type.connect( + self.get_selected_type + ) + + self.widget_signal.clean_output.connect( + lambda : self.output.clear() + ) + self.load_all_types() class DownloadSignal(QObject): log = Signal(str) finished = Signal(int) failed = Signal(str) - progress = Signal(str, int) + progress = Signal(str, float) class WidgetSignal(QObject): clean_ver_types_dropdown = Signal() @@ -163,9 +185,47 @@ class TestDialog(QDialog): update_vers_dropdown = Signal(list) update_types_dropdown = Signal(list) show_error_message = Signal(str) + show_warning_message = Signal(str) show_info_message = Signal(str) enable_widget = Signal(object) disable_widget = Signal(object) + askyesno = Signal(dict) # {"title", "targetEvent", "message", "result"} + get_selected_ver = Signal(dict) + get_selected_type = Signal(dict) + clean_output = Signal() + + @Slot(dict) + def ask_request(self, context: dict): + title = context.get("title", "Info") + message = context.get("message", "No content") + event = context.get("event", None) + msg = QMessageBox.question( + self, + title, + message, + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.Yes, + ) + + if msg == QMessageBox.StandardButton.Yes and type(event) == threading.Event: + context["result"] = True + + if isinstance(event, threading.Event): + event.set() + + @Slot(dict) + def get_selected_ver(self, context: dict): + context["version"] = self.ver_dropdown.currentText() + event = context.get("event", None) + if isinstance(event, threading.Event): + event.set() + + @Slot(dict) + def get_selected_type(self, context: dict): + context["type"] = self.ver_type_dropdown.currentText() + event = context.get("event", None) + if isinstance(event, threading.Event): + event.set() @staticmethod def get_progress(value): @@ -199,84 +259,112 @@ class TestDialog(QDialog): if len(self.ver_dropdown.currentText()) == 0: self.ver_dropdown.setCurrentIndex(0) + def _get_selected_version_unsafe(self): + if is_main_thread(): + ver = self.ver_dropdown.currentText() + else: + event = threading.Event() + context = { + "event": event, + } + self.widget_signal.get_selected_ver.emit(context) + + if not event.wait(10): + raise VersionNotSelectedException( + "Timed out while waiting for selected version from the UI thread.", + user_message="Unable to read selected version from the UI. Please try again." + ) + + ver = context.get("version", None) + + if not ver: + raise VersionNotSelectedException() + + return ver + + def get_selected_version(self): + try: + ver = self._get_selected_version_unsafe() + return ver + except VersionNotSelectedException as e: + self.widget_signal.show_error_message.emit( + "Select a version before continuing.\n" + "Details: {}".format(e.user_message) + ) + return None + def run_test(self): self.button.setDisabled(True) - self.output.clear() - data = get_version_manifest() - ver = self.ver_dropdown.currentText() + try: + self.output.clear() + ver = self.get_selected_version() - if len(ver) == 0 or ver is None: - messagebox.error( - self.parent, - "Error", - "Please select a version before running the test.", + if not ver: + return + + ver_data = self.get_cached_version_manifest(ver) + + arguments, asset_index, downloads, java_version, libraries, main_class, legacy_game_args = ( + find_useful_part_in_specific_version_manifest(ver_data)) + + content = textwrap.dedent(f"""\ + Arguments: {arguments}\n + + Asset: {asset_index}\n + + Downloads: {downloads}\n + + Java Version: {java_version}\n + + Libraries: {libraries}\n + + Main Class: {main_class}\n + + Extra: + minecraftArguments: {legacy_game_args} + """) + self.output.setPlainText(content) + except Exception as e: + self.widget_signal.show_error_message.emit( + "Failed to run test. Details: {}".format(e) ) - - ver_data = fetch_specific_version_manifest(data, ver) - arguments, asset_index, downloads, java_version, libraries, main_class = ( - find_useful_part_in_specific_version_manifest(ver_data)) - - content = textwrap.dedent(f"""\ - Arguments: {arguments}\n - - Asset: {asset_index}\n - - Downloads: {downloads}\n - - Java Version: {java_version}\n - - Libraries: {libraries}\n - - Main Class: {main_class}\n - """) - - self.output.setPlainText(content) - self.button.setDisabled(False) - - def run_download_specific_version_manifest(self): - self.button_6.setDisabled(True) - - self.output.clear() - - thread = threading.Thread( - target=self._download_specific_version_manifest, - daemon=True, - ) - thread.start() + finally: + self.button.setDisabled(False) def run_download_client(self): self.button_2.setDisabled(True) self.output.clear() + ver = self.get_selected_version() + + if not ver: + self.widget_signal.enable_widget.emit(self.button_2) + return + thread = threading.Thread( target=self._download_client, daemon=True, + args=(ver,) ) thread.start() - # self.output.setPlainText("Client downloaded." if dest.exists() else "Client not found.") - def run_download_libraries(self): self.button_3.setDisabled(True) self.output.clear() + ver = self.get_selected_version() + + if not ver: + self.widget_signal.enable_widget.emit(self.button_3) + return + thread = threading.Thread( target=self._download_libraries_process, daemon=True, - ) - thread.start() - - def run_download_and_unzip_natives(self): - self.button_4.setDisabled(True) - - self.output.clear() - - thread = threading.Thread( - target=self._download_and_unzip_natives, - daemon=True, + args=(ver,) ) thread.start() @@ -302,297 +390,285 @@ class TestDialog(QDialog): # All functions below that must be run in background thread def _load_all_types(self): self.widget_signal.clean_ver_types_dropdown.emit() - data = get_version_manifest() + + data = self.get_cached_version_manifest() + if not data: + return + ver_types = get_available_version_types(data) self.widget_signal.update_types_dropdown.emit(ver_types) def _load_all_version(self, ver_type): self.widget_signal.clean_vers_dropdown.emit() - data = get_version_manifest() + + data = self.get_cached_version_manifest() + if not data: + return + vers = get_all_versions( data, - specific_type=ver_type + specified_type=ver_type ) self.widget_signal.update_vers_dropdown.emit(vers) - def _download_specific_version_manifest(self, ver=None): - data = get_version_manifest() + # + # ============ Version Data Cache ============ + # + @property + def version_manifest_path(self) -> Path: + return Path(self.app.work_dir, "versions", "version_manifest_v2.json") - if not ver: - ver = self.ver_dropdown.currentText() + def get_specific_version_manifest_path(self, target_version: str) -> Path | None: + return Path(self.app.work_dir, "versions", target_version, f"{target_version}.json") - if not ver: - self.widget_signal.show_error_message.emit( - "Select a version before download the version manifest." - ) - return - - ver_manifest_url, sha1 = get_specific_version_manifest_url_and_hash(data, ver) - - if ver_manifest_url is None: - self.widget_signal.show_error_message.emit( - f"Unable to get the version {ver} manifest url." - ) - return + def switch_between_specific_path(self, specified_ver: str | None = None): + if specified_ver is None: + path = self.version_manifest_path else: - self.download_signal.log.emit(f"Version {ver} manifest url: {ver_manifest_url} sha1: {sha1}") + path = self.get_specific_version_manifest_path(specified_ver) - filepath = Path(self.app.work_dir, "versions", ver, f"{ver}.json") + return path - manifest_file = FileObject( - filepath, - url=ver_manifest_url, - sha1=sha1, - ) - - if not manifest_file.exists or manifest_file.verify("sha1", sha1): - download_file(manifest_file, progress_callback=self.download_signal.progress.emit) - - self.download_signal.log.emit(f"File {filepath} downloaded.") - self.widget_signal.enable_widget.emit(self.button_6) - - def _download_client(self, ver=None): - data = get_version_manifest() - - if not ver: - ver = self.ver_dropdown.currentText() - - if not ver: + def _cache_version_manifest(self, specified_ver: str | None = None, root_manifest: dict | None = None) \ + -> tuple[bool, dict | None]: + data = None + path = self.switch_between_specific_path(specified_ver) + try: + data = fetch_and_save_version_manifest(path, target_version=specified_ver, root_manifest=root_manifest) + except VersionManifestFetchException as e: self.widget_signal.show_error_message.emit( - "Select a version before download the client." + "Unable to fetch version manifest: {}".format(e.user_message), ) - return - - ver_data = fetch_specific_version_manifest(data, ver) - url, sha1 = get_core_file_download_url_and_hash(ver_data) - - dest = Path(self.app.work_dir, "versions", ver, "client.jar") - - file = FileObject(dest, url=url) - - if sha1 is None: - self.app.logger.warning("Client has no sha1.") - - if file.exists and file.verify("sha1", sha1): - self.download_signal.log.emit(f"Client existed and sha1 matches.") - elif url is not None: - download_file(file, progress_callback=self.download_signal.progress.emit) - elif not url: - self.download_signal.failed.emit("Download URL not found.") - - self.download_signal.log.emit(f"Downloaded URL: {url}") - self.widget_signal.enable_widget.emit(self.button_2) - - def _download_libraries_process(self, ver=None): - self.output.clear() - - def log_file_downloaded(filepath): - self.download_signal.log.emit(f"Downloaded {filepath}.") - - data = get_version_manifest() - - if not ver: - ver = self.ver_dropdown.currentText() - - if not ver: + except VersionManifestSaveException as e: self.widget_signal.show_error_message.emit( - "Select a version before download libraries." + "Unable to save version manifest: {}".format(e.user_message), + ) + except Exception as e: + self.widget_signal.show_error_message.emit( + "An unexpected error occurred while caching version manifest: {}".format(e), ) - self.widget_signal.enable_widget.emit(self.button_3) - return - ver_data = fetch_specific_version_manifest(data, ver) - _, _, _, _, libraries, _ = find_useful_part_in_specific_version_manifest(ver_data) + return True if data else False, data - output_dir = Path(self.app.work_dir, "libraries") + def _read_cached_version_manifest(self, specified_ver: str = None) -> dict | None: + path = self.switch_between_specific_path(specified_ver) + if not path.exists(): + self.widget_signal.show_error_message.emit( + "Version manifest missing after check. Please try again later.", + ) + return None - output_dir.mkdir(parents=True, exist_ok=True) + try: + version_manifest = json.loads(path.read_text()) + return version_manifest + except Exception as e: + self.widget_signal.show_error_message.emit( + "Unable to read version manifest:\n{}".format(e), + ) + return None - items = [] + def get_cached_version_manifest(self, specified_ver: str = None, max_age: int = 3600) -> dict | None: + path = self.switch_between_specific_path(specified_ver) + file = FileObject(path) + sha1 = None + root_manifest = None - # Same as natives - _, _, os_version = get_platform_info() - platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info() + if specified_ver: + root_manifest = self.get_cached_version_manifest() - for library in libraries: - name = library.get("name", "") - url = library.get("downloads", {}).get("artifact", {}).get("url") - sha1 = library.get("downloads", {}).get("artifact", {}).get("sha1") - relative_path = library.get("downloads", {}).get("artifact", {}).get("path") - rules = library.get("rules", []) - full_path = output_dir / relative_path - classifiers = library.get("downloads", {}).get("classifiers", {}) - natives = library.get("natives", {}) + if root_manifest is None: + return None - # Ignore natives - if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or - any(native.startswith("natives-") for native in classifiers)): - continue + try: + _, sha1 = get_specific_version_manifest_url_and_hash(root_manifest, specified_ver) + except NoSpecifiedVersionKeyException as e: + self.widget_signal.show_error_message.emit( + f"Version {specified_ver} does not exist in version manifest.\n{e.user_message}" + ) + return None - if not url: - continue + if file.exists: + age = time.time() - path.stat().st_mtime - if not is_library_allowed( - rules, - platform_rule_name, - platform_arch_type, - os_version, - ): - self.app.logger.debug(f"Library {name} can't pass library rules.") - continue + # For root manifest + if not specified_ver and age < max_age: + return self._read_cached_version_manifest() - file = FileObject(full_path, url=url, sha1=sha1) + # Below are all for specific ver + if specified_ver and sha1 is None: + self.app.logger.warning("Specified version '{}' hash doesn't exist.".format(specified_ver)) + + # Also check sha1 + if specified_ver and age < max_age and (sha1 is None or file.verify("sha1", sha1)): + return self._read_cached_version_manifest(specified_ver) + + result, data = self._cache_version_manifest(specified_ver, root_manifest=root_manifest) + + if not result: + return None + + return data + + # + # ============ Game Files ============ + # + # INFO: All are not in main thread + def _download_client(self, ver: str): + try: + ver_data = self.get_cached_version_manifest(ver) + + if not ver_data: + return + + url, sha1 = get_core_file_download_url_and_hash(ver_data) + + dest = Path(self.app.work_dir, "versions", ver, "client.jar") + + file = FileObject(dest, url=url) if sha1 is None: - self.app.logger.warning(f"File {full_path} has no SHA1 hash.") + self.app.logger.warning("Client has no sha1.") - if file.exists and (sha1 is None or file.verify("sha1", sha1)): - log_file_downloaded(file.posix_path) - continue + if file.exists and file.verify("sha1", sha1): + self.download_signal.log.emit(f"Client existed and sha1 matches.") + elif url is not None: + download_file(file, progress_callback=self.download_signal.progress.emit) + elif not url: + self.download_signal.failed.emit("Download URL not found.") - items.append(file) - - try: - fails = download_multiple_files(items, - progress_callback=self.download_signal.progress.emit) - self.download_signal.finished.emit(len(fails)) - except Exception as e: - self.download_signal.failed.emit(e) - - self.widget_signal.enable_widget.emit(self.button_3) - - def _download_and_unzip_natives(self, ver=None, unspecified_arch_policy: str= "x86_64"): - valid_non_arch_values = { - "universal", - "x86", - "x86_64", - "arm64", - } - - if unspecified_arch_policy not in valid_non_arch_values: - raise ValueError( - f"Unsupported considered_non_arch: " - f"{unspecified_arch_policy!r}" - ) - - try: - self.output.clear() - data = get_version_manifest() - - if not ver: - ver = self.ver_dropdown.currentText() - - if not ver: - self.widget_signal.show_error_message.emit( - "Select a version before download natives." - ) - self.widget_signal.enable_widget.emit(self.button_3) - return - - filtered = [] - - # Convert current platform and architecture type to same formula - _, _, os_version = get_platform_info() - platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info() - - if platform_key_name is None or platform_arch_type is None or platform_rule_name is None: - self.app.logger.error("Platform or arch are not supported.") - self.widget_signal.show_error_message.emit("Current platform or arch are not supported.") - return - - ver_data = fetch_specific_version_manifest(data, ver) - _, _, _, _, libraries, _ = find_useful_part_in_specific_version_manifest(ver_data) - - for library in libraries: - name = library.get("name", "") - artifact = library.get("downloads", {}).get("artifact", {}) - # For newer Minecraft (VER>1.7.10) - rules = library.get("rules", []) - # For legacy version (VER<1.8) - classifiers = library.get("downloads", {}).get("classifiers", {}) - natives = library.get("natives", {}) - - result_artifact = None - is_natives = False - - # Check if library is a native - if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or - any(native.startswith("natives-") for native in classifiers)): - is_natives = True - - if not is_natives: - continue - - content = f"Name: {name}\nRaw Rules: {rules}\n\n" - - if rules and classifiers: - content += f"This native use multiple native rules!\n" - elif rules: - content += f"This native use new native rules!\n" - elif classifiers: - content += f"This native use legacy native rules!\n" - - # =============== New rules =============== - - # Parse action section - content += "Rules:\n" - - if not is_library_allowed( - rules, - platform_rule_name, - platform_arch_type, - os_version, - ): - self.app.logger.debug(f"Library {name} can't pass library rules.") - continue - - if is_native_allowed( - name, - platform_key_name, - platform_arch_type, - ): - result_artifact = generate_formatted_artifact_with_name(name, artifact) - filtered.append(result_artifact) - # For some reason, we still need to continue process because some version use multiple type of - # classifier. (Such as 1.14.2) - else: - self.app.logger.debug(f"Library {name} can't pass native rules.") - - # =============== Old rules =============== - self.app.logger.debug( - "Library {} doesn't have native key. This library may use old rule.".format(name) - ) - - result, result_artifact = is_legacy_native_allowed( - natives, - classifiers, - platform_rule_name, - platform_arch_type, - ) - - if not result: - self.app.logger.debug(f"Library {name} can't pass legacy native rules.") - continue - else: - result_artifact = generate_formatted_artifact_with_name(name, result_artifact) - - if result_artifact: - filtered.append(result_artifact) - - self.output.appendPlainText( - content - ) - self.app.logger.debug(content) - - self.output.appendPlainText( - "Supported native count: {}".format(len(filtered)) - ) - except Exception as exc: - self.app.logger.exception( - "Unexpected exception occurred: {}".format(exc) - ) + self.download_signal.log.emit(f"Downloaded URL: {url}") + except DownloadException as e: self.widget_signal.show_error_message.emit( - "Unexpected exception occurred while finding natives: {}".format(exc) + "Unable to download client while downloading it: {}".format(e.user_message), + ) + except Exception as e: + self.widget_signal.show_error_message.emit( + "Unable to download client: {}".format(e), ) finally: - self.button_4.setDisabled(False) + self.widget_signal.enable_widget.emit(self.button_2) + + def _download_libraries_process(self, ver: str): + output_dir = Path(self.app.work_dir, "libraries") + output_dir.mkdir(parents=True, exist_ok=True) + + try: + self.widget_signal.clean_output.emit() + + ver_data = self.get_cached_version_manifest(ver) + + if not ver_data: + return + + # libraries + libraries = collect_library_artifacts(ver_data) + + # natives + natives = collect_native_artifacts(ver_data) + + # all + full = list(libraries) + full.extend(natives) + + self.download_signal.log.emit( + "Supported native count: {}".format(len(natives)) + ) + + # Start download + fails, successes = download_libraries(full, output_dir, + progress_callback=self.download_signal.progress.emit) + + download_ok, fails = self.redownload_files(fails) + + if not download_ok: + self.widget_signal.show_error_message.emit( + "Unable to continue process due to certain libraries download failed:\n" + + "\n".join(file.posix_path for file in fails) + ) + else: + # Start unzip natives + natives_dest = Path(self.app.work_dir, "versions", ver, "natives") + unzip_natives(natives, output_dir, natives_dest) + except DependencyException as e: + self.widget_signal.show_error_message.emit( + "Unable to collect library artifacts.\n" + f"{e.user_message}" + ) + except DownloadException as e: + self.widget_signal.show_error_message.emit( + "Unable to collect library artifacts. An download error occurred:\n" + f"{e.user_message}" + ) + except UnsupportedPlatformException as e: + self.widget_signal.show_error_message.emit( + "Unable to collect native artifacts.\n" + f"{e.user_message}" + ) + except NativeResolveException as e: + self.widget_signal.show_error_message.emit( + "Unable to collect native artifacts.\n" + f"{e.user_message}" + ) + except Exception as e: + self.widget_signal.show_error_message.emit( + "Unable to collect library artifacts. An unknown error occurred:\n" + f"{str(e) + f"\n{type(e)}"}" + ) + finally: + self.widget_signal.enable_widget.emit(self.button_3) + + def redownload_files(self, files: list[FileObject]) -> tuple[bool, list[FileObject]]: + if not files: + return True, [] + + message = ( + "Some files failed to download.\n" + "Would you like to retry downloading them?\n\n" + + "\n".join(file.posix_path for file in files) + ) + ask_event = threading.Event() + context = { + "event": ask_event, + "title": "Retry downloading", + "message": message, + } + + self.widget_signal.askyesno.emit(context) + + if not ask_event.wait(60): + self.widget_signal.show_error_message.emit( + "Timed out while waiting for confirmation. Please try again." + ) + return False, files + + if not context.get("result", False): + return False, files + + # copy + pending = list(files) + + for retry_time in range(1, MAX_DOWNLOAD_ATTEMPTS + 1): + if not pending: + return True, [] + + self.download_signal.log.emit(f"Attempt {retry_time}/{len(pending)}") + + fails, successes = download_multiple_files(pending, progress_callback=self.download_signal.progress.emit) + + for file in successes: + self.download_signal.log.emit(f"Downloaded {file.posix_path}.") + + if fails: + pending = fails + else: + pending = [] + + if not pending: + return True, [] + + failed_list = "\n".join(file.posix_path for file in pending) + self.app.logger.warning( + "Unable to download these files:\n{}".format(failed_list) + ) + + return False, pending