Add a test method to download client and libraries.

Add a simple File object to simplify certain file operations.

INFO: Libraries are unfiltered. Some of the library that for other platform will be downloaded too.

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_()

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
+

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)

core_lib/core_init.py

@@ -1,3 +0,0 @@
-import logging
-
-logger = logging.getLogger("Launcher.CoreLib")

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