From 976308ac854e4854baccb798bdee3ea8b23794a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20Th=E1=BA=BF=20H=C6=B0ng?= Date: Sat, 1 Feb 2025 23:51:53 +0700 Subject: [PATCH] feat: unify multiple games into one --- docs/game/launcher/game.md | 7 + vollerei/abc/launcher/game.py | 14 + vollerei/common/enums.py | 7 + vollerei/game/genshin/functions.py | 86 +++++ vollerei/game/hsr/constants.py | 15 + vollerei/game/hsr/functions.py | 98 ++++++ vollerei/game/launcher/__init__.py | 5 + vollerei/game/launcher/api.py | 32 ++ vollerei/game/launcher/interface.py | 0 vollerei/game/launcher/manager.py | 510 ++++++++++++++++++++++++++++ vollerei/game/patcher.py | 223 ++++++++++++ vollerei/game/zzz/functions.py | 66 ++++ vollerei/genshin/launcher/game.py | 443 +----------------------- vollerei/hsr/constants.py | 16 +- vollerei/hsr/launcher/game.py | 478 +------------------------- vollerei/zzz/launcher/game.py | 437 +----------------------- 16 files changed, 1077 insertions(+), 1360 deletions(-) create mode 100644 docs/game/launcher/game.md create mode 100644 vollerei/game/genshin/functions.py create mode 100644 vollerei/game/hsr/constants.py create mode 100644 vollerei/game/hsr/functions.py create mode 100644 vollerei/game/launcher/__init__.py create mode 100644 vollerei/game/launcher/api.py create mode 100644 vollerei/game/launcher/interface.py create mode 100644 vollerei/game/launcher/manager.py create mode 100644 vollerei/game/patcher.py create mode 100644 vollerei/game/zzz/functions.py diff --git a/docs/game/launcher/game.md b/docs/game/launcher/game.md new file mode 100644 index 0000000..906bd16 --- /dev/null +++ b/docs/game/launcher/game.md @@ -0,0 +1,7 @@ +# Game + +::: vollerei.game.launcher.Game + handler: python + options: + show_root_heading: true + show_source: true diff --git a/vollerei/abc/launcher/game.py b/vollerei/abc/launcher/game.py index 663866a..6ce3cd6 100644 --- a/vollerei/abc/launcher/game.py +++ b/vollerei/abc/launcher/game.py @@ -114,6 +114,20 @@ class GameABC(ABC): """ pass + def get_version_config(self) -> tuple[int, int, int]: + """ + Gets the current installed game version from config.ini. + + Using this is not recommended, as only official launcher creates + and uses this file, instead you should use `get_version()`. + + This returns (0, 0, 0) if the version could not be found. + + Returns: + tuple[int, int, int]: Game version. + """ + pass + def get_update(self) -> resource.Patch | None: """ Get the game update diff --git a/vollerei/common/enums.py b/vollerei/common/enums.py index c562179..0de6fc2 100644 --- a/vollerei/common/enums.py +++ b/vollerei/common/enums.py @@ -1,6 +1,13 @@ from enum import Enum +class GameType(Enum): + Genshin = 0 + HSR = 1 + ZZZ = 3 + HI3 = 4 + + class GameChannel(Enum): Overseas = 0 China = 1 diff --git a/vollerei/game/genshin/functions.py b/vollerei/game/genshin/functions.py new file mode 100644 index 0000000..321aca5 --- /dev/null +++ b/vollerei/game/genshin/functions.py @@ -0,0 +1,86 @@ +from vollerei.common.enums import GameChannel +from vollerei.abc.launcher.game import GameABC +from vollerei.exceptions.game import GameNotInstalledError + + +def get_channel(game: GameABC) -> GameChannel: + """ + Gets the current game channel. + + Returns: + GameChannel: The current game channel. + """ + if game.channel_override: + return game.channel_override + if not game.is_installed(): + raise GameNotInstalledError("Game path is not set.") + if game.path.joinpath("YuanShen.exe").is_file(): + return GameChannel.China + return GameChannel.Overseas + + +def get_version(game: GameABC) -> tuple[int, int, int]: + """ + Gets the current installed game version. + + Credits to An Anime Team for the code that does the magic: + https://github.com/an-anime-team/anime-game-core/blob/main/src/games/genshin/game.rs#L52 + + If the above method fails, it'll fallback to read the config.ini file + for the version, which is not recommended (as described in + `get_version_config()` docs) + + This returns (0, 0, 0) if the version could not be found + (usually indicates the game is not installed), and in fact `is_installed()` uses + this method to check if the game is installed too. + + Returns: + tuple[int, int, int]: The version as a tuple of integers. + """ + + data_file = game.data_folder().joinpath("globalgamemanagers") + if not data_file.exists(): + return game.get_version_config() + + def bytes_to_int(byte_array: list[bytes]) -> int: + bytes_as_int = int.from_bytes(byte_array, byteorder="big") + actual_int = bytes_as_int - 48 # 48 is the ASCII code for 0 + return actual_int + + version_bytes: list[list[bytes]] = [[], [], []] + version_ptr = 0 + correct = True + try: + with data_file.open("rb") as f: + f.seek(4000) + for byte in f.read(10000): + match byte: + case 0: + version_bytes = [[], [], []] + version_ptr = 0 + correct = True + case 46: + version_ptr += 1 + if version_ptr > 2: + correct = False + case 95: + if ( + correct + and len(version_bytes[0]) > 0 + and len(version_bytes[1]) > 0 + and len(version_bytes[2]) > 0 + ): + return ( + bytes_to_int(version_bytes[0]), + bytes_to_int(version_bytes[1]), + bytes_to_int(version_bytes[2]), + ) + case _: + if correct and byte in b"0123456789": + version_bytes[version_ptr].append(byte) + else: + correct = False + except Exception: + pass + # Fallback to config.ini + return game.get_version_config() diff --git a/vollerei/game/hsr/constants.py b/vollerei/game/hsr/constants.py new file mode 100644 index 0000000..964cef4 --- /dev/null +++ b/vollerei/game/hsr/constants.py @@ -0,0 +1,15 @@ +MD5SUMS = { + "1.0.5": { + "cn": { + "StarRailBase.dll": "66c42871ce82456967d004ccb2d7cf77", + "UnityPlayer.dll": "0c866c44bb3752031a8c12ffe935b26f", + }, + "os": { + "StarRailBase.dll": "8aa3790aafa3dd176678392f3f93f435", + "UnityPlayer.dll": "f17b9b7f9b8c9cbd211bdff7771a80c2", + }, + } +} +# Patches +ASTRA_REPO = "https://notabug.org/mkrsym1/astra" +JADEITE_REPO = "https://codeberg.org/mkrsym1/jadeite/" diff --git a/vollerei/game/hsr/functions.py b/vollerei/game/hsr/functions.py new file mode 100644 index 0000000..8bce76a --- /dev/null +++ b/vollerei/game/hsr/functions.py @@ -0,0 +1,98 @@ +from hashlib import md5 +from vollerei.common.enums import GameChannel +from vollerei.abc.launcher.game import GameABC +from vollerei.game.hsr.constants import MD5SUMS + + +def get_channel(game: GameABC) -> GameChannel | None: + """ + Gets the current game channel. + + Only works for Star Rail version 1.0.5, other versions will return the + overridden channel or GameChannel.Overseas if no channel is overridden. + + This is not needed for game patching, since the patcher will automatically + detect the channel. + + Returns: + GameChannel: The current game channel. + """ + version = game.version_override or game.get_version() + if version == (1, 0, 5): + for channel, v in MD5SUMS["1.0.5"].values(): + for file, md5sum in v.values(): + if md5(game.path.joinpath(file).read_bytes()).hexdigest() != md5sum: + continue + match channel: + case "cn": + return GameChannel.China + case "os": + return GameChannel.Overseas + return None + + +def get_version(game: GameABC) -> tuple[int, int, int]: + """ + Gets the current installed game version. + + Credits to An Anime Team for the code that does the magic: + https://github.com/an-anime-team/anime-game-core/blob/main/src/games/star_rail/game.rs#L49 + + If the above method fails, it'll fallback to read the config.ini file + for the version, which is not recommended (as described in + `get_version_config()` docs) + + This returns (0, 0, 0) if the version could not be found + (usually indicates the game is not installed), and in fact `is_installed()` uses + this method to check if the game is installed too. + + Returns: + tuple[int, int, int]: The version as a tuple of integers. + """ + + data_file = game.data_folder().joinpath("data.unity3d") + if not data_file.exists(): + return game.get_version_config() + + def bytes_to_int(byte_array: list[bytes]) -> int: + bytes_as_int = int.from_bytes(byte_array, byteorder="big") + actual_int = bytes_as_int - 48 # 48 is the ASCII code for 0 + return actual_int + + version_bytes: list[list[bytes]] = [[], [], []] + version_ptr = 0 + correct = True + try: + with data_file.open("rb") as f: + f.seek(0x7D0) # 2000 in decimal + for byte in f.read(10000): + match byte: + case 0: + version_bytes = [[], [], []] + version_ptr = 0 + correct = True + case 46: + version_ptr += 1 + if version_ptr > 2: + correct = False + case 38: + if ( + correct + and len(version_bytes[0]) > 0 + and len(version_bytes[1]) > 0 + and len(version_bytes[2]) > 0 + ): + return ( + bytes_to_int(version_bytes[0]), + bytes_to_int(version_bytes[1]), + bytes_to_int(version_bytes[2]), + ) + case _: + if correct and byte in b"0123456789": + version_bytes[version_ptr].append(byte) + else: + correct = False + except Exception: + pass + # Fallback to config.ini + return game.get_version_config() diff --git a/vollerei/game/launcher/__init__.py b/vollerei/game/launcher/__init__.py new file mode 100644 index 0000000..4f5a40a --- /dev/null +++ b/vollerei/game/launcher/__init__.py @@ -0,0 +1,5 @@ +# Re-exports +from vollerei.game.launcher.manager import Game + + +__all__ = ["Game"] diff --git a/vollerei/game/launcher/api.py b/vollerei/game/launcher/api.py new file mode 100644 index 0000000..0194f37 --- /dev/null +++ b/vollerei/game/launcher/api.py @@ -0,0 +1,32 @@ +from vollerei.common.api import get_game_packages, resource +from vollerei.common.enums import GameChannel, GameType + + +def get_game_package( + game_type: GameType, channel: GameChannel = GameChannel.Overseas +) -> resource.GameInfo: + """ + Get game package information from the launcher API. + + Doesn't work with HI3 but well, we haven't implemented anything for that game yet. + + Default channel is overseas. + + Args: + channel: Game channel to get the resource information from. + + Returns: + GameInfo: Game resource information. + """ + find_str: str + match game_type: + case GameType.HSR: + find_str = "hkrpg" + case GameType.Genshin: + find_str = "hk4e" + case GameType.ZZZ: + find_str = "nap" + game_packages = get_game_packages(channel=channel) + for package in game_packages: + if find_str in package.game.biz: + return package diff --git a/vollerei/game/launcher/interface.py b/vollerei/game/launcher/interface.py new file mode 100644 index 0000000..e69de29 diff --git a/vollerei/game/launcher/manager.py b/vollerei/game/launcher/manager.py new file mode 100644 index 0000000..70c69be --- /dev/null +++ b/vollerei/game/launcher/manager.py @@ -0,0 +1,510 @@ +from configparser import ConfigParser +from io import IOBase +from os import PathLike +from pathlib import Path, PurePath +from vollerei.abc.launcher.game import GameABC +from vollerei.common import ConfigFile, functions +from vollerei.common.api import resource +from vollerei.common.enums import GameType, VoicePackLanguage, GameChannel +from vollerei.exceptions.game import ( + GameAlreadyUpdatedError, + GameNotInstalledError, + PreDownloadNotAvailable, +) +from vollerei.game.launcher import api +from vollerei.game.hsr import functions as hsr_functions +from vollerei.game.genshin import functions as genshin_functions +from vollerei.game.zzz import functions as zzz_functions +from vollerei import paths +from vollerei.utils import download + + +class Game(GameABC): + """ + Manages the game installation + + For Star Rail and Zenless Zone Zero: + + Since channel detection isn't implemented yet, most functions assume you're + using the overseas version of the game. You can override channel by setting + the property `channel_override` to the channel you want to use. + """ + + def __init__( + self, game_type: GameType, path: PathLike = None, cache_path: PathLike = None + ): + self._path: Path | None = Path(path) if path else None + if not cache_path: + cache_path = paths.cache_path + cache_path = Path(cache_path) + self._game_type = game_type + self.cache: Path = cache_path.joinpath(f"game/{self._game_type.name.lower()}/") + self.cache.mkdir(parents=True, exist_ok=True) + self._version_override: tuple[int, int, int] | None = None + self._channel_override: GameChannel | None = None + + @property + def version_override(self) -> tuple[int, int, int] | None: + """ + Overrides the game version. + + This can be useful if you want to override the version of the game + and additionally working around bugs. + """ + return self._version_override + + @version_override.setter + def version_override(self, version: tuple[int, int, int] | str | None): + if isinstance(version, str): + version = tuple(int(i) for i in version.split(".")) + self._version_override = version + + @property + def channel_override(self) -> GameChannel | None: + """ + Overrides the game channel. + + Because game channel detection isn't implemented yet, you may need + to use this for some functions to work. + + This can be useful if you want to override the channel of the game + and additionally working around bugs. + """ + return self._channel_override + + @channel_override.setter + def channel_override(self, channel: GameChannel | str | None): + if isinstance(channel, str): + channel = GameChannel[channel] + self._channel_override = channel + + @property + def path(self) -> Path | None: + """ + Paths to the game folder. + """ + return self._path + + @path.setter + def path(self, path: PathLike): + self._path = Path(path) + + def data_folder(self) -> Path: + """ + Paths to the game data folder. + + Returns: + Path: The path to the game data folder. + """ + try: + match self._game_type: + case GameType.Genshin: + match self.get_channel(): + case GameChannel.China: + return self._path.joinpath("YuanShen_Data") + case GameChannel.Overseas: + return self._path.joinpath("GenshinImpact_Data") + case GameType.HSR: + return self._path.joinpath("StarRail_Data") + case GameType.ZZZ: + return self._path.joinpath("ZenlessZoneZero_Data") + except AttributeError: + raise GameNotInstalledError("Game path is not set.") + + def is_installed(self) -> bool: + """ + Checks if the game is installed. + + Returns: + bool: True if the game is installed, False otherwise. + """ + if self._path is None: + return False + match self._game_type: + case GameType.Genshin: + match self.get_channel(): + case GameChannel.China: + if not self._path.joinpath("YuanShen.exe").exists(): + return False + case GameChannel.Overseas: + if not self._path.joinpath("GenshinImpact.exe").exists(): + return False + case GameType.HSR: + if not self._path.joinpath("StarRail.exe").exists(): + return False + case GameType.ZZZ: + if not self._path.joinpath("ZenlessZoneZero.exe").exists(): + return False + if not self.data_folder().is_dir(): + return False + if self.get_version() == (0, 0, 0): + return False + return True + + def get_channel(self) -> GameChannel: + """ + Gets the current game channel. + + Only works for Genshin and Star Rail version 1.0.5, other versions will return + the overridden channel or GameChannel.Overseas if no channel is overridden. + + This is not needed for game patching, since the patcher will automatically + detect the channel. + + Returns: + GameChannel: The current game channel. + """ + match self._game_type: + case GameType.HSR: + return ( + hsr_functions.get_channel(self) + or self._channel_override + or GameChannel.Overseas + ) + case GameType.Genshin: + return ( + genshin_functions.get_channel(self) + or self._channel_override + or GameChannel.Overseas + ) + case _: + return self._channel_override or GameChannel.Overseas + + def get_version_config(self) -> tuple[int, int, int]: + """ + Gets the current installed game version from config.ini. + + Using this is not recommended, as only official launcher creates + and uses this file, instead you should use `get_version()`. + + This returns (0, 0, 0) if the version could not be found. + + Returns: + tuple[int, int, int]: Game version. + """ + cfg_file = self._path.joinpath("config.ini") + if not cfg_file.exists(): + return (0, 0, 0) + cfg = ConfigFile(cfg_file) + # Fk u miHoYo + if "general" in cfg.sections(): + version_str = cfg.get("general", "game_version", fallback="0.0.0") + elif "General" in cfg.sections(): + version_str = cfg.get("General", "game_version", fallback="0.0.0") + else: + return (0, 0, 0) + if version_str.count(".") != 2: + return (0, 0, 0) + try: + version = tuple(int(i) for i in version_str.split(".")) + except Exception: + return (0, 0, 0) + return version + + def set_version_config(self): + """ + Sets the current installed game version to config.ini. + + Only works for the global version of the game, not the Chinese one (since I don't have + them installed to test). + + This method is meant to keep compatibility with the official launcher only. + """ + cfg_file = self._path.joinpath("config.ini") + if cfg_file.exists(): + cfg = ConfigFile(cfg_file) + cfg.set("general", "game_version", self.get_version_str()) + cfg.save() + else: + cfg_dict = { + "general": { + "channel": 1, + "cps": "hyp_hoyoverse", + "game_version": self.get_version_str(), + "sub_channel": 0, + # This probably should be fetched from the server but well + "plugin_n06mjyc2r3_version": "1.1.0", + "uapc": None, # Honestly what's this? + } + } + + match self._game_type: + case GameType.Genshin: + cfg_dict["general"]["uapc"] = { + "hk4e_global": {"uapc": "f55586a8ce9f_"}, + "hyp": {"uapc": "f55586a8ce9f_"}, + } + case GameType.HSR: + cfg_dict["general"]["uapc"] = { + "hkrpg_global": {"uapc": "f5c7c6262812_"}, + "hyp": {"uapc": "f55586a8ce9f_"}, + } + case GameType.ZZZ: + cfg_dict["general"]["uapc"] = { + "nap_global": {"uapc": "f55586a8ce9f_"}, + "hyp": {"uapc": "f55586a8ce9f_"}, + } + cfg = ConfigParser() + cfg.read_dict(cfg_dict) + cfg.write(cfg_file.open("w")) + + def get_version(self) -> tuple[int, int, int]: + """ + Gets the current installed game version. + + Credits to An Anime Team for the code that does the magic, see the source + in `hsr/functions.py`, `genshin/functions.py` and `zzz/functions.py` for more info + + If the above method fails, it'll fallback to read the config.ini file + for the version, which is not recommended (as described in + `get_version_config()` docs) + + This returns (0, 0, 0) if the version could not be found + (usually indicates the game is not installed), and in fact `is_installed()` uses + this method to check if the game is installed too. + + Returns: + tuple[int, int, int]: The version as a tuple of integers. + """ + match self._game_type: + case GameType.HSR: + return hsr_functions.get_version(self) + case GameType.Genshin: + return genshin_functions.get_version(self) + case GameType.ZZZ: + return zzz_functions.get_version(self) + case _: + return self.get_version_config() + + def get_version_str(self) -> str: + """ + Gets the current installed game version as a string. + + Because this method uses `get_version()`, you should read the docs of + that method too. + + Returns: + str: The version as a string. + """ + return ".".join(str(i) for i in self.get_version()) + + def get_installed_voicepacks(self) -> list[VoicePackLanguage]: + """ + Gets the installed voicepacks. + + Returns: + list[VoicePackLanguage]: A list of installed voicepacks. + """ + if not self.is_installed(): + raise GameNotInstalledError("Game is not installed.") + voicepacks = [] + blacklisted_words = ["SFX"] + audio_package: Path + match self._game_type: + case GameType.Genshin: + audio_package = self.data_folder().joinpath( + "StreamingAssets/AudioAssets/" + ) + if audio_package.joinpath("AudioPackage").is_dir(): + audio_package = audio_package.joinpath("AudioPackage") + case GameType.HSR: + audio_package = self.data_folder().joinpath( + "Persistent/Audio/AudioPackage/Windows/" + ) + case GameType.ZZZ: + audio_package = self.data_folder().joinpath( + "StreamingAssets/Audio/Windows/Full/" + ) + for child in audio_package.iterdir(): + if child.resolve().is_dir() and child.name not in blacklisted_words: + name = child.name + if name.startswith("English"): + name = "English" + voicepack: VoicePackLanguage + try: + if self._game_type == GameType.ZZZ: + voicepack = VoicePackLanguage.from_zzz_name(child.name) + else: + voicepack = VoicePackLanguage[name] + voicepacks.append(voicepack) + except (ValueError, KeyError): + pass + return voicepacks + + def get_remote_game( + self, pre_download: bool = False + ) -> resource.Main | resource.PreDownload: + """ + Gets the current game information from remote. + + Args: + pre_download (bool): Whether to get the pre-download version. + Defaults to False. + + Returns: + A `Main` or `PreDownload` object that contains the game information. + """ + channel = self._channel_override or self.get_channel() + if pre_download: + game = api.get_game_package( + game_type=self._game_type, channel=channel + ).pre_download + if not game: + raise PreDownloadNotAvailable("Pre-download version is not available.") + return game + return api.get_game_package(game_type=self._game_type, channel=channel).main + + def get_update(self, pre_download: bool = False) -> resource.Patch | None: + """ + Gets the current game update. + + Args: + pre_download (bool): Whether to get the pre-download version. + Defaults to False. + + Returns: + A `Patch` object that contains the update information or + `None` if the game is not installed or already up-to-date. + """ + if not self.is_installed(): + return None + version = ( + ".".join(str(x) for x in self._version_override) + if self._version_override + else self.get_version_str() + ) + for patch in self.get_remote_game(pre_download=pre_download).patches: + if patch.version == version: + return patch + return None + + def repair_file( + self, + file: PathLike, + pre_download: bool = False, + game_info: resource.Game = None, + ) -> None: + """ + Repairs a game file. + + This will automatically handle backup and restore the file if the repair + fails. + + Args: + file (PathLike): The file to repair. + pre_download (bool): Whether to get the pre-download version. + Defaults to False. + """ + return self.repair_files([file], pre_download=pre_download, game_info=game_info) + + def repair_files( + self, + files: list[PathLike], + pre_download: bool = False, + game_info: resource.Game = None, + ) -> None: + """ + Repairs multiple game files. + + This will automatically handle backup and restore the file if the repair + fails. + + Args: + files (PathLike): The files to repair. + pre_download (bool): Whether to get the pre-download version. + Defaults to False. + game_info (resource.Game): The game information to use for repair. + """ + functions.repair_files( + self, files, pre_download=pre_download, game_info=game_info + ) + + def repair_game(self) -> None: + """ + Tries to repair the game by reading "pkg_version" file and downloading the + mismatched files from the server. + """ + functions.repair_game(self) + + def install_archive(self, archive_file: PathLike | IOBase) -> None: + """ + Applies an install archive to the game, it can be the game itself or a + voicepack one. + + `archive_file` can be a path to the archive file or a file-like object, + like if you have very high amount of RAM and want to download the archive + to memory instead of disk, this can be useful for you. + + Args: + archive_file (PathLike | IOBase): The archive file. + """ + if not isinstance(archive_file, IOBase): + archive_file = Path(archive_file) + functions.install_archive(self, archive_file) + + def apply_update_archive( + self, archive_file: PathLike | IOBase, auto_repair: bool = True + ) -> None: + """ + Applies an update archive to the game, it can be the game update or a + voicepack update. + + `archive_file` can be a path to the archive file or a file-like object, + like if you have very high amount of RAM and want to download the update + to memory instead of disk, this can be useful for you. + + `auto_repair` is used to determine whether to repair the file if it's + broken. If it's set to False, then it'll raise an exception if the file + is broken. + + Args: + archive_file (PathLike | IOBase): The archive file. + auto_repair (bool, optional): Whether to repair the file if it's broken. + Defaults to True. + """ + if not self.is_installed(): + raise GameNotInstalledError("Game is not installed.") + if not isinstance(archive_file, IOBase): + archive_file = Path(archive_file) + # Hello hell again, dealing with HDiffPatch and all the things again. + functions.apply_update_archive(self, archive_file, auto_repair=auto_repair) + + def install_update( + self, update_info: resource.Patch = None, auto_repair: bool = True + ): + """ + Installs an update from a `Patch` object. + + You may want to download the update manually and pass it to + `apply_update_archive()` instead for better control, and after that + execute `set_version_config()` to set the game version. + + Args: + update_info (Diff, optional): The update information. Defaults to None. + auto_repair (bool, optional): Whether to repair the file if it's broken. + Defaults to True. + """ + if not self.is_installed(): + raise GameNotInstalledError("Game is not installed.") + if not update_info: + update_info = self.get_update() + if not update_info or update_info.version == self.get_version_str(): + raise GameAlreadyUpdatedError("Game is already updated.") + update_url = update_info.game_pkgs[0].url + # Base game update + archive_file = self.cache.joinpath(PurePath(update_url).name) + download(update_url, archive_file) + self.apply_update_archive(archive_file=archive_file, auto_repair=auto_repair) + # Get installed voicepacks + installed_voicepacks = self.get_installed_voicepacks() + # Voicepack update + for remote_voicepack in update_info.audio_pkgs: + if remote_voicepack.language not in installed_voicepacks: + continue + # Voicepack is installed, update it + archive_file = self.cache.joinpath(PurePath(remote_voicepack.url).name) + download(remote_voicepack.url, archive_file) + self.apply_update_archive( + archive_file=archive_file, auto_repair=auto_repair + ) + self.set_version_config() diff --git a/vollerei/game/patcher.py b/vollerei/game/patcher.py new file mode 100644 index 0000000..af9053d --- /dev/null +++ b/vollerei/game/patcher.py @@ -0,0 +1,223 @@ +from enum import Enum +from shutil import copy2, rmtree +from packaging import version +from vollerei.abc.patcher import PatcherABC +from vollerei.common import telemetry +from vollerei.exceptions.game import GameNotInstalledError +from vollerei.exceptions.patcher import ( + VersionNotSupportedError, + PatcherError, + PatchUpdateError, +) +from vollerei.game.launcher.manager import Game, GameChannel +from vollerei.utils import download_and_extract, Git, Xdelta3 +from vollerei.paths import tools_data_path +from vollerei.hsr.constants import ASTRA_REPO, JADEITE_REPO + + +class PatchType(Enum): + """ + Patch type + + Astra: The old patch which patch the game directly (not recommended). + Jadeite: The new patch which patch the game in memory by DLL injection. + """ + + Astra = 0 + Jadeite = 1 + + +class Patcher(PatcherABC): + """ + Patch helper for HSR and HI3. + + By default this will use Jadeite as it is maintained and more stable. + """ + + def __init__(self, patch_type: PatchType = PatchType.Jadeite): + self._patch_type: PatchType = patch_type + self._path = tools_data_path.joinpath("patcher") + self._path.mkdir(parents=True, exist_ok=True) + self._jadeite = self._path.joinpath("jadeite") + self._astra = self._path.joinpath("astra") + self._git = Git() + self._xdelta3 = Xdelta3() + + @property + def patch_type(self) -> PatchType: + """ + Patch type, can be either Astra or Jadeite + """ + return self._patch_type + + @patch_type.setter + def patch_type(self, value: PatchType): + self._patch_type = value + + def _update_astra(self): + self._git.pull_or_clone(ASTRA_REPO, self._astra) + + def _update_jadeite(self): + release_info = self._git.get_latest_release(JADEITE_REPO) + file = self._git.get_latest_release_dl(release_info)[0] + file_version = release_info["tag_name"][1:] # Remove "v" prefix + current_version = None + if self._jadeite.joinpath("version").exists(): + with open(self._jadeite.joinpath("version"), "r") as f: + current_version = f.read() + if current_version: + if version.parse(file_version) <= version.parse(current_version): + return + download_and_extract(file, self._jadeite) + with open(self._jadeite.joinpath("version"), "w") as f: + f.write(file_version) + + def update_patch(self): + """ + Update the patch + """ + try: + match self._patch_type: + case PatchType.Astra: + self._update_astra() + case PatchType.Jadeite: + self._update_jadeite() + except Exception as e: + raise PatchUpdateError("Failed to update patch.") from e + + def _patch_astra(self, game: Game): + if game.get_version() != (1, 0, 5): + raise VersionNotSupportedError( + "Only version 1.0.5 is supported by Astra patch." + ) + self._update_astra() + file_type = None + match game.get_channel(): + case GameChannel.China: + file_type = "cn" + case GameChannel.Overseas: + file_type = "os" + # Backup and patch + for file in ["UnityPlayer.dll", "StarRailBase.dll"]: + game.path.joinpath(file).rename(game.path.joinpath(f"{file}.bak")) + self._xdelta3.patch_file( + self._astra.joinpath(f"{file_type}/diffs/{file}.vcdiff"), + game.path.joinpath(f"{file}.bak"), + game.path.joinpath(file), + ) + # Copy files + for file in self._astra.joinpath(f"{file_type}/files/").rglob("*"): + if file.suffix == ".bat": + continue + if file.is_dir(): + game.path.joinpath( + file.relative_to(self._astra.joinpath(f"{file_type}/files/")) + ).mkdir(parents=True, exist_ok=True) + copy2( + file, + game.path.joinpath( + file.relative_to(self._astra.joinpath(f"{file_type}/files/")) + ), + ) + + def _patch_jadeite(self): + """ + "Patch" the game with Jadeite patch. + + Unlike Astra patch, Jadeite patch does not modify the game files directly + but uses DLLs to patch the game in memory and it has an injector to do that + automatically. + """ + self._update_jadeite() + return self._jadeite + + def _unpatch_astra(self, game: Game): + if game.get_version() != (1, 0, 5): + raise VersionNotSupportedError( + "Only version 1.0.5 is supported by Astra patch." + ) + self._update_astra() + file_type = None + match game.get_channel(): + case GameChannel.China: + file_type = "cn" + case GameChannel.Overseas: + file_type = "os" + # Restore + for file in ["UnityPlayer.dll", "StarRailBase.dll"]: + if game.path.joinpath(f"{file}.bak").exists(): + game.path.joinpath(file).unlink() + game.path.joinpath(f"{file}.bak").rename(game.path.joinpath(file)) + # Remove files + for file in self._astra.joinpath(f"{file_type}/files/").rglob("*"): + if file.suffix == ".bat": + continue + file_rel = file.relative_to(self._astra.joinpath(f"{file_type}/files/")) + game_path = game.path.joinpath(file_rel) + if game_path.is_file(): + game_path.unlink() + elif game_path.is_dir(): + try: + game_path.rmdir() + except OSError: + pass + + def _unpatch_jadeite(self): + rmtree(self._jadeite, ignore_errors=True) + + def patch_game(self, game: Game): + """ + Patch the game + + If you use Jadeite (by default), this will just download Jadeite files + and won't actually patch the game because Jadeite will do that automatically. + + Args: + game (Game): The game to patch + """ + if not game.is_installed(): + raise PatcherError(GameNotInstalledError("Game is not installed")) + match self._patch_type: + case PatchType.Astra: + self._patch_astra(game) + case PatchType.Jadeite: + return self._patch_jadeite() + + def unpatch_game(self, game: Game): + """ + Unpatch the game + + If you use Jadeite (by default), this will just delete Jadeite files. + Note that Honkai Impact 3rd uses Jadeite too, so executing this will + delete the files needed by both games. + + Args: + game (Game): The game to unpatch + """ + if not game.is_installed(): + raise PatcherError(GameNotInstalledError("Game is not installed")) + match self._patch_type: + case PatchType.Astra: + self._unpatch_astra(game) + case PatchType.Jadeite: + self._unpatch_jadeite() + + def check_telemetry(self) -> list[str]: + """ + Check if telemetry servers are accessible by the user + + Returns: + list[str]: A list of telemetry servers that are accessible + """ + return telemetry.check_telemetry() + + def block_telemetry(self, telemetry_list: list[str] = None): + """ + Block the telemetry servers + + If telemetry_list is not provided, it will be checked automatically. + + Args: + telemetry_list (list[str], optional): A list of telemetry servers to block. + """ + telemetry.block_telemetry(telemetry_list) diff --git a/vollerei/game/zzz/functions.py b/vollerei/game/zzz/functions.py new file mode 100644 index 0000000..23fddf5 --- /dev/null +++ b/vollerei/game/zzz/functions.py @@ -0,0 +1,66 @@ +from vollerei.abc.launcher.game import GameABC + + +def get_version(game: GameABC) -> tuple[int, int, int]: + """ + Gets the current installed game version. + + Credits to An Anime Team for the code that does the magic: + https://github.com/an-anime-team/anime-game-core/blob/main/src/games/zzz/game.rs#L49 + + If the above method fails, it'll fallback to read the config.ini file + for the version, which is not recommended (as described in + `get_version_config()` docs) + + This returns (0, 0, 0) if the version could not be found + (usually indicates the game is not installed), and in fact `is_installed()` uses + this method to check if the game is installed too. + + Returns: + tuple[int, int, int]: The version as a tuple of integers. + """ + + data_file = game.data_folder().joinpath("globalgamemanagers") + if not data_file.exists(): + return game.get_version_config() + + def bytes_to_int(byte_array: list[bytes]) -> int: + bytes_as_int = int.from_bytes(byte_array, byteorder="big") + actual_int = bytes_as_int - 48 # 48 is the ASCII code for 0 + return actual_int + + version_bytes: list[list[bytes]] = [[], [], []] + version_ptr = 0 + correct = True + try: + with data_file.open("rb") as f: + f.seek(4000) + for byte in f.read(10000): + match byte: + case 0: + if ( + correct + and len(version_bytes[0]) > 0 + and len(version_bytes[1]) > 0 + and len(version_bytes[2]) > 0 + ): + found_version = tuple( + bytes_to_int(i) for i in version_bytes + ) + return found_version + version_bytes = [[], [], []] + version_ptr = 0 + correct = True + case b".": + version_ptr += 1 + if version_ptr > 2: + correct = False + case _: + if correct and byte in b"0123456789": + version_bytes[version_ptr].append(byte) + else: + correct = False + except Exception: + pass + # Fallback to config.ini + return game.get_version_config() diff --git a/vollerei/genshin/launcher/game.py b/vollerei/genshin/launcher/game.py index d7d439a..d0826ee 100644 --- a/vollerei/genshin/launcher/game.py +++ b/vollerei/genshin/launcher/game.py @@ -1,447 +1,12 @@ -from configparser import ConfigParser -from io import IOBase from os import PathLike -from pathlib import Path, PurePath -from vollerei.abc.launcher.game import GameABC -from vollerei.common import ConfigFile, functions -from vollerei.common.api import resource -from vollerei.common.enums import VoicePackLanguage, GameChannel -from vollerei.exceptions.game import ( - GameAlreadyUpdatedError, - GameNotInstalledError, - PreDownloadNotAvailable, -) -from vollerei.genshin.launcher import api -from vollerei import paths -from vollerei.utils import download +from vollerei.game.launcher.manager import Game as CommonGame +from vollerei.common.enums import GameType -class Game(GameABC): +class Game(CommonGame): """ Manages the game installation """ def __init__(self, path: PathLike = None, cache_path: PathLike = None): - self._path: Path | None = Path(path) if path else None - if not cache_path: - cache_path = paths.cache_path - cache_path = Path(cache_path) - self.cache: Path = cache_path.joinpath("game/genshin/") - self.cache.mkdir(parents=True, exist_ok=True) - self._version_override: tuple[int, int, int] | None = None - self._channel_override: GameChannel | None = None - - @property - def version_override(self) -> tuple[int, int, int] | None: - """ - Overrides the game version. - - This can be useful if you want to override the version of the game - and additionally working around bugs. - """ - return self._version_override - - @version_override.setter - def version_override(self, version: tuple[int, int, int] | str | None): - if isinstance(version, str): - version = tuple(int(i) for i in version.split(".")) - self._version_override = version - - @property - def channel_override(self) -> GameChannel | None: - """ - Overrides the game channel. - - This can be useful if you want to override the channel of the game - and additionally working around bugs. - """ - return self._channel_override - - @channel_override.setter - def channel_override(self, channel: GameChannel | str | None): - if isinstance(channel, str): - channel = GameChannel[channel] - self._channel_override = channel - - @property - def path(self) -> Path | None: - """ - Paths to the game folder. - """ - return self._path - - @path.setter - def path(self, path: PathLike): - self._path = Path(path) - - def data_folder(self) -> Path: - """ - Paths to the game data folder. - """ - try: - match self.get_channel(): - case GameChannel.China: - return self._path.joinpath("YuanShen_Data") - case GameChannel.Overseas: - return self._path.joinpath("GenshinImpact_Data") - except AttributeError: - raise GameNotInstalledError("Game path is not set.") - - def is_installed(self) -> bool: - """ - Checks if the game is installed. - - Returns: - bool: True if the game is installed, False otherwise. - """ - if self._path is None: - return False - try: - self.get_channel() - if self.data_folder().is_dir(): - return True - except GameNotInstalledError: - return False - if self.get_version() == (0, 0, 0): - return False - return True - - def get_channel(self) -> GameChannel: - """ - Gets the current game channel. - - Returns: - GameChannel: The current game channel. - """ - if self._channel_override: - return self._channel_override - if not self.is_installed(): - raise GameNotInstalledError("Game path is not set.") - if self._path.joinpath("YuanShen.exe").is_file(): - return GameChannel.China - return GameChannel.Overseas - - def get_version_config(self) -> tuple[int, int, int]: - """ - Gets the current installed game version from config.ini. - - Using this is not recommended, as only official launcher creates - and uses this file, instead you should use `get_version()`. - - This returns (0, 0, 0) if the version could not be found. - - Returns: - tuple[int, int, int]: Game version. - """ - cfg_file = self._path.joinpath("config.ini") - if not cfg_file.exists(): - return (0, 0, 0) - cfg = ConfigFile(cfg_file) - # Fk u miHoYo - if "general" in cfg.sections(): - version_str = cfg.get("general", "game_version", fallback="0.0.0") - elif "General" in cfg.sections(): - version_str = cfg.get("General", "game_version", fallback="0.0.0") - else: - return (0, 0, 0) - if version_str.count(".") != 2: - return (0, 0, 0) - try: - version = tuple(int(i) for i in version_str.split(".")) - except Exception: - return (0, 0, 0) - return version - - def set_version_config(self): - """ - Sets the current installed game version to config.ini. - - This method is meant to keep compatibility with the official launcher only. - """ - cfg_file = self._path.joinpath("config.ini") - if cfg_file.exists(): - cfg = ConfigFile(cfg_file) - cfg.set("general", "game_version", self.get_version_str()) - cfg.save() - else: - cfg = ConfigParser() - cfg.read_dict( - { - "general": { - "downloading_mode": None, - "channel": 1, - "cps": "hyp_hoyoverse", - "game_version": self.get_version_str(), - "sub_channel": 0, - # This is probably should be fetched from the server but well - "plugin_vt8u0pl2cc_version": "1.1.0", - # What's this in the Chinese version? - "uapc": { - "hk4e_global": {"uapc": "f55586a8ce9f_"}, - "hyp": {"uapc": "f55586a8ce9f_"}, - }, # Honestly what's this? - } - } - ) - cfg.write(cfg_file.open("w")) - - def get_version(self) -> tuple[int, int, int]: - """ - Gets the current installed game version. - - Credits to An Anime Team for the code that does the magic: - https://github.com/an-anime-team/anime-game-core/blob/main/src/games/genshin/game.rs#L52 - - If the above method fails, it'll fallback to read the config.ini file - for the version, which is not recommended (as described in - `get_version_config()` docs) - - This returns (0, 0, 0) if the version could not be found - (usually indicates the game is not installed), and in fact `is_installed()` uses - this method to check if the game is installed too. - - Returns: - tuple[int, int, int]: The version as a tuple of integers. - """ - - data_file = self.data_folder().joinpath("globalgamemanagers") - if not data_file.exists(): - return self.get_version_config() - - def bytes_to_int(byte_array: list[bytes]) -> int: - bytes_as_int = int.from_bytes(byte_array, byteorder="big") - actual_int = bytes_as_int - 48 # 48 is the ASCII code for 0 - return actual_int - - version_bytes: list[list[bytes]] = [[], [], []] - version_ptr = 0 - correct = True - try: - with data_file.open("rb") as f: - f.seek(4000) - for byte in f.read(10000): - match byte: - case 0: - version_bytes = [[], [], []] - version_ptr = 0 - correct = True - case 46: - version_ptr += 1 - if version_ptr > 2: - correct = False - case 95: - if ( - correct - and len(version_bytes[0]) > 0 - and len(version_bytes[1]) > 0 - and len(version_bytes[2]) > 0 - ): - return ( - bytes_to_int(version_bytes[0]), - bytes_to_int(version_bytes[1]), - bytes_to_int(version_bytes[2]), - ) - case _: - if correct and byte in b"0123456789": - version_bytes[version_ptr].append(byte) - else: - correct = False - except Exception: - pass - # Fallback to config.ini - return self.get_version_config() - - def get_version_str(self) -> str: - """ - Gets the current installed game version as a string. - - Because this method uses `get_version()`, you should read the docs of - that method too. - - Returns: - str: The version as a string. - """ - return ".".join(str(i) for i in self.get_version()) - - def get_installed_voicepacks(self) -> list[VoicePackLanguage]: - """ - Gets the installed voicepacks. - - Returns: - list[VoicePackLanguage]: A list of installed voicepacks. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - audio_assets = self.data_folder().joinpath("StreamingAssets/AudioAssets/") - if audio_assets.joinpath("AudioPackage").is_dir(): - audio_assets = audio_assets.joinpath("AudioPackage") - voicepacks = [] - for child in audio_assets.iterdir(): - if child.resolve().is_dir(): - name = child.name - if name.startswith("English"): - name = "English" - try: - voicepacks.append(VoicePackLanguage[name]) - except (ValueError, KeyError): - pass - return voicepacks - - def get_remote_game( - self, pre_download: bool = False - ) -> resource.Main | resource.PreDownload: - """ - Gets the current game information from remote. - - Args: - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - - Returns: - A `Main` or `PreDownload` object that contains the game information. - """ - channel = self._channel_override or self.get_channel() - if pre_download: - game = api.get_game_package(channel=channel).pre_download - if not game: - raise PreDownloadNotAvailable("Pre-download version is not available.") - return game - return api.get_game_package(channel=channel).main - - def get_update(self, pre_download: bool = False) -> resource.Patch | None: - """ - Gets the current game update. - - Args: - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - - Returns: - A `Patch` object that contains the update information or - `None` if the game is not installed or already up-to-date. - """ - if not self.is_installed(): - return None - version = ( - ".".join(str(x) for x in self._version_override) - if self._version_override - else self.get_version_str() - ) - for patch in self.get_remote_game(pre_download=pre_download).patches: - if patch.version == version: - return patch - return None - - def repair_file( - self, - file: PathLike, - pre_download: bool = False, - game_info: resource.Game = None, - ) -> None: - """ - Repairs a game file. - - This will automatically handle backup and restore the file if the repair - fails. - - Args: - file (PathLike): The file to repair. - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - """ - return self.repair_files([file], pre_download=pre_download, game_info=game_info) - - def repair_files( - self, - files: list[PathLike], - pre_download: bool = False, - game_info: resource.Game = None, - ) -> None: - """ - Repairs multiple game files. - - This will automatically handle backup and restore the file if the repair - fails. - - Args: - files (PathLike): The files to repair. - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - """ - functions.repair_files( - self, files, pre_download=pre_download, game_info=game_info - ) - - def repair_game(self) -> None: - """ - Tries to repair the game by reading "pkg_version" file and downloading the - mismatched files from the server. - """ - functions.repair_game(self) - - def apply_update_archive( - self, archive_file: PathLike | IOBase, auto_repair: bool = True - ) -> None: - """ - Applies an update archive to the game, it can be the game update or a - voicepack update. - - `archive_file` can be a path to the archive file or a file-like object, - like if you have very high amount of RAM and want to download the update - to memory instead of disk, this can be useful for you. - - `auto_repair` is used to determine whether to repair the file if it's - broken. If it's set to False, then it'll raise an exception if the file - is broken. - - Args: - archive_file (PathLike | IOBase): The archive file. - auto_repair (bool, optional): Whether to repair the file if it's broken. - Defaults to True. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - if not isinstance(archive_file, IOBase): - archive_file = Path(archive_file) - # Hello hell again, dealing with HDiffPatch and all the things again. - functions.apply_update_archive(self, archive_file, auto_repair=auto_repair) - - def install_update( - self, update_info: resource.Patch = None, auto_repair: bool = True - ): - """ - Installs an update from a `Patch` object. - - You may want to download the update manually and pass it to - `apply_update_archive()` instead for better control, and after that - execute `set_version_config()` to set the game version. - - Args: - update_info (Diff, optional): The update information. Defaults to None. - auto_repair (bool, optional): Whether to repair the file if it's broken. - Defaults to True. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - if not update_info: - update_info = self.get_update() - if not update_info or update_info.version == self.get_version_str(): - raise GameAlreadyUpdatedError("Game is already updated.") - update_url = update_info.game_pkgs[0].url - # Base game update - archive_file = self.cache.joinpath(PurePath(update_url).name) - download(update_url, archive_file) - self.apply_update_archive(archive_file=archive_file, auto_repair=auto_repair) - # Get installed voicepacks - installed_voicepacks = self.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_info.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = self.cache.joinpath(PurePath(remote_voicepack.url).name) - download(remote_voicepack.url, archive_file) - self.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - self.set_version_config() + super().__init__(GameType.Genshin, path, cache_path) diff --git a/vollerei/hsr/constants.py b/vollerei/hsr/constants.py index 287f3fb..f77f4cc 100644 --- a/vollerei/hsr/constants.py +++ b/vollerei/hsr/constants.py @@ -1,15 +1 @@ -MD5SUMS = { - "1.0.5": { - "cn": { - "StarRailBase.dll": "66c42871ce82456967d004ccb2d7cf77", - "UnityPlayer.dll": "0c866c44bb3752031a8c12ffe935b26f", - }, - "os": { - "StarRailBase.dll": "8aa3790aafa3dd176678392f3f93f435", - "UnityPlayer.dll": "f17b9b7f9b8c9cbd211bdff7771a80c2", - }, - } -} -# Patches -ASTRA_REPO = "https://notabug.org/mkrsym1/astra" -JADEITE_REPO = "https://codeberg.org/mkrsym1/jadeite/" +from vollerei.game.hsr.constants import * # noqa: F403 because we just want to re-export diff --git a/vollerei/hsr/launcher/game.py b/vollerei/hsr/launcher/game.py index 91ebb8c..f98c769 100644 --- a/vollerei/hsr/launcher/game.py +++ b/vollerei/hsr/launcher/game.py @@ -1,24 +1,9 @@ -from configparser import ConfigParser -from hashlib import md5 -from io import IOBase from os import PathLike -from pathlib import Path, PurePath -from vollerei.abc.launcher.game import GameABC -from vollerei.common import ConfigFile, functions -from vollerei.common.api import resource -from vollerei.common.enums import VoicePackLanguage, GameChannel -from vollerei.exceptions.game import ( - GameAlreadyUpdatedError, - GameNotInstalledError, - PreDownloadNotAvailable, -) -from vollerei.hsr.constants import MD5SUMS -from vollerei.hsr.launcher import api -from vollerei import paths -from vollerei.utils import download +from vollerei.game.launcher.manager import Game as CommonGame +from vollerei.common.enums import GameType -class Game(GameABC): +class Game(CommonGame): """ Manages the game installation @@ -28,459 +13,4 @@ class Game(GameABC): """ def __init__(self, path: PathLike = None, cache_path: PathLike = None): - self._path: Path | None = Path(path) if path else None - if not cache_path: - cache_path = paths.cache_path - cache_path = Path(cache_path) - self.cache: Path = cache_path.joinpath("game/hsr/") - self.cache.mkdir(parents=True, exist_ok=True) - self._version_override: tuple[int, int, int] | None = None - self._channel_override: GameChannel | None = None - - @property - def version_override(self) -> tuple[int, int, int] | None: - """ - Overrides the game version. - - This can be useful if you want to override the version of the game - and additionally working around bugs. - """ - return self._version_override - - @version_override.setter - def version_override(self, version: tuple[int, int, int] | str | None): - if isinstance(version, str): - version = tuple(int(i) for i in version.split(".")) - self._version_override = version - - @property - def channel_override(self) -> GameChannel | None: - """ - Overrides the game channel. - - Because game channel detection isn't implemented yet, you may need - to use this for some functions to work. - - This can be useful if you want to override the channel of the game - and additionally working around bugs. - """ - return self._channel_override - - @channel_override.setter - def channel_override(self, channel: GameChannel | str | None): - if isinstance(channel, str): - channel = GameChannel[channel] - self._channel_override = channel - - @property - def path(self) -> Path | None: - """ - Paths to the game folder. - """ - return self._path - - @path.setter - def path(self, path: PathLike): - self._path = Path(path) - - def data_folder(self) -> Path: - """ - Paths to the game data folder. - """ - try: - return self._path.joinpath("StarRail_Data") - except AttributeError: - raise GameNotInstalledError("Game path is not set.") - - def is_installed(self) -> bool: - """ - Checks if the game is installed. - - Returns: - bool: True if the game is installed, False otherwise. - """ - if self._path is None: - return False - if ( - not self._path.joinpath("StarRail.exe").exists() - or not self._path.joinpath("StarRail_Data").exists() - ): - return False - if self.get_version() == (0, 0, 0): - return False - return True - - def get_channel(self) -> GameChannel: - """ - Gets the current game channel. - - Only works for Star Rail version 1.0.5, other versions will return the - overridden channel or GameChannel.Overseas if no channel is overridden. - - This is not needed for game patching, since the patcher will automatically - detect the channel. - - Returns: - GameChannel: The current game channel. - """ - version = self._version_override or self.get_version() - if version == (1, 0, 5): - for channel, v in MD5SUMS["1.0.5"].values(): - for file, md5sum in v.values(): - if ( - md5(self._path.joinpath(file).read_bytes()).hexdigest() - != md5sum - ): - continue - match channel: - case "cn": - return GameChannel.China - case "os": - return GameChannel.Overseas - else: - # if self._path.joinpath("StarRail_Data").is_dir(): - # return GameChannel.Overseas - # elif self._path.joinpath("StarRail_Data").exists(): - # return GameChannel.China - # No reliable method there, so we'll just return the overridden channel or - # fallback to overseas. - return self._channel_override or GameChannel.Overseas - - def get_version_config(self) -> tuple[int, int, int]: - """ - Gets the current installed game version from config.ini. - - Using this is not recommended, as only official launcher creates - and uses this file, instead you should use `get_version()`. - - This returns (0, 0, 0) if the version could not be found. - - Returns: - tuple[int, int, int]: Game version. - """ - cfg_file = self._path.joinpath("config.ini") - if not cfg_file.exists(): - return (0, 0, 0) - cfg = ConfigFile(cfg_file) - # Fk u miHoYo - if "general" in cfg.sections(): - version_str = cfg.get("general", "game_version", fallback="0.0.0") - elif "General" in cfg.sections(): - version_str = cfg.get("General", "game_version", fallback="0.0.0") - else: - return (0, 0, 0) - if version_str.count(".") != 2: - return (0, 0, 0) - try: - version = tuple(int(i) for i in version_str.split(".")) - except Exception: - return (0, 0, 0) - return version - - def set_version_config(self): - """ - Sets the current installed game version to config.ini. - - This method is meant to keep compatibility with the official launcher only. - """ - cfg_file = self._path.joinpath("config.ini") - if cfg_file.exists(): - cfg = ConfigFile(cfg_file) - cfg.set("general", "game_version", self.get_version_str()) - cfg.save() - else: - cfg = ConfigParser() - cfg.read_dict( - { - "general": { - "channel": 1, - "cps": "hyp_hoyoverse", - "game_version": self.get_version_str(), - "sub_channel": 0, - # This is probably should be fetched from the server but well - "plugin_n06mjyc2r3_version": "1.1.0", - "uapc": { - "hkrpg_global": {"uapc": "f5c7c6262812_"}, - "hyp": {"uapc": "f55586a8ce9f_"}, - }, # Honestly what's this? - } - } - ) - cfg.write(cfg_file.open("w")) - - def get_version(self) -> tuple[int, int, int]: - """ - Gets the current installed game version. - - Credits to An Anime Team for the code that does the magic: - https://github.com/an-anime-team/anime-game-core/blob/main/src/games/star_rail/game.rs#L49 - - If the above method fails, it'll fallback to read the config.ini file - for the version, which is not recommended (as described in - `get_version_config()` docs) - - This returns (0, 0, 0) if the version could not be found - (usually indicates the game is not installed), and in fact `is_installed()` uses - this method to check if the game is installed too. - - Returns: - tuple[int, int, int]: The version as a tuple of integers. - """ - - data_file = self.data_folder().joinpath("data.unity3d") - if not data_file.exists(): - return self.get_version_config() - - def bytes_to_int(byte_array: list[bytes]) -> int: - bytes_as_int = int.from_bytes(byte_array, byteorder="big") - actual_int = bytes_as_int - 48 # 48 is the ASCII code for 0 - return actual_int - - version_bytes: list[list[bytes]] = [[], [], []] - version_ptr = 0 - correct = True - try: - with data_file.open("rb") as f: - f.seek(0x7D0) # 2000 in decimal - for byte in f.read(10000): - match byte: - case 0: - version_bytes = [[], [], []] - version_ptr = 0 - correct = True - case 46: - version_ptr += 1 - if version_ptr > 2: - correct = False - case 38: - if ( - correct - and len(version_bytes[0]) > 0 - and len(version_bytes[1]) > 0 - and len(version_bytes[2]) > 0 - ): - return ( - bytes_to_int(version_bytes[0]), - bytes_to_int(version_bytes[1]), - bytes_to_int(version_bytes[2]), - ) - case _: - if correct and byte in b"0123456789": - version_bytes[version_ptr].append(byte) - else: - correct = False - except Exception: - pass - # Fallback to config.ini - return self.get_version_config() - - def get_version_str(self) -> str: - """ - Gets the current installed game version as a string. - - Because this method uses `get_version()`, you should read the docs of - that method too. - - Returns: - str: The version as a string. - """ - return ".".join(str(i) for i in self.get_version()) - - def get_installed_voicepacks(self) -> list[VoicePackLanguage]: - """ - Gets the installed voicepacks. - - Returns: - list[VoicePackLanguage]: A list of installed voicepacks. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - voicepacks = [] - blacklisted_words = ["SFX"] - for child in ( - self.data_folder() - .joinpath("Persistent/Audio/AudioPackage/Windows/") - .iterdir() - ): - if child.resolve().is_dir() and child.name not in blacklisted_words: - try: - voicepacks.append(VoicePackLanguage[child.name]) - except ValueError: - pass - return voicepacks - - def get_remote_game( - self, pre_download: bool = False - ) -> resource.Main | resource.PreDownload: - """ - Gets the current game information from remote. - - Args: - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - - Returns: - A `Main` or `PreDownload` object that contains the game information. - """ - channel = self._channel_override or self.get_channel() - if pre_download: - game = api.get_game_package(channel=channel).pre_download - if not game: - raise PreDownloadNotAvailable("Pre-download version is not available.") - return game - return api.get_game_package(channel=channel).main - - def get_update(self, pre_download: bool = False) -> resource.Patch | None: - """ - Gets the current game update. - - Args: - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - - Returns: - A `Patch` object that contains the update information or - `None` if the game is not installed or already up-to-date. - """ - if not self.is_installed(): - return None - version = ( - ".".join(str(x) for x in self._version_override) - if self._version_override - else self.get_version_str() - ) - for patch in self.get_remote_game(pre_download=pre_download).patches: - if patch.version == version: - return patch - return None - - def repair_file( - self, - file: PathLike, - pre_download: bool = False, - game_info: resource.Game = None, - ) -> None: - """ - Repairs a game file. - - This will automatically handle backup and restore the file if the repair - fails. - - Args: - file (PathLike): The file to repair. - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - """ - return self.repair_files([file], pre_download=pre_download, game_info=game_info) - - def repair_files( - self, - files: list[PathLike], - pre_download: bool = False, - game_info: resource.Game = None, - ) -> None: - """ - Repairs multiple game files. - - This will automatically handle backup and restore the file if the repair - fails. - - Args: - files (PathLike): The files to repair. - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - game_info (resource.Game): The game information to use for repair. - """ - functions.repair_files( - self, files, pre_download=pre_download, game_info=game_info - ) - - def repair_game(self) -> None: - """ - Tries to repair the game by reading "pkg_version" file and downloading the - mismatched files from the server. - """ - functions.repair_game(self) - - def install_archive(self, archive_file: PathLike | IOBase) -> None: - """ - Applies an install archive to the game, it can be the game itself or a - voicepack one. - - `archive_file` can be a path to the archive file or a file-like object, - like if you have very high amount of RAM and want to download the archive - to memory instead of disk, this can be useful for you. - - Args: - archive_file (PathLike | IOBase): The archive file. - """ - if not isinstance(archive_file, IOBase): - archive_file = Path(archive_file) - functions.install_archive(self, archive_file) - - def apply_update_archive( - self, archive_file: PathLike | IOBase, auto_repair: bool = True - ) -> None: - """ - Applies an update archive to the game, it can be the game update or a - voicepack update. - - `archive_file` can be a path to the archive file or a file-like object, - like if you have very high amount of RAM and want to download the update - to memory instead of disk, this can be useful for you. - - `auto_repair` is used to determine whether to repair the file if it's - broken. If it's set to False, then it'll raise an exception if the file - is broken. - - Args: - archive_file (PathLike | IOBase): The archive file. - auto_repair (bool, optional): Whether to repair the file if it's broken. - Defaults to True. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - if not isinstance(archive_file, IOBase): - archive_file = Path(archive_file) - # Hello hell again, dealing with HDiffPatch and all the things again. - functions.apply_update_archive(self, archive_file, auto_repair=auto_repair) - - def install_update( - self, update_info: resource.Patch = None, auto_repair: bool = True - ): - """ - Installs an update from a `Patch` object. - - You may want to download the update manually and pass it to - `apply_update_archive()` instead for better control, and after that - execute `set_version_config()` to set the game version. - - Args: - update_info (Diff, optional): The update information. Defaults to None. - auto_repair (bool, optional): Whether to repair the file if it's broken. - Defaults to True. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - if not update_info: - update_info = self.get_update() - if not update_info or update_info.version == self.get_version_str(): - raise GameAlreadyUpdatedError("Game is already updated.") - update_url = update_info.game_pkgs[0].url - # Base game update - archive_file = self.cache.joinpath(PurePath(update_url).name) - download(update_url, archive_file) - self.apply_update_archive(archive_file=archive_file, auto_repair=auto_repair) - # Get installed voicepacks - installed_voicepacks = self.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_info.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = self.cache.joinpath(PurePath(remote_voicepack.url).name) - download(remote_voicepack.url, archive_file) - self.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - self.set_version_config() + super().__init__(GameType.HSR, path, cache_path) diff --git a/vollerei/zzz/launcher/game.py b/vollerei/zzz/launcher/game.py index 5f67963..84c4876 100644 --- a/vollerei/zzz/launcher/game.py +++ b/vollerei/zzz/launcher/game.py @@ -1,443 +1,16 @@ -from configparser import ConfigParser -from io import IOBase from os import PathLike -from pathlib import Path, PurePath -from vollerei.abc.launcher.game import GameABC -from vollerei.common import ConfigFile, functions -from vollerei.common.api import resource -from vollerei.common.enums import VoicePackLanguage, GameChannel -from vollerei.exceptions.game import ( - GameAlreadyUpdatedError, - GameNotInstalledError, - PreDownloadNotAvailable, -) -from vollerei.zzz.launcher import api -from vollerei import paths -from vollerei.utils import download +from vollerei.game.launcher.manager import Game as CommonGame +from vollerei.common.enums import GameType -class Game(GameABC): +class Game(CommonGame): """ Manages the game installation - Since channel detection isn't (properly) implemented yet, most functions assume you're + Since channel detection isn't implemented yet, most functions assume you're using the overseas version of the game. You can override channel by setting the property `channel_override` to the channel you want to use. """ def __init__(self, path: PathLike = None, cache_path: PathLike = None): - self._path: Path | None = Path(path) if path else None - if not cache_path: - cache_path = paths.cache_path - cache_path = Path(cache_path) - self.cache: Path = cache_path.joinpath("game/zzz/") - self.cache.mkdir(parents=True, exist_ok=True) - self._version_override: tuple[int, int, int] | None = None - self._channel_override: GameChannel | None = None - - @property - def version_override(self) -> tuple[int, int, int] | None: - """ - Overrides the game version. - - This can be useful if you want to override the version of the game - and additionally working around bugs. - """ - return self._version_override - - @version_override.setter - def version_override(self, version: tuple[int, int, int] | str | None): - if isinstance(version, str): - version = tuple(int(i) for i in version.split(".")) - self._version_override = version - - @property - def channel_override(self) -> GameChannel | None: - """ - Overrides the game channel. - - Because game channel detection isn't implemented yet, you may need - to use this for some functions to work. - - This can be useful if you want to override the channel of the game - and additionally working around bugs. - """ - return self._channel_override - - @channel_override.setter - def channel_override(self, channel: GameChannel | str | None): - if isinstance(channel, str): - channel = GameChannel[channel] - self._channel_override = channel - - @property - def path(self) -> Path | None: - """ - Paths to the game folder. - """ - return self._path - - @path.setter - def path(self, path: PathLike): - self._path = Path(path) - - def data_folder(self) -> Path: - """ - Paths to the game data folder. - """ - try: - return self._path.joinpath("ZenlessZoneZero_Data") - except AttributeError: - raise GameNotInstalledError("Game path is not set.") - - def is_installed(self) -> bool: - """ - Checks if the game is installed. - - Returns: - bool: True if the game is installed, False otherwise. - """ - if self._path is None: - return False - try: - self.get_channel() - if self.data_folder().is_dir(): - return True - except GameNotInstalledError: - return False - if self.get_version() == (0, 0, 0): - return False - return True - - def get_channel(self) -> GameChannel: - """ - THIS METHOD CURRENTLY DOESN'T WORK YET. - - It'll return the overridden channel if set, otherwise it'll return the - overseas channel. - - Gets the current game channel. - - Returns: - GameChannel: The current game channel. - """ - return self._channel_override or GameChannel.Overseas - - def get_version_config(self) -> tuple[int, int, int]: - """ - Gets the current installed game version from config.ini. - - Using this is not recommended, as only official launcher creates - and uses this file, instead you should use `get_version()`. - - This returns (0, 0, 0) if the version could not be found. - - Returns: - tuple[int, int, int]: Game version. - """ - cfg_file = self._path.joinpath("config.ini") - if not cfg_file.exists(): - return (0, 0, 0) - cfg = ConfigFile(cfg_file) - # Fk u miHoYo - if "general" in cfg.sections(): - version_str = cfg.get("general", "game_version", fallback="0.0.0") - elif "General" in cfg.sections(): - version_str = cfg.get("General", "game_version", fallback="0.0.0") - else: - return (0, 0, 0) - if version_str.count(".") != 2: - return (0, 0, 0) - try: - version = tuple(int(i) for i in version_str.split(".")) - except Exception: - return (0, 0, 0) - return version - - def set_version_config(self): - """ - Sets the current installed game version to config.ini. - - This method is meant to keep compatibility with the official launcher only. - """ - cfg_file = self._path.joinpath("config.ini") - if cfg_file.exists(): - cfg = ConfigFile(cfg_file) - cfg.set("general", "game_version", self.get_version_str()) - cfg.save() - else: - cfg = ConfigParser() - cfg.read_dict( - { - "general": { - "downloading_mode": None, - "channel": 1, - "cps": "hyp_hoyoverse", - "game_version": self.get_version_str(), - "sub_channel": 0, - # This is probably should be fetched from the server but well - "plugin_cuqph0fsfw_version": "1.0.0", - # What's this in the Chinese version? - "uapc": { - "nap_global": {"uapc": "f55586a8ce9f_"}, - "hyp": {"uapc": "f55586a8ce9f_"}, - }, # Honestly what's this? - } - } - ) - cfg.write(cfg_file.open("w")) - - def get_version(self) -> tuple[int, int, int]: - """ - Gets the current installed game version. - - Credits to An Anime Team for the code that does the magic: - https://github.com/an-anime-team/anime-game-core/blob/main/src/games/zzz/game.rs#L49 - - If the above method fails, it'll fallback to read the config.ini file - for the version, which is not recommended (as described in - `get_version_config()` docs) - - This returns (0, 0, 0) if the version could not be found - (usually indicates the game is not installed), and in fact `is_installed()` uses - this method to check if the game is installed too. - - Returns: - tuple[int, int, int]: The version as a tuple of integers. - """ - - data_file = self.data_folder().joinpath("globalgamemanagers") - if not data_file.exists(): - return self.get_version_config() - - def bytes_to_int(byte_array: list[bytes]) -> int: - bytes_as_int = int.from_bytes(byte_array, byteorder="big") - actual_int = bytes_as_int - 48 # 48 is the ASCII code for 0 - return actual_int - - version_bytes: list[list[bytes]] = [[], [], []] - version_ptr = 0 - correct = True - try: - with data_file.open("rb") as f: - f.seek(4000) - for byte in f.read(10000): - match byte: - case 0: - if ( - correct - and len(version_bytes[0]) > 0 - and len(version_bytes[1]) > 0 - and len(version_bytes[2]) > 0 - ): - found_version = tuple( - bytes_to_int(i) for i in version_bytes - ) - return found_version - version_bytes = [[], [], []] - version_ptr = 0 - correct = True - case b".": - version_ptr += 1 - if version_ptr > 2: - correct = False - case _: - if correct and byte in b"0123456789": - version_bytes[version_ptr].append(byte) - else: - correct = False - except Exception: - pass - # Fallback to config.ini - return self.get_version_config() - - def get_version_str(self) -> str: - """ - Gets the current installed game version as a string. - - Because this method uses `get_version()`, you should read the docs of - that method too. - - Returns: - str: The version as a string. - """ - return ".".join(str(i) for i in self.get_version()) - - def get_installed_voicepacks(self) -> list[VoicePackLanguage]: - """ - Gets the installed voicepacks. - - Returns: - list[VoicePackLanguage]: A list of installed voicepacks. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - voicepacks = [] - for child in ( - self.data_folder().joinpath("StreamingAssets/Audio/Windows/Full/").iterdir() - ): - if child.resolve().is_dir(): - try: - voicepacks.append(VoicePackLanguage.from_zzz_name(child.name)) - except ValueError: - pass - return voicepacks - - def get_remote_game( - self, pre_download: bool = False - ) -> resource.Main | resource.PreDownload: - """ - Gets the current game information from remote. - - Args: - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - - Returns: - A `Main` or `PreDownload` object that contains the game information. - """ - channel = self._channel_override or self.get_channel() - if pre_download: - game = api.get_game_package(channel=channel).pre_download - if not game: - raise PreDownloadNotAvailable("Pre-download version is not available.") - return game - return api.get_game_package(channel=channel).main - - def get_update(self, pre_download: bool = False) -> resource.Patch | None: - """ - Gets the current game update. - - Args: - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - - Returns: - A `Patch` object that contains the update information or - `None` if the game is not installed or already up-to-date. - """ - if not self.is_installed(): - return None - version = ( - ".".join(str(x) for x in self._version_override) - if self._version_override - else self.get_version_str() - ) - for patch in self.get_remote_game(pre_download=pre_download).patches: - if patch.version == version: - return patch - return None - - def repair_file( - self, - file: PathLike, - pre_download: bool = False, - game_info: resource.Game = None, - ) -> None: - """ - Repairs a game file. - - This will automatically handle backup and restore the file if the repair - fails. - - Args: - file (PathLike): The file to repair. - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - """ - return self.repair_files([file], pre_download=pre_download, game_info=game_info) - - def repair_files( - self, - files: list[PathLike], - pre_download: bool = False, - game_info: resource.Game = None, - ) -> None: - """ - Repairs multiple game files. - - This will automatically handle backup and restore the file if the repair - fails. - - Args: - files (PathLike): The file to repair. - pre_download (bool): Whether to get the pre-download version. - Defaults to False. - """ - functions.repair_files( - self, files, pre_download=pre_download, game_info=game_info - ) - - def repair_game(self) -> None: - """ - Tries to repair the game by reading "pkg_version" file and downloading the - mismatched files from the server. - """ - functions.repair_game(self) - - def apply_update_archive( - self, archive_file: PathLike | IOBase, auto_repair: bool = True - ) -> None: - """ - Applies an update archive to the game, it can be the game update or a - voicepack update. - - `archive_file` can be a path to the archive file or a file-like object, - like if you have very high amount of RAM and want to download the update - to memory instead of disk, this can be useful for you. - - `auto_repair` is used to determine whether to repair the file if it's - broken. If it's set to False, then it'll raise an exception if the file - is broken. - - Args: - archive_file (PathLike | IOBase): The archive file. - auto_repair (bool, optional): Whether to repair the file if it's broken. - Defaults to True. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - if not isinstance(archive_file, IOBase): - archive_file = Path(archive_file) - # Hello hell again, dealing with HDiffPatch and all the things again. - functions.apply_update_archive(self, archive_file, auto_repair=auto_repair) - - def install_update( - self, update_info: resource.Patch = None, auto_repair: bool = True - ): - """ - Installs an update from a `Patch` object. - - You may want to download the update manually and pass it to - `apply_update_archive()` instead for better control, and after that - execute `set_version_config()` to set the game version. - - Args: - update_info (Diff, optional): The update information. Defaults to None. - auto_repair (bool, optional): Whether to repair the file if it's broken. - Defaults to True. - """ - if not self.is_installed(): - raise GameNotInstalledError("Game is not installed.") - if not update_info: - update_info = self.get_update() - if not update_info or update_info.version == self.get_version_str(): - raise GameAlreadyUpdatedError("Game is already updated.") - update_url = update_info.game_pkgs[0].url - # Base game update - archive_file = self.cache.joinpath(PurePath(update_url).name) - download(update_url, archive_file) - self.apply_update_archive(archive_file=archive_file, auto_repair=auto_repair) - # Get installed voicepacks - installed_voicepacks = self.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_info.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = self.cache.joinpath(PurePath(remote_voicepack.url).name) - download(remote_voicepack.url, archive_file) - self.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - self.set_version_config() + super().__init__(GameType.ZZZ, path, cache_path)