diff --git a/app.py b/app.py index 5bd3196..ca8b3bb 100644 --- a/app.py +++ b/app.py @@ -1,12 +1,15 @@ import logging import textwrap +from pathlib import Path from PySide6.QtCore import Qt -from PySide6.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QDialog, QPlainTextEdit, \ +from PySide6.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QDialog, QPlainTextEdit, \ QHBoxLayout + +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, get_specific_version_manifest_url, \ - fetch_specific_version_manifest, find_useful_part_in_specific_version_manifest +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 class Launcher(QApplication): @@ -27,13 +30,21 @@ class Launcher(QApplication): def __init__(self, parent): super().__init__(parent) self.parent = parent - self.resize(300, 200) - self.setWindowFlags(self.windowFlags() | Qt.WindowMaximizeButtonHint) + 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) @@ -44,6 +55,8 @@ class Launcher(QApplication): 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) @@ -63,6 +76,7 @@ class Launcher(QApplication): 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) @@ -85,6 +99,57 @@ class Launcher(QApplication): 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_() diff --git a/core_lib/common/common.py b/core_lib/common/common.py new file mode 100644 index 0000000..2c43530 --- /dev/null +++ b/core_lib/common/common.py @@ -0,0 +1,208 @@ +import hashlib +import logging +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path +from typing import List + +import requests + +from core_lib.common.exception import HashMismatchException + +DEFAULT_HASH_CHUNK_SIZE = 4096 +DEFAULT_DOWNLOAD_CHUNK_SIZE = 4096 +THREADED_DOWNLOAD_MAX_WORKERS = 10 + +logger = logging.getLogger("Launcher.CoreLib") + +class FileObject: + def __init__(self, path: Path, url=None, sha1=None, sha256=None, md5=None): + self.__path = path + + self.expected_sha1 = sha1 + self.expected_sha256 = sha256 + self.expected_md5 = md5 + + self.__sha1 = None + self.__sha256 = None + self.__md5 = None + + self.url = url + + def __eq__(self, other): + """ + THIS OPERATION ONLY COMPARE AND FILE SIZE! Use + :param other: + :return: + """ + if self.path == other.path: + return True + + if self.size == other.size: + return True + + return False + + def is_same(self, file: FileObject, hash_name="sha256") -> bool: + if hash_name == "sha1": + return self.sha1 == file.sha1 + + if hash_name == "md5": + return self.md5 == file.md5 + + if hash_name != "sha256": + logger.warning("Unsupported hash_name {}. Ignored and use sha256 instead.".format(hash_name)) + + return self.sha256 == file.sha256 + + @property + def sha1(self): + if self.__sha1 is None: + self.__sha1 = self._calculate_hash("sha1") + + return self.__sha1 + + @property + def sha256(self): + if self.__sha256 is None: + self.__sha256 = self._calculate_hash("sha256") + + return self.__sha256 + + @property + def md5(self): + if self.__md5 is None: + self.__md5 = self._calculate_hash("md5") + + return self.__md5 + + def verify_sha1(self, expected_hash=None): + return self.sha1 == expected_hash if expected_hash else self.expected_sha1 == self.sha1 + + def verify_sha256(self, expected_hash=None): + return self.sha256 == expected_hash if expected_hash else self.expected_sha256 == self.sha256 + + def verify_md5(self, expected_hash=None): + return self.sha1 == expected_hash if expected_hash else self.expected_sha1 == self.sha1 + + def verify(self, algorithm, expected_hash=None): + if algorithm == "sha1": + return self.verify_sha1(expected_hash if expected_hash else self.expected_sha1) + + if algorithm == "sha256": + return self.verify_sha256(expected_hash if expected_hash else self.expected_sha256) + + if algorithm == "md5": + return self.verify_md5(expected_hash if expected_hash else self.expected_md5) + + raise Exception("Unsupported algorithm: {}".format(algorithm)) + + def _calculate_hash(self, algorithm): + h = hashlib.new(algorithm) + + with self.__path.open("rb") as f: + for chunk in iter(lambda: f.read(DEFAULT_HASH_CHUNK_SIZE), b""): + h.update(chunk) + + return h.hexdigest() + + @property + def path(self): + return self.__path + + @property + def posix_path(self): + return self.__path.absolute().as_posix() + + @property + def size(self): + return self.__path.stat().st_size + + @property + def name(self): + return self.__path.name + + @property + def extension(self): + return self.__path.suffix + + def __hash__(self): + return hash(self.__path) + + @property + def exists(self): + return self.__path.exists() + + +def download_file(file: FileObject, overwrite=True) -> FileObject | Exception: + """ + Download file from url + :param file: File (Objected) + :param overwrite: Overwrite existing file + :return: + """ + + 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)) + + if file.url is None: + raise Exception("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: + 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 + + f.write(chunk) + + if file.expected_sha256 and not file.verify_sha256(): + raise HashMismatchException("sha256", file.sha256, file.expected_sha256, file.posix_path) + elif file.expected_sha1 and not file.verify_sha1(): + raise HashMismatchException("sha1", file.sha1, file.expected_sha1, file.posix_path) + elif file.expected_md5 and not file.verify_md5(): + raise HashMismatchException("md5", file.md5, file.expected_md5, file.posix_path) + + if any([file.expected_sha1, file.expected_md5, file.expected_sha256]): + logger.debug("File hash is match as the expected hash.") + else: + logger.warning("File hash is not checked.") + + return file + +def download_multiple_files(files: List[FileObject], download_callback: Callable[str]=None, + max_workers=THREADED_DOWNLOAD_MAX_WORKERS) \ + -> List[FileObject]: + """ + Download multiple files + :param download_callback: + :param files: + :param max_workers: + :return: List of file objects that download failed. + """ + + fails = [] + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_file = {executor.submit(download_file, file): file for file in files} + + for future in as_completed(future_to_file): + file = future_to_file[future] + + try: + future.result() + if callable(download_callback): + download_callback(file.posix_path) + except Exception as exc: + logger.error("Unable to download file {}: {}".format(file.posix_path, exc)) + fails.append(file) + + return fails + diff --git a/core_lib/common/exception.py b/core_lib/common/exception.py new file mode 100644 index 0000000..2275696 --- /dev/null +++ b/core_lib/common/exception.py @@ -0,0 +1,15 @@ +from pathlib import Path + + +class LauncherException(Exception): + def __init__(self, message): + self.message = message + + def __str__(self): + return self.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) diff --git a/core_lib/core_init.py b/core_lib/core_init.py deleted file mode 100644 index cc28335..0000000 --- a/core_lib/core_init.py +++ /dev/null @@ -1,3 +0,0 @@ -import logging - -logger = logging.getLogger("Launcher.CoreLib") \ No newline at end of file diff --git a/core_lib/game/version_info.py b/core_lib/game/version_info.py index c5a99bd..23bd7d6 100644 --- a/core_lib/game/version_info.py +++ b/core_lib/game/version_info.py @@ -1,10 +1,10 @@ import json +import logging import requests -from .core_init import logger - MOJANG_VERSION_MANIFEST_V2 = "https://piston-meta.mojang.com/mc/game/version_manifest_v2.json" +logger = logging.getLogger("Launcher.CoreLib") def get_version_manifest(manifest_url=MOJANG_VERSION_MANIFEST_V2, timeout=10) -> dict | None: logger.debug("Getting version manifest from {}".format(manifest_url)) @@ -32,7 +32,6 @@ def get_latest_version(version_manifest: dict, is_snapshot=False) -> None | str: if len(latest) == 0: logger.debug("No latest version section available in version manifest.") - logger.debug("Data: {}".format(json.dumps(version_manifest, indent=4))) return None latest_ver = latest.get("release" if not is_snapshot else "snapshot", None) @@ -99,7 +98,7 @@ def find_useful_part_in_specific_version_manifest(spec_version_manifest: dict)\ 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("java_version", {}) + java_version = spec_version_manifest.get("javaVersion", {}) libraries = spec_version_manifest.get("libraries", []) main_class = spec_version_manifest.get("mainClass", None) @@ -122,3 +121,29 @@ def find_useful_part_in_specific_version_manifest(spec_version_manifest: dict)\ logger.debug("No main class section available in version manifest.") return arguments, asset_index, downloads, java_version, libraries, main_class + +def get_core_file_download_url_and_hash(version_manifest: dict, is_server=False) -> tuple[str, str] | tuple[None, None]: + core_downloads = version_manifest.get("downloads", {}) + + if len(core_downloads) == 0: + logger.debug("No downloads section available in version manifest.") + return None, None + + 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 + + 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") + + if sha1 is None: + logger.debug("Core file sha1 is None.\n") + + return download_url, sha1 \ No newline at end of file