feat: unify multiple games into one

This commit is contained in:
2025-02-01 23:51:53 +07:00
parent fba6cecc38
commit 976308ac85
16 changed files with 1077 additions and 1360 deletions

View File

@ -0,0 +1,7 @@
# Game
::: vollerei.game.launcher.Game
handler: python
options:
show_root_heading: true
show_source: true

View File

@ -114,6 +114,20 @@ class GameABC(ABC):
""" """
pass 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: def get_update(self) -> resource.Patch | None:
""" """
Get the game update Get the game update

View File

@ -1,6 +1,13 @@
from enum import Enum from enum import Enum
class GameType(Enum):
Genshin = 0
HSR = 1
ZZZ = 3
HI3 = 4
class GameChannel(Enum): class GameChannel(Enum):
Overseas = 0 Overseas = 0
China = 1 China = 1

View File

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

View File

@ -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/"

View File

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

View File

@ -0,0 +1,5 @@
# Re-exports
from vollerei.game.launcher.manager import Game
__all__ = ["Game"]

View File

@ -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

View File

View File

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

223
vollerei/game/patcher.py Normal file
View File

@ -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)

View File

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

View File

@ -1,447 +1,12 @@
from configparser import ConfigParser
from io import IOBase
from os import PathLike from os import PathLike
from pathlib import Path, PurePath from vollerei.game.launcher.manager import Game as CommonGame
from vollerei.abc.launcher.game import GameABC from vollerei.common.enums import GameType
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
class Game(GameABC): class Game(CommonGame):
""" """
Manages the game installation Manages the game installation
""" """
def __init__(self, path: PathLike = None, cache_path: PathLike = None): def __init__(self, path: PathLike = None, cache_path: PathLike = None):
self._path: Path | None = Path(path) if path else None super().__init__(GameType.Genshin, path, cache_path)
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()

View File

@ -1,15 +1 @@
MD5SUMS = { from vollerei.game.hsr.constants import * # noqa: F403 because we just want to re-export
"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/"

View File

@ -1,24 +1,9 @@
from configparser import ConfigParser
from hashlib import md5
from io import IOBase
from os import PathLike from os import PathLike
from pathlib import Path, PurePath from vollerei.game.launcher.manager import Game as CommonGame
from vollerei.abc.launcher.game import GameABC from vollerei.common.enums import GameType
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
class Game(GameABC): class Game(CommonGame):
""" """
Manages the game installation Manages the game installation
@ -28,459 +13,4 @@ class Game(GameABC):
""" """
def __init__(self, path: PathLike = None, cache_path: PathLike = None): def __init__(self, path: PathLike = None, cache_path: PathLike = None):
self._path: Path | None = Path(path) if path else None super().__init__(GameType.HSR, path, cache_path)
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()

View File

@ -1,443 +1,16 @@
from configparser import ConfigParser
from io import IOBase
from os import PathLike from os import PathLike
from pathlib import Path, PurePath from vollerei.game.launcher.manager import Game as CommonGame
from vollerei.abc.launcher.game import GameABC from vollerei.common.enums import GameType
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
class Game(GameABC): class Game(CommonGame):
""" """
Manages the game installation 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 using the overseas version of the game. You can override channel by setting
the property `channel_override` to the channel you want to use. the property `channel_override` to the channel you want to use.
""" """
def __init__(self, path: PathLike = None, cache_path: PathLike = None): def __init__(self, path: PathLike = None, cache_path: PathLike = None):
self._path: Path | None = Path(path) if path else None super().__init__(GameType.ZZZ, path, cache_path)
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()