From e42a9b0a50eb6079d6462f5012bfb9f4ea209f67 Mon Sep 17 00:00:00 2001 From: Wei Hong Date: Sun, 12 Jul 2026 00:06:11 +0800 Subject: [PATCH] Now the *launcher* can parse natives in version manifest. (Never let me did it again...it's a bullsh*t) --- app.py | 159 ++------- core_lib/common/common.py | 55 +++- core_lib/game/library.py | 203 ++++++++++++ core_lib/game/version_info.py | 124 ++++++- core_lib/qt/messagebox.py | 44 +++ dialog/test.py | 598 ++++++++++++++++++++++++++++++++++ main.py | 23 +- 7 files changed, 1065 insertions(+), 141 deletions(-) create mode 100644 core_lib/game/library.py create mode 100644 core_lib/qt/messagebox.py create mode 100644 dialog/test.py diff --git a/app.py b/app.py index ca8b3bb..471d386 100644 --- a/app.py +++ b/app.py @@ -1,155 +1,48 @@ import logging -import textwrap from pathlib import Path -from PySide6.QtCore import Qt -from PySide6.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QDialog, QPlainTextEdit, \ - QHBoxLayout +from PySide6.QtWidgets import QApplication, QMainWindow -from core_lib.common.common import download_file, FileObject, download_multiple_files from core_lib.qt.hack import move_window_to_center -from core_lib.game.version_info import get_version_manifest, get_latest_version, \ - fetch_specific_version_manifest, find_useful_part_in_specific_version_manifest, get_core_file_download_url_and_hash - +from dialog.test import TestDialog class Launcher(QApplication): def __init__(self): super().__init__() + # Window config self.setApplicationName("TestLauncher") self.main_window = QMainWindow() self.main_window.setGeometry(0, 0, 800, 600) - move_window_to_center(self.main_window) + # logger self.logger = logging.getLogger("Launcher.MainWindow") - self.test_dialog = self.TestDialog(self.main_window) + # dirs + self.program_dir = Path(__file__).parent + self.work_dir = Path.cwd() + + # test page + self.test_dialog = TestDialog(self, self.main_window) self.test_dialog.show() - class TestDialog(QDialog): - def __init__(self, parent): - super().__init__(parent) - self.parent = parent - self.resize(800, 600) - self.setWindowFlags(self.windowFlags() | Qt.WindowType.WindowMaximizeButtonHint) - - self.button = QPushButton("Fetch Version Data") - self.button.setObjectName("ActionButton") - self.button.clicked.connect(self.run_test) - - self.button_2 = QPushButton("Download client") - self.button_2.setObjectName("ActionButton") - self.button_2.clicked.connect(self.run_download_client) - - self.button_3 = QPushButton("Download libraries") - self.button_3.setObjectName("ActionButton") - self.button_3.clicked.connect(self.run_download_libraries) - - self.output = QPlainTextEdit() - self.output.setReadOnly(True) - - self.layout = QHBoxLayout() - self.layout.addWidget(self.output) - - self.right_layout = QVBoxLayout() - self.right_layout.setObjectName("RightLayout") - self.right_layout.setAlignment(Qt.AlignmentFlag.AlignRight) - self.right_layout.addWidget(self.button) - self.right_layout.addWidget(self.button_2) - self.right_layout.addWidget(self.button_3) - self.right_layout.addStretch() - self.right_layout.setContentsMargins(0, 10, 0, 0) - - self.layout.addLayout(self.right_layout) - - self.setLayout(self.layout) - - self.setStyleSheet( - textwrap.dedent(""" - QPushButton#ActionButton { - padding: 6px 12px; - } - """)) - - @property - def app(self): - return self.parent.parent - - def run_test(self): - self.output.clear() - data = get_version_manifest() - ver = get_latest_version(data) - 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) - - def run_download_client(self): - self.output.clear() - - data = get_version_manifest() - ver = get_latest_version(data) - ver_data = fetch_specific_version_manifest(data, ver) - url, sha1 = get_core_file_download_url_and_hash(ver_data) - - dest = Path(Path(__file__).parent, "temp", f"client-{ver}.jar") - - if url: - file = FileObject(dest, url=url) - download_file(file) - else: - self.output.setPlainText("Download URL not found.") - return - - self.output.setPlainText("Client downloaded." if dest.exists() else "Client not found.") - - def run_download_libraries(self): - self.output.clear() - - def log_file_downloaded(filepath): - self.output.appendPlainText(f"Downloaded {filepath}.") - - data = get_version_manifest() - ver = get_latest_version(data) - ver_data = fetch_specific_version_manifest(data, ver) - _, _, _, _, libraries, _ = find_useful_part_in_specific_version_manifest(ver_data) - - output_dir = Path(Path(__file__).parent, "tmp", f"libraries-{ver}") - - output_dir.mkdir(parents=True, exist_ok=True) - - items = [] - - for library in libraries: - 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") - full_path = output_dir / relative_path - - if not url: - continue - - file = FileObject(full_path, url=url, sha1=sha1) - items.append(file) - - download_multiple_files(items, download_callback=log_file_downloaded) - def exec(self): self.main_window.show() self.exec_() + + @property + def temp_dir(self): + return self.work_dir / "temp" + + @property + def config_dir(self): + return self.work_dir / "config" + + @property + def data_dir(self): + return self.work_dir / "data" + + @property + def log_dir(self): + return self.work_dir / "log" \ No newline at end of file diff --git a/core_lib/common/common.py b/core_lib/common/common.py index 2c43530..319a5f4 100644 --- a/core_lib/common/common.py +++ b/core_lib/common/common.py @@ -1,5 +1,8 @@ +from __future__ import annotations import hashlib import logging +import platform +import zipfile from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path @@ -7,7 +10,7 @@ from typing import List import requests -from core_lib.common.exception import HashMismatchException +from .exception import HashMismatchException DEFAULT_HASH_CHUNK_SIZE = 4096 DEFAULT_DOWNLOAD_CHUNK_SIZE = 4096 @@ -134,11 +137,12 @@ class FileObject: return self.__path.exists() -def download_file(file: FileObject, overwrite=True) -> FileObject | Exception: +def download_file(file: FileObject, overwrite=True, progress_callback: Callable[int]=None) -> FileObject | Exception: """ Download file from url :param file: File (Objected) :param overwrite: Overwrite existing file + :param progress_callback: Callback function called with progress information :return: """ @@ -153,6 +157,8 @@ def download_file(file: FileObject, overwrite=True) -> FileObject | Exception: 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)) @@ -162,6 +168,11 @@ def download_file(file: FileObject, overwrite=True) -> FileObject | Exception: break 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 file.expected_sha256 and not file.verify_sha256(): raise HashMismatchException("sha256", file.sha256, file.expected_sha256, file.posix_path) @@ -178,18 +189,20 @@ def download_file(file: FileObject, overwrite=True) -> FileObject | Exception: return file def download_multiple_files(files: List[FileObject], download_callback: Callable[str]=None, - max_workers=THREADED_DOWNLOAD_MAX_WORKERS) \ + max_workers=THREADED_DOWNLOAD_MAX_WORKERS, progress_callback: Callable[str]=None) \ -> List[FileObject]: """ Download multiple files :param download_callback: :param files: :param max_workers: + :param progress_callback: :return: List of file objects that download failed. """ fails = [] + files_count = 0 with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_file = {executor.submit(download_file, file): file for file in files} @@ -197,12 +210,48 @@ def download_multiple_files(files: List[FileObject], download_callback: Callable file = future_to_file[future] try: + files_count += 1 future.result() if callable(download_callback): download_callback(file.posix_path) + + if callable(progress_callback): + progress_callback(f"{len(files)} Files", files_count / len(files) * 100) except Exception as exc: logger.error("Unable to download file {}: {}".format(file.posix_path, exc)) fails.append(file) + return fails +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: + + 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) + +def get_platform_info(): + """ + Get information about the current platform. + :return: + platform_name: str, architecture: str, os_version: str + """ + platform_name = platform.system().lower() + arch = platform.uname().machine.lower() + os_version = platform.version() + + if platform_name == "darwin": + os_version = platform.mac_ver()[0] + elif platform_name == "linux": + os_version = platform.release() + + return platform_name, arch, os_version + diff --git a/core_lib/game/library.py b/core_lib/game/library.py new file mode 100644 index 0000000..cacc786 --- /dev/null +++ b/core_lib/game/library.py @@ -0,0 +1,203 @@ +import json +import logging +import re + +from ..common.common import get_platform_info +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, +) + +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): + """ + Checks if the given rule is allowed for the given platform to use it. + :param rules: Rules from library data + The following parameters can get from get_natives_platform_info + :param platform_rule_name: Platform rule name + :param platform_arch_type: Platform arch type + :param platform_version: Platform version + :return: + """ + allowed = False + + if not rules: + return True + + for rule in rules: + rule_action = rule.get("action", None) + os_section = rule.get("os", {}) + rule_os = os_section.get("name", None) + rule_arch = os_section.get("arch") # Not sure if is available in official source + rule_version = os_section.get("version", None) + + if rule_os is not None and platform_rule_name != rule_os: + continue + + if rule_arch is not None and platform_arch_type != rule_arch: + continue + + if rule_version is not None: + try: + if re.search(rule_version, platform_version) is None: + continue + except re.error as exc: + logger.warning( + f"Unsupported OS version regex {rule_version!r}: {exc}" + ) + continue + + if rule_action == "allow": + allowed = True + elif rule_action == "disallow": + allowed = False + else: + logger.warning(f"Unknown action {rule_action}\n") + + return allowed + + +def get_natives_platform_info(): + """ + Get platform information that converted to natives supported formula type + :return: + platform_key_name: str, platform_rule_name: str, platform_arch_type: str + """ + platform_name, arch, os_version = get_platform_info() + + platform_key_name = NATIVE_OS_KEY_MAP.get(platform_name, None) + platform_rule_name = NATIVE_OS_RULE_MAP.get(platform_name, None) + platform_arch_type = NATIVE_ARCH_KEY_MAP.get(arch, None) + + return platform_key_name, platform_rule_name, platform_arch_type + +def is_native_allowed(lib_name, platform_key_name, platform_arch_type, unspecified_arch_policy: str="x86_64") -> bool: + # Parse native key + match = re.search(r":(natives-[^:]+)$", lib_name) + + if match: + native_key = match.group(1) + logger.debug(f"Found native key {native_key} in {lib_name}") + + try: + parts = native_key.split("-") + + if len(parts) == 2: # Only os name exists + 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") + 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") + + 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") + else: + logger.warning("Unsupported native key '{}'".format(native_key)) + return False + + arch_type_converted = NATIVE_ARCH_KEY_MAP.get(arch_type, arch_type) + + if platform_key_name == native_os_name and arch_type_converted == platform_arch_type: + return True + + except ValueError: + logger.warning( + "Unsupported native key '{}'".format(native_key) + ) + + 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]: + classifier = natives.get(platform_rule_name) + + if classifier: + if "${arch}" in classifier: + legacy_arch = NATIVE_NEW_TO_OLD_ARCH.get(platform_arch_type) + + if legacy_arch is None: + logger.warning( + f"No legacy architecture mapping for {platform_arch_type}" + ) + classifier = None + else: + classifier = classifier.replace("${arch}", legacy_arch) + + if classifier: + native_artifact = classifiers.get(classifier) + + if native_artifact: + return True, native_artifact + + # classifiers-rule use NATIVE_OS_RULE_MAP + if classifiers: + logger.debug( + f"\nClassifiers: {json.dumps(classifiers, indent=4)}\n" + ) + + for native in classifiers.keys(): + data_raw = classifiers[native] + data = json.dumps(classifiers[native], indent=4) + logger.debug(f"Contains native: {data}\n") + + parts = native.split("-") + + universal = False + + if len(parts) == 2: + native_os_name = parts[1].lower() + 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") + 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") + 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") + else: + logger.warning("Unsupported legacy native key '{}'".format(native)) + continue + + arch_type_converted = NATIVE_OLD_TO_NEW_ARCH.get(str(arch_type), None) if not universal else None + if not universal and arch_type_converted != platform_arch_type: + logger.warning("Unsupported legacy native architecture '{}'".format(native)) + continue + + if native_os_name == platform_rule_name and (arch_type_converted == platform_arch_type or universal): + return True, data_raw + + return False, None \ No newline at end of file diff --git a/core_lib/game/version_info.py b/core_lib/game/version_info.py index 23bd7d6..ec1c34b 100644 --- a/core_lib/game/version_info.py +++ b/core_lib/game/version_info.py @@ -3,6 +3,56 @@ import logging import requests MOJANG_VERSION_MANIFEST_V2 = "https://piston-meta.mojang.com/mc/game/version_manifest_v2.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", +} + logger = logging.getLogger("Launcher.CoreLib") @@ -43,18 +93,60 @@ def get_latest_version(version_manifest: dict, is_snapshot=False) -> None | str: return latest_ver +def get_all_versions(version_manifest: dict, is_snapshot=False, + include_all_type=False, specific_type=None) -> list: + all_vers = version_manifest.get("versions", []) + + if len(all_vers) == 0: + logger.debug("No version section available in version manifest.") + return [] + + if not specific_type: + target_type = "release" if not is_snapshot else "snapshot" + else: + target_type = specific_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: + logger.debug("No version matches specified type: {}".format(target_type)) + + return filtered + +def get_available_version_types(version_manifest: dict) -> list: + all_vers = version_manifest.get("versions", []) + + if len(all_vers) == 0: + logger.debug("No version section available in version manifest.") + return [] + + 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, is_snapshot=False) -> str | None: + version_id: str) -> str | None: versions = version_manifest.get("versions", []) if len(versions) == 0: logger.debug("No version section available in version manifest.") return None - ver_type = "release" if not is_snapshot else "snapshot" - versions = [version for version in versions if version.get("type") == ver_type] - for version in versions: if version.get("id") == version_id: specific_ver_manifest_url = version.get("url", None) @@ -69,6 +161,30 @@ def get_specific_version_manifest_url(version_manifest: dict, return None +def get_specific_version_manifest_url_and_hash(version_manifest: dict, + version_id: str) -> tuple[str,str] | tuple[None, None]: + versions = version_manifest.get("versions", []) + + if len(versions) == 0: + logger.debug("No version section available in version manifest.") + return None, None + + for version in versions: + if version.get("id") == version_id: + specific_ver_manifest_url = version.get("url", None) + 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.") + + 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.") + + 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) diff --git a/core_lib/qt/messagebox.py b/core_lib/qt/messagebox.py new file mode 100644 index 0000000..5650816 --- /dev/null +++ b/core_lib/qt/messagebox.py @@ -0,0 +1,44 @@ +from PySide6.QtWidgets import QMessageBox, QMainWindow, QWidget + +def create_messagebox(parent: QWidget, title: str, message: str, + icon: QMessageBox.Icon = QMessageBox.Icon.Information, + button: QMessageBox.StandardButton = QMessageBox.StandardButton.Ok) -> QMessageBox: + msg = QMessageBox(parent) + msg.setWindowTitle(title) + msg.setText(message) + msg.setIcon(icon) + msg.setStandardButtons(button) + return msg + + +def error(parent: QWidget, title: str, message: str, + button: QMessageBox.StandardButton = QMessageBox.StandardButton.Ok) -> None: + create_messagebox( + parent, + title=title, + message=message, + icon=QMessageBox.Icon.Critical, + button=button + ).exec() + + +def info(parent: QWidget, title: str, message: str, + button: QMessageBox.StandardButton = QMessageBox.StandardButton.Ok) -> None: + create_messagebox( + parent, + title=title, + message=message, + icon=QMessageBox.Icon.Information, + button=button + ).exec() + + +def warning(parent: QWidget, title: str, message: str, + button: QMessageBox.StandardButton = QMessageBox.StandardButton.Ok) -> None: + create_messagebox( + parent, + title=title, + message=message, + icon=QMessageBox.Icon.Warning, + button=button + ).exec() diff --git a/dialog/test.py b/dialog/test.py new file mode 100644 index 0000000..a91a8e9 --- /dev/null +++ b/dialog/test.py @@ -0,0 +1,598 @@ +import re +import textwrap +import threading +from pathlib import Path + +from PySide6.QtCore import Qt, QObject, Signal +from PySide6.QtWidgets import QPushButton, QVBoxLayout, QDialog, QPlainTextEdit, \ + QHBoxLayout, QComboBox + +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.qt import messagebox + + +class TestDialog(QDialog): + def __init__(self, app, parent): + super().__init__(parent) + self.app = app + self.parent = parent + self.resize(800, 600) + self.setWindowFlags(self.windowFlags() | Qt.WindowType.WindowMaximizeButtonHint) + + self.button = QPushButton("Fetch Version Data") + 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) + + self.button_3 = QPushButton("Download libraries") + 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.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.ver_dropdown = QComboBox() + self.ver_dropdown.setObjectName("Dropdown") + + self.ver_type_dropdown = QComboBox() + self.ver_type_dropdown.setObjectName("Dropdown") + + self.ver_type_dropdown.currentTextChanged.connect( + self.load_all_version + ) + + self.output = QPlainTextEdit() + self.output.setReadOnly(True) + + self.layout = QHBoxLayout() + self.layout.addWidget(self.output) + + self.right_layout = QVBoxLayout() + 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.setContentsMargins(0, 10, 0, 0) + + self.layout.addLayout(self.right_layout) + + self.setLayout(self.layout) + + self.setStyleSheet( + textwrap.dedent(""" + QPushButton#ActionButton { + padding: 6px 12px; + } + """)) + + # Bind signal functions + self.download_signal = self.DownloadSignal(self) + self.widget_signal = self.WidgetSignal(self) + + self.download_signal.log.connect( + self.output.appendPlainText + ) + + self.download_signal.finished.connect( + lambda count: self.output.appendPlainText( + f"Download finished, failed: {count}" + ) + ) + + self.download_signal.progress.connect( + lambda name, progress: self.output.setPlainText( + f"Downloading {name}: {self.get_progress(progress)}" + ) + ) + + self.widget_signal.clean_ver_types_dropdown.connect( + self.ver_type_dropdown.clear + ) + + self.widget_signal.update_types_dropdown.connect( + lambda types: self.update_type_dropdown(types) + ) + + self.widget_signal.clean_vers_dropdown.connect( + self.ver_dropdown.clear + ) + + self.widget_signal.update_vers_dropdown.connect( + lambda vers: self.update_version_dropdown(vers) + ) + + self.widget_signal.show_error_message.connect( + lambda msg: messagebox.error( + self.parent, + "Error", + message=msg + ) + ) + + self.widget_signal.show_info_message.connect( + lambda msg: messagebox.info( + self.parent, + "Info", + message=msg + ) + ) + + self.widget_signal.enable_widget.connect( + lambda widget: widget.setEnabled(True) + ) + + self.widget_signal.disable_widget.connect( + lambda widget: widget.setDisabled(True) + ) + + self.load_all_types() + + class DownloadSignal(QObject): + log = Signal(str) + finished = Signal(int) + failed = Signal(str) + progress = Signal(str, int) + + class WidgetSignal(QObject): + clean_ver_types_dropdown = Signal() + clean_vers_dropdown = Signal() + update_vers_dropdown = Signal(list) + update_types_dropdown = Signal(list) + show_error_message = Signal(str) + show_info_message = Signal(str) + enable_widget = Signal(object) + disable_widget = Signal(object) + + @staticmethod + def get_progress(value): + width = 30 + filled = int(width * value / 100) + + bar = "█" * filled + "░" * (width - filled) + + return f"[{bar}] {value:.1f}%" + + def update_type_dropdown(self, types: list): + if "release" in types: + types.pop(types.index("release")) + types.insert(0, "release") + + self.ver_type_dropdown.addItems(types) + + if len(self.ver_type_dropdown.currentText()) == 0: + self.ver_type_dropdown.setCurrentIndex(0) + + def update_version_dropdown(self, versions: list): + ids = [] + + for version in versions: + ver_id = version.get("id", None) + if ver_id is not None: + ids.append(ver_id) + + self.ver_dropdown.addItems(ids) + + if len(self.ver_dropdown.currentText()) == 0: + self.ver_dropdown.setCurrentIndex(0) + + def run_test(self): + self.button.setDisabled(True) + + self.output.clear() + data = get_version_manifest() + ver = self.ver_dropdown.currentText() + + if len(ver) == 0 or ver is None: + messagebox.error( + self.parent, + "Error", + "Please select a version before running the test.", + ) + + 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() + + def run_download_client(self): + self.button_2.setDisabled(True) + + self.output.clear() + + thread = threading.Thread( + target=self._download_client, + daemon=True, + ) + 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() + + 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, + ) + thread.start() + + def load_all_version(self): + ver_type = self.ver_type_dropdown.currentText() + + thread = threading.Thread( + target=self._load_all_version, + daemon=True, + args=(ver_type,) + ) + thread.start() + self.download_signal.log.emit("Loading all versions...") + + def load_all_types(self): + thread = threading.Thread( + target=self._load_all_types, + daemon=True + ) + thread.start() + self.download_signal.log.emit("Loading all types...") + + # 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() + 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() + vers = get_all_versions( + data, + specific_type=ver_type + ) + self.widget_signal.update_vers_dropdown.emit(vers) + + def _download_specific_version_manifest(self, ver=None): + 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 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 + else: + self.download_signal.log.emit(f"Version {ver} manifest url: {ver_manifest_url} sha1: {sha1}") + + filepath = Path(self.app.work_dir, "versions", ver, f"{ver}.json") + + 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: + self.widget_signal.show_error_message.emit( + "Select a version before download the client." + ) + 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: + self.widget_signal.show_error_message.emit( + "Select a version before download libraries." + ) + 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) + + output_dir = Path(self.app.work_dir, "libraries") + + output_dir.mkdir(parents=True, exist_ok=True) + + 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", "") + 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", {}) + + # Ignore natives + if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or + any(native.startswith("natives-") for native in classifiers)): + continue + + if not url: + continue + + 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 + + file = FileObject(full_path, url=url, sha1=sha1) + + if sha1 is None: + self.app.logger.warning(f"File {full_path} has no SHA1 hash.") + + if file.exists and (sha1 is None or file.verify("sha1", sha1)): + log_file_downloaded(file.posix_path) + continue + + 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.widget_signal.show_error_message.emit( + "Unexpected exception occurred while finding natives: {}".format(exc) + ) + finally: + self.button_4.setDisabled(False) diff --git a/main.py b/main.py index 7e98e0f..8d77e66 100644 --- a/main.py +++ b/main.py @@ -1,9 +1,11 @@ +import os import sys import argparse import logging from app import Launcher import tkinter as tk +from tkinter import messagebox from core_lib.log.default import DEFAULT_FORMAT, DEFAULT_DATE_FORMAT, get_colorful_handler @@ -45,6 +47,8 @@ def select_mode() -> None | str: def tui_entry(): logger = logging.getLogger("TestLauncher.Tui") logger.info("Wow tui is unfinished") + return 0 + def main(): root_logger = logging.getLogger() @@ -54,9 +58,10 @@ def main(): parser = argparse.ArgumentParser() parser.add_argument("-m", "--mode", choices=['lightweight', 'full', 'choice'], default='choice', help="Select mode (lightweight for tui, full is gui," - "choice for a mode select window)", required=False) + "choice for a mode select window) (TUI is not implemented)", required=False) parser.add_argument("-d", "--debug", action="store_true", help="Enable debug mode", default=False) + parser.add_argument("-wd", "--work-dir", default=os.getcwd(), help="Work directory") args, unknown = parser.parse_known_args() for arg in unknown: @@ -74,6 +79,21 @@ def main(): debug = args.debug + # check workdir + if not os.path.exists(args.work_dir) or not os.path.isdir(args.work_dir): + if mode == "full": + messagebox.showerror("Error", f"Specified work directory \"{args.work_dir}\" does not exist or" + " is not a directory.") + else: + logger.error("Work directory '%s' does not exist. Exiting..." % args.work_dir) + + sys.exit(-1) + + if os.getcwd() != args.work_dir: + os.chdir(args.work_dir) + + logger.info("Current working directory is {}".format(args.work_dir)) + # Init lib level = logging.DEBUG if debug else logging.INFO @@ -94,5 +114,6 @@ def main(): logger.error("Unknown mode '%s'" % mode) sys.exit(1) + if __name__ == '__main__': main()