diff --git a/vollerei/cli/__init__.py b/vollerei/cli/__init__.py index bce6a16..bf9c2c5 100644 --- a/vollerei/cli/__init__.py +++ b/vollerei/cli/__init__.py @@ -1,8 +1,8 @@ from cleo.application import Application -from vollerei.cli import hsr +from vollerei.cli import hsr, genshin application = Application() -for command in hsr.commands: +for command in hsr.commands + genshin.commands: application.add(command()) diff --git a/vollerei/cli/genshin.py b/vollerei/cli/genshin.py new file mode 100644 index 0000000..239a583 --- /dev/null +++ b/vollerei/cli/genshin.py @@ -0,0 +1,748 @@ +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("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") + 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 = "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 = "genshin 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 = "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 index d4609da..e29e615 100644 --- a/vollerei/cli/hsr.py +++ b/vollerei/cli/hsr.py @@ -450,7 +450,7 @@ class PatchInstallCommand(Command): return try: patcher.block_telemetry(telemetry_list=telemetry_list) - except Exception as e: + except Exception: self.line_error( f"Couldn't block telemetry hosts: {traceback.format_exc()}" ) @@ -509,7 +509,7 @@ class PatchTelemetryCommand(Command): return try: patcher.block_telemetry(telemetry_list=telemetry_list) - except Exception as e: + except Exception: self.line_error( f"Couldn't block telemetry hosts: {traceback.format_exc()}" ) diff --git a/vollerei/common/functions.py b/vollerei/common/functions.py index 39ce34f..a1edc3b 100644 --- a/vollerei/common/functions.py +++ b/vollerei/common/functions.py @@ -5,7 +5,7 @@ import py7zr from io import IOBase from os import PathLike from pathlib import Path -from shutil import move, copyfile +from shutil import move from vollerei.abc.launcher.game import GameABC from vollerei.common.api import resource from vollerei.exceptions.game import (