From 4a7bc3d0b478d28ccebfad2181ce45dee49c7c93 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20Th=E1=BA=BF=20H=C6=B0ng?= Date: Wed, 18 Dec 2024 15:20:24 +0700 Subject: [PATCH] chore(cli): migrate to one file Easier to maintain but slightly slower execution speed lol --- vollerei/abc/launcher/game.py | 2 +- vollerei/cli/__init__.py | 6 +- vollerei/cli/commands.py | 20 +- vollerei/cli/genshin.py | 747 -------------------------- vollerei/cli/hsr.py | 954 ---------------------------------- 5 files changed, 16 insertions(+), 1713 deletions(-) delete mode 100644 vollerei/cli/genshin.py delete mode 100644 vollerei/cli/hsr.py diff --git a/vollerei/abc/launcher/game.py b/vollerei/abc/launcher/game.py index 6256132..08d64d9 100644 --- a/vollerei/abc/launcher/game.py +++ b/vollerei/abc/launcher/game.py @@ -129,7 +129,7 @@ class GameABC(ABC): """ pass - def get_remote_game(self, pre_download: bool = False) -> resource.Game: + def get_remote_game(self, pre_download: bool = False) -> resource.Main | resource.PreDownload: """ Gets the current game information from remote. diff --git a/vollerei/cli/__init__.py b/vollerei/cli/__init__.py index bf9c2c5..6ad0499 100644 --- a/vollerei/cli/__init__.py +++ b/vollerei/cli/__init__.py @@ -1,9 +1,9 @@ from cleo.application import Application -from vollerei.cli import hsr, genshin +from vollerei.cli import commands application = Application() -for command in hsr.commands + genshin.commands: - application.add(command()) +for command in commands.exports: + application.add(command) def run(): diff --git a/vollerei/cli/commands.py b/vollerei/cli/commands.py index 960b2d3..0596e31 100644 --- a/vollerei/cli/commands.py +++ b/vollerei/cli/commands.py @@ -1,4 +1,3 @@ -# THIS FILE CURRENTLY DOESN'T WORK YET. import copy import traceback from cleo.commands.command import Command @@ -130,7 +129,7 @@ class VoicepackList(Command): remote_game = State.game.get_remote_game(pre_download=pre_download) available_voicepacks_str = [ f"{x.language.name} ({x.language.value})" - for x in remote_game.latest.voice_packs + for x in remote_game.major.audio_pkgs ] self.line(f"Available voicepacks: {', '.join(available_voicepacks_str)}") @@ -939,7 +938,7 @@ class ApplyUpdateArchive(Command): set_version_config() # This is the list for HSR commands, we'll add Genshin commands later -hsr_exports = [ +classes = [ ApplyInstallArchive, ApplyUpdateArchive, GetVersionCommand, @@ -959,10 +958,15 @@ hsr_exports = [ VoicepackUpdate, ] exports = [] -for command in hsr_exports: - exports.append(command) +for command in classes: + init_command = command() + exports.append(init_command) if "patch" in command.name: continue - new_command = copy.deepcopy(command) - new_command.name = f"genshin {new_command.name[4:]}" - exports.append(new_command) + command_name = command.name[4:] + genshin_init_command = copy.deepcopy(init_command) + genshin_init_command.name = f"genshin {command_name}" + exports.append(genshin_init_command) + zzz_init_command = copy.deepcopy(init_command) + zzz_init_command.name = f"zzz {command_name}" + exports.append(zzz_init_command) diff --git a/vollerei/cli/genshin.py b/vollerei/cli/genshin.py deleted file mode 100644 index 3d32494..0000000 --- a/vollerei/cli/genshin.py +++ /dev/null @@ -1,747 +0,0 @@ -import traceback -from cleo.commands.command import Command -from cleo.helpers import option, argument -from pathlib import PurePath -from vollerei.common.enums import GameChannel, VoicePackLanguage -from vollerei.cli import utils -from vollerei.exceptions.game import GameError -from vollerei.genshin import Game -from vollerei import paths - - -default_options = [ - option("channel", "c", description="Game channel.", flag=False, default="overseas"), - option("force", "f", description="Force the command to run."), - option( - "game-path", - "g", - description="Path to the game installation.", - flag=False, - default=".", - ), - option("temporary-path", "t", description="Temporary path.", flag=False), - option("silent", "s", description="Silent mode."), - option("noconfirm", "y", description="Do not ask for confirmation (yes to all)."), -] - - -class State: - game: Game = None - - -def callback( - command: Command, -): - """ - Base callback for all commands - """ - game_path = command.option("game-path") - channel = command.option("channel") - silent = command.option("silent") - noconfirm = command.option("noconfirm") - temporary_path = command.option("temporary-path") - if isinstance(channel, str): - channel = GameChannel[channel.capitalize()] - elif isinstance(channel, int): - channel = GameChannel(channel) - if temporary_path: - paths.set_base_path(temporary_path) - State.game = Game(game_path, temporary_path) - if channel: - State.game.channel_override = channel - utils.silent_message = silent - if noconfirm: - utils.no_confirm = noconfirm - - def confirm( - question: str, default: bool = False, true_answer_regex: str = r"(?i)^y" - ): - command.line( - f"{question} (yes/no) [{'yes' if default else 'no'}] y" - ) - return True - - command.confirm = confirm - command.add_style("warn", fg="yellow") - - -def set_version_config(self: Command): - self.line("Setting version config... ") - try: - State.game.set_version_config() - except Exception as e: - self.line_error(f"Couldn't set version config: {e}") - self.line_error( - "This won't affect the overall experience, but if you're using the official launcher" - ) - self.line_error( - "you may have to edit the file 'config.ini' manually to reflect the latest version." - ) - self.line( - f"The game has been updated to version: {State.game.get_version_str()}" - ) - - -class VoicepackListInstalled(Command): - name = "genshin voicepack list-installed" - description = "Gets the installed voicepacks." - options = default_options - - def handle(self): - callback(command=self) - installed_voicepacks_str = [ - f"{x.name}" - for x in State.game.get_installed_voicepacks() - ] - self.line(f"Installed voicepacks: {', '.join(installed_voicepacks_str)}") - - -class VoicepackList(Command): - name = "genshin voicepack list" - description = "Gets all available voicepacks." - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - remote_game = State.game.get_remote_game(pre_download=pre_download) - available_voicepacks_str = [ - f"{x.language.name} ({x.language.value})" - for x in remote_game.latest.voice_packs - ] - self.line(f"Available voicepacks: {', '.join(available_voicepacks_str)}") - - -class VoicepackInstall(Command): - name = "genshin voicepack install" - description = ( - "Installs the specified installed voicepacks." - ) - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - arguments = [ - argument( - "language", description="Languages to install", multiple=True, optional=True - ) - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - # Typing manually because pylance detect it as Any - languages: list[str] = self.argument("language") - # Get installed voicepacks - language_objects = [] - for language in languages: - language = language.lower() - try: - language_objects.append(VoicePackLanguage[language.capitalize()]) - except KeyError: - try: - language_objects.append(VoicePackLanguage.from_remote_str(language)) - except ValueError: - self.line_error(f"Invalid language: {language}") - if len(language_objects) == 0: - self.line_error( - "No valid languages specified, you must specify a language to install" - ) - return - progress = utils.ProgressIndicator(self) - progress.start("Fetching install package information... ") - try: - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Fetching failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - "Installation information fetched successfully." - ) - if not self.confirm("Do you want to install the specified voicepacks?"): - self.line("Installation aborted.") - return - # Voicepack update - for remote_voicepack in game_info.major.audio_pkgs: - if remote_voicepack.language not in language_objects: - continue - self.line( - f"Downloading install package for language: {remote_voicepack.language.name}... " - ) - archive_file = State.game.cache.joinpath(PurePath(remote_voicepack.url).name) - try: - download_result = utils.download( - remote_voicepack.url, archive_file, file_len=remote_voicepack.size - ) - except Exception as e: - self.line_error(f"Couldn't download package: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Installing package...") - try: - State.game.install_archive(archive_file) - except Exception as e: - progress.finish( - f"Couldn't apply package: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - f"Package applied for language {remote_voicepack.language.name}." - ) - self.line( - f"The voicepacks have been installed to version: {State.game.get_version_str()}" - ) - - -class VoicepackUpdate(Command): - name = "genshin voicepack update" - description = ( - "Updates the specified installed voicepacks, if not specified, updates all." - ) - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - option("pre-download", description="Pre-download the game if available."), - option( - "from-version", description="Update from a specific version", flag=False - ), - ] - arguments = [ - argument( - "language", description="Languages to update", multiple=True, optional=True - ) - ] - - def handle(self): - callback(command=self) - auto_repair = self.option("auto-repair") - pre_download = self.option("pre-download") - from_version = self.option("from-version") - # Typing manually because pylance detect it as Any - languages: list[str] = self.argument("language") - if auto_repair: - self.line("Auto-repair is enabled.") - if from_version: - self.line(f"Updating from version: {from_version}") - State.game.version_override = from_version - # Get installed voicepacks - if len(languages) == 0: - self.line( - "No languages specified, updating all installed voicepacks..." - ) - installed_voicepacks = State.game.get_installed_voicepacks() - if len(languages) > 0: - languages = [x.lower() for x in languages] - # Support both English and en-us and en - installed_voicepacks = [ - x - for x in installed_voicepacks - if x.name.lower() in languages - or x.value.lower() in languages - or x.name.lower()[:2] in languages - ] - installed_voicepacks_str = [ - f"{str(x.name)}" for x in installed_voicepacks - ] - self.line(f"Updating voicepacks: {', '.join(installed_voicepacks_str)}") - progress = utils.ProgressIndicator(self) - progress.start("Checking for updates... ") - try: - update_diff = State.game.get_update(pre_download=pre_download) - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Update checking failed with following error: {e} \n{traceback.format_exc()}" - ) - return - if update_diff is None: - progress.finish("Game is already updated.") - return - progress.finish("Update available.") - self.line( - f"The current version is: {State.game.get_version_str()}" - ) - self.line( - f"The latest version is: {game_info.latest.version}" - ) - if not self.confirm("Do you want to update the game?"): - self.line("Update aborted.") - return - # Voicepack update - for remote_voicepack in update_diff.voice_packs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - self.line( - f"Downloading update package for language: {remote_voicepack.language.name}... " - ) - archive_file = State.game.cache.joinpath(remote_voicepack.name) - try: - download_result = utils.download( - remote_voicepack.path, archive_file, file_len=update_diff.size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - f"Update applied for language {remote_voicepack.language.name}." - ) - set_version_config(self=self) - self.line( - f"The game has been updated to version: {State.game.get_version_str()}" - ) - - -class GetVersionCommand(Command): - name = "genshin version" - description = "Gets the local game version." - options = default_options - - def handle(self): - callback(command=self) - try: - self.line( - f"Version: {'.'.join(str(x) for x in State.game.get_version())}" - ) - except GameError as e: - self.line_error(f"Couldn't get game version: {e}") - - -class InstallCommand(Command): - name = "genshin install" - description = ( - "Installs the latest version of the game to the specified path (default: current directory). " - + "Note that this will not install the default voicepack (English), you need to install it manually." - ) - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - progress = utils.ProgressIndicator(self) - progress.start("Fetching install package information... ") - try: - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Fetching failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - "Installation information fetched successfully." - ) - if not self.confirm("Do you want to install the game?"): - self.line("Installation aborted.") - return - self.line("Downloading install package...") - first_pkg_out_path = None - for game_pkg in game_info.major.game_pkgs: - out_path = State.game.cache.joinpath(PurePath(game_pkg.url).name) - if not first_pkg_out_path: - first_pkg_out_path = out_path - try: - download_result = utils.download( - game_pkg.url, out_path, file_len=game_pkg.size - ) - except Exception as e: - self.line_error( - f"Couldn't download install package: {e}" - ) - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Installing package...") - try: - State.game.install_archive(first_pkg_out_path) - except Exception as e: - progress.finish( - f"Couldn't install package: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Package applied for the base game.") - self.line("Setting version config... ") - State.game.version_override = game_info.major.version - set_version_config() - State.game.version_override = None - self.line( - f"The game has been installed to version: {State.game.get_version_str()}" - ) - - -class UpdateCommand(Command): - name = "genshin update" - description = "Updates the local game if available" - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - option("pre-download", description="Pre-download the game if available."), - option( - "from-version", description="Update from a specific version", flag=False - ), - ] - - def handle(self): - callback(command=self) - auto_repair = self.option("auto-repair") - pre_download = self.option("pre-download") - from_version = self.option("from-version") - if auto_repair: - self.line("Auto-repair is enabled.") - if from_version: - self.line(f"Updating from version: {from_version}") - State.game.version_override = from_version - progress = utils.ProgressIndicator(self) - progress.start("Checking for updates... ") - try: - update_diff = State.game.get_update(pre_download=pre_download) - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Update checking failed with following error: {e} \n{traceback.format_exc()}" - ) - return - if update_diff is None or isinstance(game_info.major, str | None): - progress.finish("Game is already updated.") - return - progress.finish("Update available.") - self.line( - f"The current version is: {State.game.get_version_str()}" - ) - self.line( - f"The latest version is: {game_info.major.version}" - ) - if not self.confirm("Do you want to update the game?"): - self.line("Update aborted.") - return - self.line("Downloading update package...") - update_game_url = update_diff.game_pkgs[0].url - out_path = State.game.cache.joinpath(PurePath(update_game_url).name) - try: - download_result = utils.download( - update_game_url, out_path, file_len=update_diff.game_pkgs[0].size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive(out_path, auto_repair=auto_repair) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Update applied for base game.") - # Get installed voicepacks - installed_voicepacks = State.game.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_diff.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = State.game.cache.joinpath( - PurePath(remote_voicepack.url).name - ) - self.line( - f"Downloading update package for voicepack language '{remote_voicepack.language.name}'..." - ) - try: - download_result = utils.download( - remote_voicepack.url, archive_file, file_len=remote_voicepack.size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - f"Update applied for language {remote_voicepack.language.name}." - ) - self.line("Setting version config... ") - State.game.version_override = game_info.major.version - set_version_config() - State.game.version_override = None - self.line( - f"The game has been updated to version: {State.game.get_version_str()}" - ) - - -class RepairCommand(Command): - name = "genshin repair" - description = "Tries to repair the local game" - options = default_options - - def handle(self): - callback(command=self) - self.line( - "This command will try to repair the game by downloading missing/broken files." - ) - self.line( - "There will be no progress available, so please be patient and just wait." - ) - if not self.confirm( - "Do you want to repair the game (this will take a long time!)?" - ): - self.line("Repairation aborted.") - return - progress = utils.ProgressIndicator(self) - progress.start("Repairing game files (no progress available)... ") - try: - State.game.repair_game() - except Exception as e: - progress.finish( - f"Repairation failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Repairation completed.") - - -class InstallDownloadCommand(Command): - name = "genshin install download" - description = ( - "Downloads the latest version of the game. " - + "Note that this will not download the default voicepack (English), you need to download it manually." - ) - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - progress = utils.ProgressIndicator(self) - progress.start("Fetching install package information... ") - try: - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Fetching failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - "Installation information fetched successfully." - ) - if not self.confirm("Do you want to download the game?"): - self.line("Download aborted.") - return - self.line("Downloading install package...") - first_pkg_out_path = None - for game_pkg in game_info.major.game_pkgs: - out_path = State.game.cache.joinpath(PurePath(game_pkg.url).name) - if not first_pkg_out_path: - first_pkg_out_path = out_path - try: - download_result = utils.download( - game_pkg.url, out_path, file_len=game_pkg.size - ) - except Exception as e: - self.line_error( - f"Couldn't download install package: {e}" - ) - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - - -class UpdateDownloadCommand(Command): - name = "genshin update download" - description = "Download the update for the local game if available" - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - option("pre-download", description="Pre-download the game if available."), - option( - "from-version", description="Update from a specific version", flag=False - ), - ] - - def handle(self): - callback(command=self) - auto_repair = self.option("auto-repair") - pre_download = self.option("pre-download") - from_version = self.option("from-version") - if auto_repair: - self.line("Auto-repair is enabled.") - if from_version: - self.line(f"Updating from version: {from_version}") - State.game.version_override = from_version - progress = utils.ProgressIndicator(self) - progress.start("Checking for updates... ") - try: - update_diff = State.game.get_update(pre_download=pre_download) - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Update checking failed with following error: {e} \n{traceback.format_exc()}" - ) - return - if update_diff is None or isinstance(game_info.major, str | None): - progress.finish("Game is already updated.") - return - progress.finish("Update available.") - self.line( - f"The current version is: {State.game.get_version_str()}" - ) - self.line( - f"The latest version is: {game_info.major.version}" - ) - if not self.confirm("Do you want to download the update?"): - self.line("Download aborted.") - return - self.line("Downloading update package...") - update_game_url = update_diff.game_pkgs[0].url - out_path = State.game.cache.joinpath(PurePath(update_game_url).name) - try: - download_result = utils.download( - update_game_url, out_path, file_len=update_diff.game_pkgs[0].size - ) - except Exception as e: - self.line_error( - f"Couldn't download update: {e} \n{traceback.format_exc()}" - ) - return - - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - # Get installed voicepacks - installed_voicepacks = State.game.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_diff.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = State.game.cache.joinpath( - PurePath(remote_voicepack.url).name - ) - self.line( - f"Downloading update package for voicepack language '{remote_voicepack.language.name}'..." - ) - try: - download_result = utils.download( - remote_voicepack.url, archive_file, file_len=remote_voicepack.size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - - -class ApplyInstallArchive(Command): - name = "genshin install apply-archive" - description = "Applies the install archive" - arguments = [argument("path", description="Path to the install archive")] - options = default_options - - def handle(self): - callback(command=self) - install_archive = self.argument("path") - progress = utils.ProgressIndicator(self) - progress.start("Applying install package...") - try: - State.game.install_archive(install_archive) - except Exception as e: - progress.finish( - f"Couldn't apply package: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Package applied.") - set_version_config(self=self) - - -class ApplyUpdateArchive(Command): - name = "genshin update apply-archive" - description = "Applies the update archive to the local game" - arguments = [argument("path", description="Path to the update archive")] - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - ] - - def handle(self): - callback(command=self) - update_archive = self.argument("path") - auto_repair = self.option("auto-repair") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive(update_archive, auto_repair=auto_repair) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Update applied.") - set_version_config() - - -commands = [ - ApplyInstallArchive, - ApplyUpdateArchive, - GetVersionCommand, - InstallCommand, - InstallDownloadCommand, - RepairCommand, - UpdateCommand, - UpdateDownloadCommand, - VoicepackInstall, - VoicepackList, - VoicepackListInstalled, - VoicepackUpdate, -] diff --git a/vollerei/cli/hsr.py b/vollerei/cli/hsr.py deleted file mode 100644 index 520e833..0000000 --- a/vollerei/cli/hsr.py +++ /dev/null @@ -1,954 +0,0 @@ -import traceback -from cleo.commands.command import Command -from cleo.helpers import option, argument -from copy import deepcopy -from pathlib import PurePath -from platform import system -from vollerei.common.enums import GameChannel, VoicePackLanguage -from vollerei.cli import utils -from vollerei.exceptions.game import GameError -from vollerei.hsr import Game, Patcher -from vollerei.exceptions.patcher import PatcherError, PatchUpdateError -from vollerei.hsr.patcher import PatchType -from vollerei import paths - -patcher = Patcher() - - -default_options = [ - option("channel", "c", description="Game channel", flag=False, default="overseas"), - option("force", "f", description="Force the command to run"), - option( - "game-path", - "g", - description="Path to the game installation", - flag=False, - default=".", - ), - option("patch-type", "p", description="Patch type", flag=False), - option("temporary-path", "t", description="Temporary path", flag=False), - option("silent", "s", description="Silent mode"), - option("noconfirm", "y", description="Do not ask for confirmation (yes to all)"), -] - - -class State: - game: Game = None - - -def callback( - command: Command, -): - """ - Base callback for all commands - """ - game_path = command.option("game-path") - patch_type = command.option("patch-type") - channel = command.option("channel") - silent = command.option("silent") - noconfirm = command.option("noconfirm") - temporary_path = command.option("temporary-path") - if isinstance(channel, str): - channel = GameChannel[channel.capitalize()] - elif isinstance(channel, int): - channel = GameChannel(channel) - if temporary_path: - paths.set_base_path(temporary_path) - State.game = Game(game_path, temporary_path) - if channel: - State.game.channel_override = channel - if patch_type is None: - patch_type = PatchType.Jadeite - elif isinstance(patch_type, str): - patch_type = PatchType[patch_type] - elif isinstance(patch_type, int): - patch_type = PatchType(patch_type) - patcher.patch_type = patch_type - utils.silent_message = silent - if noconfirm: - utils.no_confirm = noconfirm - - def confirm( - question: str, default: bool = False, true_answer_regex: str = r"(?i)^y" - ): - command.line( - f"{question} (yes/no) [{'yes' if default else 'no'}] y" - ) - return True - - command.confirm = confirm - command.add_style("warn", fg="yellow") - - -def set_version_config(self: Command): - self.line("Setting version config... ") - try: - State.game.set_version_config() - except Exception as e: - self.line_error(f"Couldn't set version config: {e}") - self.line_error( - "This won't affect the overall experience, but if you're using the official launcher" - ) - self.line_error( - "you may have to edit the file 'config.ini' manually to reflect the latest version." - ) - self.line( - f"The game has been updated to version: {State.game.get_version_str()}" - ) - - -class VoicepackListInstalled(Command): - name = "hsr voicepack list-installed" - description = "Get the installed voicepacks" - options = default_options - - def handle(self): - callback(command=self) - installed_voicepacks_str = [ - f"{x.name}" - for x in State.game.get_installed_voicepacks() - ] - self.line(f"Installed voicepacks: {', '.join(installed_voicepacks_str)}") - - -class VoicepackList(Command): - name = "hsr voicepack list" - description = "Get all available voicepacks" - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - remote_game = State.game.get_remote_game(pre_download=pre_download) - available_voicepacks_str = [ - f"{x.language.name} ({x.language.value})" - for x in remote_game.latest.voice_packs - ] - self.line(f"Available voicepacks: {', '.join(available_voicepacks_str)}") - - -class VoicepackInstall(Command): - name = "hsr voicepack install" - description = ( - "Installs the specified installed voicepacks" - ) - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - arguments = [ - argument( - "language", description="Languages to install", multiple=True, optional=True - ) - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - # Typing manually because pylance detect it as Any - languages: list[str] = self.argument("language") - # Get installed voicepacks - language_objects = [] - for language in languages: - language = language.lower() - try: - language_objects.append(VoicePackLanguage[language.capitalize()]) - except KeyError: - try: - language_objects.append(VoicePackLanguage.from_remote_str(language)) - except ValueError: - self.line_error(f"Invalid language: {language}") - if len(language_objects) == 0: - self.line_error( - "No valid languages specified, you must specify a language to install" - ) - return - progress = utils.ProgressIndicator(self) - progress.start("Fetching install package information... ") - try: - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Fetching failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - "Installation information fetched successfully." - ) - if not self.confirm("Do you want to install the specified voicepacks?"): - self.line("Installation aborted.") - return - # Voicepack update - for remote_voicepack in game_info.major.audio_pkgs: - if remote_voicepack.language not in language_objects: - continue - self.line( - f"Downloading install package for language: {remote_voicepack.language.name}... " - ) - archive_file = State.game.cache.joinpath(PurePath(remote_voicepack.url).name) - try: - download_result = utils.download( - remote_voicepack.url, archive_file, file_len=remote_voicepack.size - ) - except Exception as e: - self.line_error(f"Couldn't download package: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Installing package...") - try: - State.game.install_archive(archive_file) - except Exception as e: - progress.finish( - f"Couldn't apply package: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - f"Package applied for language {remote_voicepack.language.name}." - ) - self.line( - f"The voicepacks have been installed to version: {State.game.get_version_str()}" - ) - - -class VoicepackUpdate(Command): - name = "hsr voicepack update" - description = ( - "Updates the specified installed voicepacks, if not specified, updates all" - ) - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - option("pre-download", description="Pre-download the game if available."), - option( - "from-version", description="Update from a specific version", flag=False - ), - ] - arguments = [ - argument( - "language", description="Languages to update", multiple=True, optional=True - ) - ] - - def handle(self): - callback(command=self) - auto_repair = self.option("auto-repair") - pre_download = self.option("pre-download") - from_version = self.option("from-version") - # Typing manually because pylance detect it as Any - languages: list[str] = self.argument("language") - if auto_repair: - self.line("Auto-repair is enabled.") - if from_version: - self.line(f"Updating from version: {from_version}") - State.game.version_override = from_version - # Get installed voicepacks - if len(languages) == 0: - self.line( - "No languages specified, updating all installed voicepacks..." - ) - installed_voicepacks = State.game.get_installed_voicepacks() - if len(languages) > 0: - languages = [x.lower() for x in languages] - # Support both English and en-us and en - installed_voicepacks = [ - x - for x in installed_voicepacks - if x.name.lower() in languages - or x.value.lower() in languages - or x.name.lower()[:2] in languages - ] - installed_voicepacks_str = [ - f"{str(x.name)}" for x in installed_voicepacks - ] - self.line(f"Updating voicepacks: {', '.join(installed_voicepacks_str)}") - progress = utils.ProgressIndicator(self) - progress.start("Checking for updates... ") - try: - update_diff = State.game.get_update(pre_download=pre_download) - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Update checking failed with following error: {e} \n{traceback.format_exc()}" - ) - return - if update_diff is None: - progress.finish("Game is already updated.") - return - progress.finish("Update available.") - self.line( - f"The current version is: {State.game.get_version_str()}" - ) - self.line( - f"The latest version is: {game_info.latest.version}" - ) - if not self.confirm("Do you want to update the game?"): - self.line("Update aborted.") - return - # Voicepack update - for remote_voicepack in update_diff.voice_packs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - self.line( - f"Downloading update package for language: {remote_voicepack.language.name}... " - ) - archive_file = State.game.cache.joinpath(remote_voicepack.name) - try: - download_result = utils.download( - remote_voicepack.path, archive_file, file_len=update_diff.size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - f"Update applied for language {remote_voicepack.language.name}." - ) - set_version_config(self=self) - self.line( - f"The game has been updated to version: {State.game.get_version_str()}" - ) - - -class PatchTypeCommand(Command): - name = "hsr patch type" - description = "Get the patch type of the game" - options = default_options - - def handle(self): - callback(command=self) - self.line(f"Patch type: {patcher.patch_type.name}") - - -class UpdatePatchCommand(Command): - name = "hsr patch update" - description = "Updates the patch" - options = default_options - - def handle(self): - callback(command=self) - progress = utils.ProgressIndicator(self) - progress.start("Updating patch... ") - try: - patcher.update_patch() - except PatchUpdateError as e: - progress.finish( - f"Patch update failed with following error: {e} \n{traceback.format_exc()}" - ) - else: - progress.finish("Patch updated!") - - -class PatchInstallCommand(Command): - name = "hsr patch install" - description = "Installs the patch" - options = default_options - - def jadeite(self): - progress = utils.ProgressIndicator(self) - progress.start("Installing patch... ") - try: - jadeite_dir = patcher.patch_game(game=State.game) - except PatcherError as e: - progress.finish( - f"Patch installation failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Patch installed!") - print() - exe_path = jadeite_dir.joinpath("jadeite.exe") - self.line(f"Jadeite executable is located at: {exe_path}") - self.line( - "You need to run the game using Jadeite to use the patch." - ) - self.line(f'E.g: {exe_path} "{State.game.path}"') - print() - self.line( - "To activate the experimental patching method, set the environment variable BREAK_CATHACK=1" - ) - self.line( - "Read more about it here: https://codeberg.org/mkrsym1/jadeite/issues/37" - ) - print() - self.line( - "Please don't spread this project to public, we just want to play the game." - ) - self.line( - "And for your own sake, please only use test accounts, as there is an extremely high risk of getting banned." - ) - - def astra(self): - progress = utils.ProgressIndicator(self) - progress.start("Installing patch... ") - try: - patcher.patch_game(game=State.game) - except PatcherError as e: - progress.finish( - f"Patch installation failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Patch installed!") - self.line() - self.line( - "Please don't spread this project to public, we just want to play the game." - ) - self.line( - "And for your own sake, please only use testing accounts, as there is an extremely high risk of getting banned." - ) - - def handle(self): - callback(command=self) - if system() == "Windows": - self.line( - "Windows is officialy supported by the game, so no patching is needed." - ) - self.line( - "By patching the game, you are violating the ToS of the game." - ) - if not self.confirm("Do you want to patch the game?"): - self.line("Patching aborted.") - return - progress = utils.ProgressIndicator(self) - progress.start("Checking telemetry hosts... ") - telemetry_list = patcher.check_telemetry() - if telemetry_list: - progress.finish("Telemetry hosts were found.") - self.line("Below is the list of telemetry hosts that need to be blocked:") - print() - for host in telemetry_list: - self.line(f"{host}") - print() - self.line( - "To prevent the game from sending data about the patch, " - + "we need to block these hosts." - ) - if not self.confirm("Do you want to block them?"): - self.line("Patching aborted.") - self.line( - "Please block these hosts manually then try again." - ) - return - try: - patcher.block_telemetry(telemetry_list=telemetry_list) - except Exception: - self.line_error( - f"Couldn't block telemetry hosts: {traceback.format_exc()}" - ) - # There's a good reason for this. - if system() != "Windows": - self.line( - "Cannot continue, please block them manually then try again." - ) - return - self.line("Continuing anyway...") - else: - progress.finish("No telemetry hosts found.") - progress = utils.ProgressIndicator(self) - progress.start("Updating patch... ") - try: - patcher.update_patch() - except PatchUpdateError as e: - progress.finish( - f"Patch update failed with following error: {e} \n{traceback.format_exc()}" - ) - else: - progress.finish("Patch updated.") - match patcher.patch_type: - case PatchType.Jadeite: - self.jadeite() - case PatchType.Astra: - self.astra() - - -PatchCommand = deepcopy(PatchInstallCommand) -PatchCommand.name = "hsr patch" - - -class PatchTelemetryCommand(Command): - name = "hsr patch telemetry" - description = "Checks for telemetry hosts and block them." - options = default_options - - def handle(self): - progress = utils.ProgressIndicator(self) - progress.start("Checking telemetry hosts... ") - telemetry_list = patcher.check_telemetry() - if telemetry_list: - progress.finish("Telemetry hosts were found.") - self.line("Below is the list of telemetry hosts that need to be blocked:") - print() - for host in telemetry_list: - self.line(f"{host}") - print() - self.line( - "To prevent the game from sending data about the patch, " - + "we need to block these hosts." - ) - if not self.confirm("Do you want to block them?"): - self.line("Blocking aborted.") - return - try: - patcher.block_telemetry(telemetry_list=telemetry_list) - except Exception: - self.line_error( - f"Couldn't block telemetry hosts: {traceback.format_exc()}" - ) - else: - progress.finish("No telemetry hosts found.") - - -class GetVersionCommand(Command): - name = "hsr version" - description = "Gets the local game version" - options = default_options - - def handle(self): - callback(command=self) - try: - self.line( - f"Version: {'.'.join(str(x) for x in State.game.get_version())}" - ) - except GameError as e: - self.line_error(f"Couldn't get game version: {e}") - - -class InstallCommand(Command): - name = "hsr install" - description = ( - "Installs the latest version of the game to the specified path (default: current directory). " - + "Note that this will not install the default voicepack (English), you need to install it manually." - ) - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - progress = utils.ProgressIndicator(self) - progress.start("Fetching install package information... ") - try: - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Fetching failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - "Installation information fetched successfully." - ) - if not self.confirm("Do you want to install the game?"): - self.line("Installation aborted.") - return - self.line("Downloading install package...") - first_pkg_out_path = None - for game_pkg in game_info.major.game_pkgs: - out_path = State.game.cache.joinpath(PurePath(game_pkg.url).name) - if not first_pkg_out_path: - first_pkg_out_path = out_path - try: - download_result = utils.download( - game_pkg.url, out_path, file_len=game_pkg.size - ) - except Exception as e: - self.line_error( - f"Couldn't download install package: {e}" - ) - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Installing package...") - try: - State.game.install_archive(first_pkg_out_path) - except Exception as e: - progress.finish( - f"Couldn't install package: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Package applied for the base game.") - self.line("Setting version config... ") - State.game.version_override = game_info.major.version - set_version_config() - State.game.version_override = None - self.line( - f"The game has been installed to version: {State.game.get_version_str()}" - ) - - -class UpdateCommand(Command): - name = "hsr update" - description = "Updates the local game if available" - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - option("pre-download", description="Pre-download the game if available."), - option( - "from-version", description="Update from a specific version", flag=False - ), - ] - - def handle(self): - callback(command=self) - auto_repair = self.option("auto-repair") - pre_download = self.option("pre-download") - from_version = self.option("from-version") - if auto_repair: - self.line("Auto-repair is enabled.") - if from_version: - self.line(f"Updating from version: {from_version}") - State.game.version_override = from_version - progress = utils.ProgressIndicator(self) - progress.start("Checking for updates... ") - try: - update_diff = State.game.get_update(pre_download=pre_download) - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Update checking failed with following error: {e} \n{traceback.format_exc()}" - ) - return - if update_diff is None or isinstance(game_info.major, str | None): - progress.finish("Game is already updated.") - return - progress.finish("Update available.") - self.line( - f"The current version is: {State.game.get_version_str()}" - ) - self.line( - f"The latest version is: {game_info.major.version}" - ) - if not self.confirm("Do you want to update the game?"): - self.line("Update aborted.") - return - self.line("Downloading update package...") - update_game_url = update_diff.game_pkgs[0].url - out_path = State.game.cache.joinpath(PurePath(update_game_url).name) - try: - download_result = utils.download( - update_game_url, out_path, file_len=update_diff.game_pkgs[0].size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive(out_path, auto_repair=auto_repair) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Update applied for base game.") - # Get installed voicepacks - installed_voicepacks = State.game.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_diff.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = State.game.cache.joinpath( - PurePath(remote_voicepack.url).name - ) - self.line( - f"Downloading update package for voicepack language '{remote_voicepack.language.name}'..." - ) - try: - download_result = utils.download( - remote_voicepack.url, archive_file, file_len=remote_voicepack.size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive( - archive_file=archive_file, auto_repair=auto_repair - ) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - f"Update applied for language {remote_voicepack.language.name}." - ) - self.line("Setting version config... ") - State.game.version_override = game_info.major.version - set_version_config() - State.game.version_override = None - self.line( - f"The game has been updated to version: {State.game.get_version_str()}" - ) - - -class RepairCommand(Command): - name = "hsr repair" - description = "Tries to repair the local game" - options = default_options - - def handle(self): - callback(command=self) - self.line( - "This command will try to repair the game by downloading missing/broken files." - ) - self.line( - "There will be no progress available, so please be patient and just wait." - ) - if not self.confirm( - "Do you want to repair the game (this will take a long time!)?" - ): - self.line("Repairation aborted.") - return - progress = utils.ProgressIndicator(self) - progress.start("Repairing game files (no progress available)... ") - try: - State.game.repair_game() - except Exception as e: - progress.finish( - f"Repairation failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Repairation completed.") - - -class InstallDownloadCommand(Command): - name = "hsr install download" - description = ( - "Downloads the latest version of the game. " - + "Note that this will not download the default voicepack (English), you need to download it manually." - ) - options = default_options + [ - option("pre-download", description="Pre-download the game if available."), - ] - - def handle(self): - callback(command=self) - pre_download = self.option("pre-download") - progress = utils.ProgressIndicator(self) - progress.start("Fetching install package information... ") - try: - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Fetching failed with following error: {e} \n{traceback.format_exc()}" - ) - return - progress.finish( - "Installation information fetched successfully." - ) - if not self.confirm("Do you want to download the game?"): - self.line("Download aborted.") - return - self.line("Downloading install package...") - first_pkg_out_path = None - for game_pkg in game_info.major.game_pkgs: - out_path = State.game.cache.joinpath(PurePath(game_pkg.url).name) - if not first_pkg_out_path: - first_pkg_out_path = out_path - try: - download_result = utils.download( - game_pkg.url, out_path, file_len=game_pkg.size - ) - except Exception as e: - self.line_error( - f"Couldn't download install package: {e}" - ) - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - - -class UpdateDownloadCommand(Command): - name = "hsr update download" - description = "Download the update for the local game if available" - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - option("pre-download", description="Pre-download the game if available."), - option( - "from-version", description="Update from a specific version", flag=False - ), - ] - - def handle(self): - callback(command=self) - auto_repair = self.option("auto-repair") - pre_download = self.option("pre-download") - from_version = self.option("from-version") - if auto_repair: - self.line("Auto-repair is enabled.") - if from_version: - self.line(f"Updating from version: {from_version}") - State.game.version_override = from_version - progress = utils.ProgressIndicator(self) - progress.start("Checking for updates... ") - try: - update_diff = State.game.get_update(pre_download=pre_download) - game_info = State.game.get_remote_game(pre_download=pre_download) - except Exception as e: - progress.finish( - f"Update checking failed with following error: {e} \n{traceback.format_exc()}" - ) - return - if update_diff is None or isinstance(game_info.major, str | None): - progress.finish("Game is already updated.") - return - progress.finish("Update available.") - self.line( - f"The current version is: {State.game.get_version_str()}" - ) - self.line( - f"The latest version is: {game_info.major.version}" - ) - if not self.confirm("Do you want to download the update?"): - self.line("Download aborted.") - return - self.line("Downloading update package...") - update_game_url = update_diff.game_pkgs[0].url - out_path = State.game.cache.joinpath(PurePath(update_game_url).name) - try: - download_result = utils.download( - update_game_url, out_path, file_len=update_diff.game_pkgs[0].size - ) - except Exception as e: - self.line_error( - f"Couldn't download update: {e} \n{traceback.format_exc()}" - ) - return - - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - # Get installed voicepacks - installed_voicepacks = State.game.get_installed_voicepacks() - # Voicepack update - for remote_voicepack in update_diff.audio_pkgs: - if remote_voicepack.language not in installed_voicepacks: - continue - # Voicepack is installed, update it - archive_file = State.game.cache.joinpath( - PurePath(remote_voicepack.url).name - ) - self.line( - f"Downloading update package for voicepack language '{remote_voicepack.language.name}'..." - ) - try: - download_result = utils.download( - remote_voicepack.url, archive_file, file_len=remote_voicepack.size - ) - except Exception as e: - self.line_error(f"Couldn't download update: {e}") - return - if not download_result: - self.line_error("Download failed.") - return - self.line("Download completed.") - - -class ApplyInstallArchive(Command): - name = "hsr install apply-archive" - description = "Applies the install archive" - arguments = [argument("path", description="Path to the install archive")] - options = default_options - - def handle(self): - callback(command=self) - install_archive = self.argument("path") - progress = utils.ProgressIndicator(self) - progress.start("Applying install package...") - try: - State.game.install_archive(install_archive) - except Exception as e: - progress.finish( - f"Couldn't apply package: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Package applied.") - set_version_config(self=self) - - -class ApplyUpdateArchive(Command): - name = "hsr update apply-archive" - description = "Applies the update archive to the local game" - arguments = [argument("path", description="Path to the update archive")] - options = default_options + [ - option( - "auto-repair", "R", description="Automatically repair the game if needed." - ), - ] - - def handle(self): - callback(command=self) - update_archive = self.argument("path") - auto_repair = self.option("auto-repair") - progress = utils.ProgressIndicator(self) - progress.start("Applying update package...") - try: - State.game.apply_update_archive(update_archive, auto_repair=auto_repair) - except Exception as e: - progress.finish( - f"Couldn't apply update: {e} \n{traceback.format_exc()}" - ) - return - progress.finish("Update applied.") - set_version_config() - - -commands = [ - ApplyInstallArchive, - ApplyUpdateArchive, - GetVersionCommand, - InstallCommand, - InstallDownloadCommand, - PatchCommand, - PatchInstallCommand, - PatchTelemetryCommand, - PatchTypeCommand, - RepairCommand, - UpdatePatchCommand, - UpdateCommand, - UpdateDownloadCommand, - VoicepackInstall, - VoicepackList, - VoicepackListInstalled, - VoicepackUpdate, -]