fix: disable repair command

The new HoYoPlay launcher doesn't have the ability to get scattered files from the remote anymore, "repairing" the game will just download the whole game again.

.github/workflows/docs.yml

@@ -0,0 +1,52 @@
+name: Build & Deploy
+
+on: 
+  push:
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+  # Build job
+  build:
+    # Specify runner +  build & upload the static files as an artifact
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        id: checkout-code
+        uses: actions/checkout@v3
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+          cache: 'pip' # caching pip dependencies
+      - name: Install dependencies
+        id: install
+        run: |
+          pipx install poetry --python $(which python)
+          poetry install --only docs
+      - name: Build docs
+        id: build
+        run: |
+          poetry run mkdocs build
+
+      - name: Upload static files as artifact
+        id: deployment
+        uses: actions/upload-pages-artifact@v3 # or specific "vX.X.X" version tag for this action
+        with:
+          path: site/
+
+  # Deployment job
+  deploy:
+    # Grant GITHUB_TOKEN the permissions required to make a Pages deployment
+    permissions:
+      pages: write      # to deploy to Pages
+      id-token: write   # to verify the deployment originates from an appropriate source
+    environment:
+      name: github-pages
+      url: ${{ steps.deployment.outputs.page_url }}
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - name: Deploy to GitHub Pages
+        id: deployment
+        uses: actions/deploy-pages@v4

README.md

@@ -4,6 +4,17 @@
 
 An open-source launcher for anime games
 
+## Documentation
+
+https://tretrauit.me/vollerei/
+
+## Installation
+
+Assumming you have `pipx` installed, execute the following command:
+```bash
+pipx install git+https://git.tretrauit.me/tretrauit/vollerei --preinstall tqdm
+```
+
 ## Why?
 
 I've done [worthless-launcher](https://tretrauit.gitlab.io/worthless-launcher) for an open-world anime game, 
@@ -11,9 +22,33 @@ since I want to support other anime games and the launcher code is very messy, t
 
 ## Features
 
-+ Nothing, I haven't written any code yet.
+### Turn-based game / Open-world game / Proxy game
+- [x] Cross-platform support
+    > Tested on Windows and Linux myself, although should work on most major OSes where `HDiffPatch` is supported.
+- [x] Does *not* require administrator/root privileges
+    > Though if issues occur during installation/upgrading process, you can always try running the program with elevated privileges to fix them.
+- [x] Download the game update (including pre-downloads if available)
+- [x] Get the game version
+- [x] Get installed voicepacks
+- [x] Installation
+- [x] Patch the game for unsupported platforms (with telemetry checking)
+    > Not applicable for the "Open-world game", since patching is unneeded to play the game.
+- [x] Repair the game (Smarter than the official launcher!)
+- [x] Update the game
+- [x] Update voicepacks 
+- [ ] Uninstall the game (Just remove the game directory lol)
+- [x] Voicepacks installation
+#### Advanced features
+- [x] Apply the update archives
+- [x] Download the update archives
+- [x] Easy to use API
 
-### Notes
+### Other games (Dreamseeker game)
+I haven't developed for them yet, but since most of the code is shared I'll do that when I have the motivation to do so.
+
+~~Help me get motivated by going to https://paypal.me/tretrauit and send me a coffee lol~~
+
+## Notes
 
 This launcher tries to mimic the official launcher behaviour as much as possible but if a ban appears, I will
 not be responsible for it. (*Turn-based game* have a ban wave already, see AAGL discord for more info)
@@ -23,7 +58,7 @@ not be responsible for it. (*Turn-based game* have a ban wave already, see AAGL
 This launcher focuses on the API and CLI, for GUI-based launcher you may want to check out:
 
 + [An Anime Game Launcher](https://aagl.launcher.moe/) - That famous launcher for an open-world anime game.
-+ [Yaagl]
++ [Yaagl](https://github.com/3Shain/yet-another-anime-game-launcher) - All in one launcher for MacOS.
 + [Honkers launcher](https://github.com/an-anime-team/honkers-launcher) - Another launcher for an anime game.
 + [Honkers Railway](https://github.com/an-anime-team/the-honkers-railway-launcher) - A launcher for a turn-based anime game.
 

docs/abc/game.md

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

docs/common/functions.md

@@ -0,0 +1,9 @@
+# Function for all games
+
+Since you should use the specific implementation for the target game instead, these functions may not be correctly documented.
+
+::: vollerei.common.functions
+    handler: python
+    options:
+      show_root_heading: true
+      show_source: true

docs/game/launcher/game.md

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

docs/genshin/game.md

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

docs/hsr/game.md

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

docs/hsr/patcher.md

@@ -0,0 +1,7 @@
+# Patcher
+
+::: vollerei.hsr.Patcher
+    handler: python
+    options:
+      show_root_heading: true
+      show_source: true

docs/hsr/patchtype.md

@@ -0,0 +1,8 @@
+# PatchType
+
+::: vollerei.hsr.PatchType
+    handler: python
+    options:
+        show_root_heading: true
+        show_source: true
+        

docs/index.md

@@ -0,0 +1,17 @@
+# Welcome to MkDocs
+
+For full documentation visit [mkdocs.org](https://www.mkdocs.org).
+
+## Commands
+
+* `mkdocs new [dir-name]` - Create a new project.
+* `mkdocs serve` - Start the live-reloading docs server.
+* `mkdocs build` - Build the documentation site.
+* `mkdocs -h` - Print help message and exit.
+
+## Project layout
+
+    mkdocs.yml    # The configuration file.
+    docs/
+        index.md  # The documentation homepage.
+        ...       # Other markdown pages, images and other files.

docs/zzz/game.md

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

mkdocs.yml

@@ -0,0 +1,38 @@
+site_name: Vollerei
+site_url: https://tretrauit.me/vollerei/
+site_author: tretrauit
+site_description: >-
+  Documentation for the Vollerei project, a Python library for managing and
+  interacting with HoYoVerse games.
+
+repo_name: teppyboy/vollerei
+repo_url: https://github.com/teppyboy/vollerei
+
+# Copyright
+copyright: Copyright © 2023 - 2024 tretrauit
+
+theme:
+  name: material
+  palette:
+    - media: "(prefers-color-scheme)"
+      toggle:
+        icon: material/link
+        name: Switch to light mode
+    - media: "(prefers-color-scheme: light)"
+      scheme: default
+      primary: indigo
+      accent: indigo
+      toggle:
+        icon: material/toggle-switch
+        name: Switch to dark mode
+    - media: "(prefers-color-scheme: dark)"
+      scheme: slate
+      primary: black
+      accent: indigo
+      toggle:
+        icon: material/toggle-switch-off
+        name: Switch to system preference
+
+plugins:
+- search
+- mkdocstrings

pyproject.toml

@@ -9,11 +9,33 @@ readme = "README.md"
 [tool.poetry.dependencies]
 python = "^3.11"
 platformdirs = "^3.5.1"
+requests = "^2.31.0"
+cleo = "^2.1.0"
+packaging = "^23.2"
+py7zr = "^0.22.0"
+multivolumefile = "^0.2.3"
+
+[tool.poetry.group.cli]
+optional = true
+
+[tool.poetry.group.cli.dependencies]
+tqdm = "^4.65.0"
+
+[tool.poetry.group.dev]
+optional = true
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.3.1"
 pre-commit = "^3.3.3"
 
+
+[tool.poetry.group.docs.dependencies]
+mkdocstrings-python = "^1.12.2"
+mkdocs-material = "^9.5.49"
+
+[tool.poetry.scripts]
+vollerei = 'vollerei.cli:run'
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

vollerei/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

vollerei/__main__.py

@@ -0,0 +1,4 @@
+from vollerei.cli import run
+
+
+run()

vollerei/abc/launcher/game.py

@@ -1,5 +1,8 @@
 from abc import ABC, abstractmethod
 from os import PathLike
+from pathlib import Path
+from typing import Any
+from vollerei.common.api import resource
 
 
 class GameABC(ABC):
@@ -7,17 +10,26 @@ class GameABC(ABC):
     Manages the game installation
     """
 
+    path: Path
+    cache: Path
+    version_override: tuple[int, int, int] | None
+    channel_override: Any
+
     def __init__(self, path: PathLike = None):
         pass
 
-    @abstractmethod
+    def data_folder(self) -> Path:
+        """
+        Get the game data folder
+        """
+        pass
+
     def is_installed(self) -> bool:
         """
         Check if the game is installed
         """
         pass
 
-    @abstractmethod
     def install_game(self, game_path: PathLike = None):
         """
         Install the game
@@ -27,13 +39,9 @@ class GameABC(ABC):
 
         Args:
             game_path (PathLike, optional): Path to install the game to.
-
-        Returns:
-            None
         """
         pass
 
-    @abstractmethod
     def install_game_from_archive(
         self, archive: PathLike, game_path: PathLike = None
     ) -> None:
@@ -48,7 +56,6 @@ class GameABC(ABC):
             game_path (PathLike, optional): Path to install the game to.
         """
 
-    @abstractmethod
     def install_update_from_archive(
         self, archive: PathLike, game_path: PathLike = None
     ) -> None:
@@ -61,7 +68,44 @@ class GameABC(ABC):
         """
         pass
 
-    @abstractmethod
+    def repair_file(
+        self, file: PathLike, pre_download: bool = False, game_info=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.
+        """
+        pass
+
+    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.
+
+        This method is not multi-threaded, so it may take a while to repair
+        multiple files.
+
+        Args:
+            files (PathLike): The files to repair.
+            pre_download (bool): Whether to get the pre-download version.
+                Defaults to False.
+        """
+        pass
+
     def get_version(self) -> tuple[int, int, int]:
         """
         Get the game version
@@ -70,23 +114,43 @@ class GameABC(ABC):
         """
         pass
 
-    @abstractmethod
+    def get_version_config(self) -> tuple[int, int, int]:
-    def get_update(self):
+        """
+        Gets the current installed game version from config.ini.
+
+        Using this is not recommended, as only official launcher creates
+        and uses this file, instead you should use `get_version()`.
+
+        This returns (0, 0, 0) if the version could not be found.
+
+        Returns:
+            tuple[int, int, int]: Game version.
+        """
+        pass
+
+    def get_update(self) -> resource.Patch | None:
         """
         Get the game update
         """
         pass
 
-    @abstractmethod
-    def get_voiceover_update(self, language: str):
-        """
-        Get the voiceover update
-        """
-        pass
-
-    @abstractmethod
     def get_channel(self):
         """
         Get the game channel
         """
         pass
+
+    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 `Game` object that contains the game information.
+        """
+        pass

vollerei/abc/patcher.py

@@ -9,8 +9,36 @@ class PatcherABC(ABC):
 
     @abstractmethod
     def patch_game(self, game: GameABC):
+        """
+        Patch the game
+
+        If the game is not installed then it'll raise `GameNotInstalledError`, if the
+        game version is not supported then it'll raise `VersionNotSupportedError` and
+        if the patching fails then it'll raise `PatchingFailedError`.
+
+        Args:
+            game (Game): Game instance to patch
+        """
         pass
 
     @abstractmethod
     def unpatch_game(self, game: GameABC):
+        """
+        Unpatch the game
+
+        This method unpatch the game by restoring backups and removing the patch files.
+        It'll fail if you removed the backup files, in that case you'll have to repair
+        the game.
+
+        If the game is not installed then it'll raise `GameNotInstalledError` and if the
+        unpatching fails then it'll raise `UnpatchingFailedError`.
+        """
+        pass
+
+    @abstractmethod
+    def check_telemetry(self):
+        pass
+
+    @abstractmethod
+    def block_telemetry(self):
         pass

vollerei/cli/__init__.py

@@ -1,6 +1,10 @@
-from typing import Any
+from cleo.application import Application
+from vollerei.cli import commands
+
+application = Application()
+for command in commands.exports:
+    application.add(command)
 
 
-class CLI:
+def run():
-    def __init__(self):
+    application.run()
-        pass

vollerei/cli/commands.py

@@ -0,0 +1,982 @@
+import copy
+import traceback
+from cleo.commands.command import Command
+from cleo.helpers import option, argument
+from pathlib import PurePath
+from platform import system
+from vollerei.abc.launcher.game import GameABC
+from vollerei.common.api import resource
+from vollerei.common.enums import GameChannel, VoicePackLanguage
+from vollerei.cli import utils
+from vollerei.exceptions.game import GameError
+from vollerei.exceptions.patcher import PatcherError, PatchUpdateError
+from vollerei.genshin import Game as GenshinGame
+from vollerei.hsr import Game as HSRGame, Patcher as HSRPatcher
+from vollerei.hsr.patcher import PatchType as HSRPatchType
+from vollerei.zzz import Game as ZZZGame
+from vollerei import paths
+
+patcher = HSRPatcher()
+
+
+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: GameABC = 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)
+    if command.name.startswith("hsr"):
+        State.game = HSRGame(game_path, temporary_path)
+        patch_type = command.option("patch-type")
+        if patch_type is None:
+            patch_type = HSRPatchType.Jadeite
+        elif isinstance(patch_type, str):
+            patch_type = HSRPatchType[patch_type]
+        elif isinstance(patch_type, int):
+            patch_type = HSRPatchType(patch_type)
+        patcher.patch_type = patch_type
+    elif command.name.startswith("genshin"):
+        State.game = GenshinGame(game_path)
+    elif command.name.startswith("zzz"):
+        State.game = ZZZGame(game_path)
+    else:
+        raise ValueError("Invalid game type")
+    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>{question} (yes/no)</question> [<comment>{'yes' if default else 'no'}</comment>] 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"<warn>Couldn't set version config: {e}</warn>")
+        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: <comment>{State.game.get_version_str()}</comment>"
+    )
+
+
+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"<comment>{x.name}</comment>"
+            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"<comment>{x.language.name} ({x.language.value})</comment>"
+            for x in remote_game.major.audio_pkgs
+        ]
+        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"<error>Invalid language: {language}</error>")
+        if len(language_objects) == 0:
+            self.line_error(
+                "<error>No valid languages specified, you must specify a language to install</error>"
+            )
+            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"<error>Fetching failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish(
+            "<comment>Installation information fetched successfully.</comment>"
+        )
+        if not self.confirm("Do you want to install the specified voicepacks?"):
+            self.line("<error>Installation aborted.</error>")
+            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: <comment>{remote_voicepack.language.name}</comment>... "
+            )
+            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"<error>Couldn't download package: {e}</error>")
+                return
+            if not download_result:
+                self.line_error("<error>Download failed.</error>")
+                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"<error>Couldn't apply package: {e} \n{traceback.format_exc()}</error>"
+                )
+                return
+            progress.finish(
+                f"<comment>Package applied for language {remote_voicepack.language.name}.</comment>"
+            )
+        self.line(
+            f"The voicepacks have been installed to version: <comment>{State.game.get_version_str()}</comment>"
+        )
+
+
+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("<comment>Auto-repair is enabled.</comment>")
+        if from_version:
+            self.line(f"Updating from version: <comment>{from_version}</comment>")
+            State.game.version_override = from_version
+        # Get installed voicepacks
+        if len(languages) == 0:
+            self.line(
+                "<comment>No languages specified, updating all installed voicepacks...</comment>"
+            )
+        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"<comment>{str(x.name)}</comment>" 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: resource.Patch | None = 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"<error>Update checking failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        if update_diff is None:
+            progress.finish("<comment>Game is already updated.</comment>")
+            return
+        progress.finish("<comment>Update available.</comment>")
+        self.line(
+            f"The current version is: <comment>{State.game.get_version_str()}</comment>"
+        )
+        self.line(
+            f"The latest version is: <comment>{game_info.major.version}</comment>"
+        )
+        if not self.confirm("Do you want to update the game?"):
+            self.line("<error>Update aborted.</error>")
+            return
+        # 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"<error>Couldn't download update: {e}</error>")
+                return
+            if not download_result:
+                self.line_error("<error>Download failed.</error>")
+                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"<error>Couldn't apply update: {e} \n{traceback.format_exc()}</error>"
+                )
+                return
+            progress.finish(
+                f"<comment>Update applied for language {remote_voicepack.language.name}.</comment>"
+            )
+        State.game.version_override = game_info.major.version
+        set_version_config(self=self)
+        State.game.version_override = None
+
+
+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: <comment>{patcher.patch_type.name}</comment>")
+
+
+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"<error>Patch update failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+        else:
+            progress.finish("<comment>Patch updated!</comment>")
+
+
+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"<error>Patch installation failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Patch installed!</comment>")
+        print()
+        exe_path = jadeite_dir.joinpath("jadeite.exe")
+        self.line(f"Jadeite executable is located at: <question>{exe_path}</question>")
+        self.line(
+            "You need to <warn>run the game using Jadeite</warn> to use the patch."
+        )
+        self.line(f'E.g: <question>{exe_path} "{State.game.path}"</question>')
+        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 <warn>use test accounts</warn>, as there is an <warn>extremely high risk of getting banned.</warn>"
+        )
+
+    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"<error>Patch installation failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Patch installed!</comment>")
+        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 <comment>officialy supported</comment> by the game, so no patching is needed."
+            )
+        self.line(
+            "By patching the game, <warn>you are violating the ToS of the game.</warn>"
+        )
+        if not self.confirm("Do you want to patch the game?"):
+            self.line("<error>Patching aborted.</error>")
+            return
+        progress = utils.ProgressIndicator(self)
+        progress.start("Checking telemetry hosts... ")
+        telemetry_list = patcher.check_telemetry()
+        if telemetry_list:
+            progress.finish("<warn>Telemetry hosts were found.</warn>")
+            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 <comment>block these hosts.</comment>"
+            )
+            if not self.confirm("Do you want to block them?"):
+                self.line("<error>Patching aborted.</error>")
+                self.line(
+                    "<error>Please block these hosts manually then try again.</error>"
+                )
+                return
+            try:
+                patcher.block_telemetry(telemetry_list=telemetry_list)
+            except Exception:
+                self.line_error(
+                    f"<error>Couldn't block telemetry hosts: {traceback.format_exc()}</error>"
+                )
+                # There's a good reason for this.
+                if system() != "Windows":
+                    self.line(
+                        "<error>Cannot continue, please block them manually then try again.</error>"
+                    )
+                    return
+                self.line("<warn>Continuing anyway...</warn>")
+        else:
+            progress.finish("<comment>No telemetry hosts found.</comment>")
+        progress = utils.ProgressIndicator(self)
+        progress.start("Updating patch... ")
+        try:
+            patcher.update_patch()
+        except PatchUpdateError as e:
+            progress.finish(
+                f"<error>Patch update failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+        else:
+            progress.finish("<comment>Patch updated.</comment>")
+        match patcher.patch_type:
+            case HSRPatchType.Jadeite:
+                self.jadeite()
+            case HSRPatchType.Astra:
+                self.astra()
+
+
+PatchCommand = copy.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("<warn>Telemetry hosts were found.</warn>")
+            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 <comment>block these hosts.</comment>"
+            )
+            if not self.confirm("Do you want to block them?"):
+                self.line("<error>Blocking aborted.</error>")
+                return
+            try:
+                patcher.block_telemetry(telemetry_list=telemetry_list)
+            except Exception:
+                self.line_error(
+                    f"<error>Couldn't block telemetry hosts: {traceback.format_exc()}</error>"
+                )
+        else:
+            progress.finish("<comment>No telemetry hosts found.</comment>")
+
+
+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"<comment>Version:</comment> {'.'.join(str(x) for x in State.game.get_version())}"
+            )
+        except GameError as e:
+            self.line_error(f"<error>Couldn't get game version: {e}</error>")
+
+
+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"<error>Fetching failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish(
+            "<comment>Installation information fetched successfully.</comment>"
+        )
+        if not self.confirm("Do you want to install the game?"):
+            self.line("<error>Installation aborted.</error>")
+            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"<error>Couldn't download install package: {e}</error>"
+                )
+                return
+            if not download_result:
+                self.line_error("<error>Download failed.</error>")
+                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"<error>Couldn't install package: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Package applied for the base game.</comment>")
+        self.line("Setting version config... ")
+        State.game.version_override = game_info.major.version
+        set_version_config(self=self)
+        State.game.version_override = None
+        self.line(
+            f"The game has been installed to version: <comment>{State.game.get_version_str()}</comment>"
+        )
+
+
+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("<comment>Auto-repair is enabled.</comment>")
+        if from_version:
+            self.line(f"Updating from version: <comment>{from_version}</comment>")
+            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"<error>Update checking failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        if update_diff is None or isinstance(game_info.major, str | None):
+            progress.finish("<comment>Game is already updated.</comment>")
+            return
+        progress.finish("<comment>Update available.</comment>")
+        self.line(
+            f"The current version is: <comment>{State.game.get_version_str()}</comment>"
+        )
+        self.line(
+            f"The latest version is: <comment>{game_info.major.version}</comment>"
+        )
+        if not self.confirm("Do you want to update the game?"):
+            self.line("<error>Update aborted.</error>")
+            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"<error>Couldn't download update: {e}</error>")
+            return
+
+        if not download_result:
+            self.line_error("<error>Download failed.</error>")
+            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"<error>Couldn't apply update: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Update applied for base game.</comment>")
+        # 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"<error>Couldn't download update: {e}</error>")
+                return
+            if not download_result:
+                self.line_error("<error>Download failed.</error>")
+                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"<error>Couldn't apply update: {e} \n{traceback.format_exc()}</error>"
+                )
+                return
+            progress.finish(
+                f"<comment>Update applied for language {remote_voicepack.language.name}.</comment>"
+            )
+        self.line("Setting version config... ")
+        State.game.version_override = game_info.major.version
+        set_version_config(self=self)
+        State.game.version_override = None
+        self.line(
+            f"The game has been updated to version: <comment>{State.game.get_version_str()}</comment>"
+        )
+
+
+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("<error>Repairation aborted.</error>")
+            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"<error>Repairation failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Repairation completed.</comment>")
+
+
+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"<error>Fetching failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish(
+            "<comment>Installation information fetched successfully.</comment>"
+        )
+        if not self.confirm("Do you want to download the game?"):
+            self.line("<error>Download aborted.</error>")
+            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"<error>Couldn't download install package: {e}</error>"
+                )
+                return
+            if not download_result:
+                self.line_error("<error>Download failed.</error>")
+                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("<comment>Auto-repair is enabled.</comment>")
+        if from_version:
+            self.line(f"Updating from version: <comment>{from_version}</comment>")
+            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"<error>Update checking failed with following error: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        if update_diff is None or isinstance(game_info.major, str | None):
+            progress.finish("<comment>Game is already updated.</comment>")
+            return
+        progress.finish("<comment>Update available.</comment>")
+        self.line(
+            f"The current version is: <comment>{State.game.get_version_str()}</comment>"
+        )
+        self.line(
+            f"The latest version is: <comment>{game_info.major.version}</comment>"
+        )
+        if not self.confirm("Do you want to download the update?"):
+            self.line("<error>Download aborted.</error>")
+            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"<error>Couldn't download update: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+
+        if not download_result:
+            self.line_error("<error>Download failed.</error>")
+            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"<error>Couldn't download update: {e}</error>")
+                return
+            if not download_result:
+                self.line_error("<error>Download failed.</error>")
+                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"<error>Couldn't apply package: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Package applied.</comment>")
+        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"<error>Couldn't apply update: {e} \n{traceback.format_exc()}</error>"
+            )
+            return
+        progress.finish("<comment>Update applied.</comment>")
+        set_version_config(self=self)
+
+
+# This is the list for HSR commands, we'll add Genshin commands later
+classes = [
+    ApplyInstallArchive,
+    ApplyUpdateArchive,
+    GetVersionCommand,
+    InstallCommand,
+    InstallDownloadCommand,
+    PatchCommand,
+    PatchInstallCommand,
+    PatchTelemetryCommand,
+    PatchTypeCommand,
+    # RepairCommand,
+    UpdatePatchCommand,
+    UpdateCommand,
+    UpdateDownloadCommand,
+    VoicepackInstall,
+    VoicepackList,
+    VoicepackListInstalled,
+    VoicepackUpdate,
+]
+exports = []
+for command in classes:
+    init_command = command()
+    exports.append(init_command)
+    if "patch" in command.name:
+        continue
+    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)

vollerei/cli/utils.py

@@ -0,0 +1,101 @@
+import requests
+from cleo.commands.command import Command
+from pathlib import Path
+from threading import Thread
+from time import sleep
+from tqdm import tqdm
+
+
+no_confirm = False
+silent_message = False
+
+
+def args_to_kwargs(args: list):
+    """
+    Convert a list of arguments to a dict of keyword arguments.
+    """
+    kwargs = {}
+    cur_key = None
+    for arg in args:
+        if "--" == arg[:2]:
+            arg_key = arg[2:].replace("-", "_")
+            kwargs[arg_key] = True
+            cur_key = arg_key
+        elif cur_key:
+            kwargs[cur_key] = arg
+    return kwargs
+
+
+class ProgressIndicator:
+    def auto_advance(self):
+        """
+        Automatically advance the progress indicator.
+        """
+        while self.progress._started:
+            self.progress.advance()
+            sleep(self.progress._interval / 1000)
+
+    def __init__(
+        self, command: Command, interval: int = None, values: list[str] = None
+    ):
+        self.command = command
+        if not interval:
+            interval = 100
+        if not values:
+            values = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
+        self.progress = self.command.progress_indicator(
+            interval=interval, values=values
+        )
+        self.thread = Thread(target=self.auto_advance)
+        self.thread.daemon = True
+
+    def start(self, message: str):
+        """
+        Start the progress indicator.
+        """
+        self.progress.start(message)
+        self.thread.start()
+
+    def finish(self, message: str, reset_indicator=False):
+        """
+        Finish the progress indicator.
+        """
+        self.progress.finish(message=message, reset_indicator=reset_indicator)
+
+
+def download(url, out: Path, file_len: int = None, overwrite: bool = False) -> bool:
+    if overwrite:
+        out.unlink(missing_ok=True)
+    headers = {}
+    if out.exists():
+        cur_len = (out.stat()).st_size
+        headers |= {"Range": f"bytes={cur_len}-{file_len if file_len else ''}"}
+    else:
+        out.touch()
+    # Streaming, so we can iterate over the response.
+    response = requests.get(url=url, headers=headers, stream=True)
+    if response.status_code == 416:
+        return True
+    response.raise_for_status()
+    # Sizes in bytes.
+    total_size = int(response.headers.get("content-length", 0))
+    block_size = 32768
+
+    with tqdm(total=total_size, unit="KB", unit_scale=True) as progress_bar:
+        with out.open("ab") as file:
+            for data in response.iter_content(block_size):
+                progress_bar.update(len(data))
+                file.write(data)
+
+    if total_size != 0 and progress_bar.n != total_size:
+        return False
+    return True
+
+
+def msg(*args, **kwargs):
+    """
+    Print but silentable
+    """
+    if silent_message:
+        return
+    print(*args, **kwargs)

vollerei/common/__init__.py

@@ -0,0 +1,4 @@
+from vollerei.common.configfile import ConfigFile
+from vollerei.common.telemetry import block_telemetry, check_telemetry
+
+__all__ = ["ConfigFile", "block_telemetry", "check_telemetry"]

vollerei/common/api/__init__.py

@@ -0,0 +1,35 @@
+import requests
+from vollerei.common.api import resource
+from vollerei.common.enums import GameChannel
+from vollerei.constants import LAUNCHER_API
+
+
+__all__ = ["GamePackage"]
+
+
+def get_game_packages(
+    channel: GameChannel = GameChannel.Overseas,
+) -> list[resource.GameInfo]:
+    """
+    Get game packages information from the launcher API.
+
+    Default channel is overseas.
+
+    Args:
+        channel: Game channel to get the resource information from.
+
+    Returns:
+        Resource: Game resource information.
+    """
+    resource_path: dict = None
+    match channel:
+        case GameChannel.Overseas:
+            resource_path = LAUNCHER_API.OS
+        case GameChannel.China:
+            resource_path = LAUNCHER_API.CN
+    return resource.from_dict(
+        requests.get(
+            resource_path["url"] + LAUNCHER_API.RESOURCE_PATH,
+            params=resource_path["params"],
+        ).json()["data"]
+    )

vollerei/common/api/resource.py

@@ -0,0 +1,156 @@
+from vollerei.common.enums import VoicePackLanguage
+from typing import Union
+
+
+class Game:
+    def __init__(self, id: str, biz: str):
+        self.id = id
+        self.biz = biz
+
+    @staticmethod
+    def from_dict(data: dict) -> "Game":
+        return Game(id=data["id"], biz=data["biz"])
+
+
+class GamePackage:
+    def __init__(self, url: str, md5: str, size: int, decompressed_size: int):
+        self.url = url
+        self.md5 = md5
+        self.size = size
+        self.decompressed_size = decompressed_size
+
+    @staticmethod
+    def from_dict(data: list[dict]) -> list["GamePackage"]:
+        game_pkgs = []
+        for pkg in data:
+            game_pkgs.append(
+                GamePackage(
+                    url=pkg["url"],
+                    md5=pkg["md5"],
+                    size=int(pkg["size"]),
+                    decompressed_size=int(pkg["decompressed_size"]),
+                )
+            )
+        return game_pkgs
+
+
+class AudioPackage:
+    def __init__(
+        self,
+        language: VoicePackLanguage,
+        url: str,
+        md5: str,
+        size: int,
+        decompressed_size: int,
+    ):
+        self.language = language
+        self.url = url
+        self.md5 = md5
+        self.size = size
+        self.decompressed_size = decompressed_size
+
+    @staticmethod
+    def from_dict(data: list[dict]) -> "AudioPackage":
+        audio_pkgs = []
+        for pkg in data:
+            audio_pkgs.append(
+                AudioPackage(
+                    language=VoicePackLanguage.from_remote_str(pkg["language"]),
+                    url=pkg["url"],
+                    md5=pkg["md5"],
+                    size=int(pkg["size"]),
+                    decompressed_size=int(pkg["decompressed_size"]),
+                )
+            )
+        return audio_pkgs
+
+
+class Major:
+    def __init__(
+        self,
+        version: str,
+        game_pkgs: list[GamePackage],
+        audio_pkgs: list[AudioPackage],
+        res_list_url: str,
+    ):
+        self.version = version
+        self.game_pkgs = game_pkgs
+        self.audio_pkgs = audio_pkgs
+        self.res_list_url = res_list_url
+
+    @staticmethod
+    def from_dict(data: dict) -> "Major":
+        return Major(
+            version=data["version"],
+            game_pkgs=GamePackage.from_dict(data["game_pkgs"]),
+            audio_pkgs=AudioPackage.from_dict(data["audio_pkgs"]),
+            res_list_url=data["res_list_url"],
+        )
+
+
+# Currently patch has the same fields as major
+Patch = Major
+
+
+class Main:
+    def __init__(self, major: Major, patches: list[Patch]):
+        self.major = major
+        self.patches = patches
+
+    @staticmethod
+    def from_dict(data: dict) -> "Main":
+        return Main(
+            major=Major.from_dict(data["major"]),
+            patches=[Patch.from_dict(x) for x in data["patches"]],
+        )
+
+
+class PreDownload:
+    def __init__(self, major: Major | str | None, patches: list[Patch]):
+        self.major = major
+        self.patches = patches
+
+    # Union to fix the typing issue.
+    @staticmethod
+    def from_dict(data: dict | None) -> Union["PreDownload", None]:
+        # pre_download can be null in the server for certain games
+        # e.g. HI3:
+        # "pre_download": null
+        # while in GI it is the following:
+        # "pre_download": {
+        #   "major": null,
+        #   "patches": []
+        # }
+        if data is None:
+            return None
+        return PreDownload(
+            major=(
+                data["major"]
+                if isinstance(data["major"], str | None)
+                else Major.from_dict(data["major"])
+            ),
+            patches=[Patch.from_dict(x) for x in data["patches"]],
+        )
+
+
+# Why miHoYo uses the same name "game_packages" for this big field and smol field
+class GameInfo:
+    def __init__(self, game: Game, main: Main, pre_download: PreDownload):
+        self.game = game
+        self.main = main
+        self.pre_download = pre_download
+
+    @staticmethod
+    def from_dict(data: dict) -> "GameInfo":
+        return GameInfo(
+            game=Game.from_dict(data["game"]),
+            main=Main.from_dict(data["main"]),
+            pre_download=PreDownload.from_dict(data["pre_download"]),
+        )
+
+
+def from_dict(data: dict) -> list[GameInfo]:
+    game_pkgs = []
+    for pkg in data["game_packages"]:
+        game_pkgs.append(GameInfo.from_dict(pkg))
+    return game_pkgs

vollerei/common/configfile.py

@@ -0,0 +1,15 @@
+from configparser import ConfigParser
+from pathlib import Path
+
+
+class ConfigFile(ConfigParser):
+    path: Path
+
+    def __init__(self, path, **kwargs):
+        super().__init__(**kwargs)
+        self.path = Path(path)
+        self.read(self.path)
+
+    def save(self):
+        with self.path.open("w") as f:
+            self.write(f)

vollerei/common/enums.py

@@ -0,0 +1,59 @@
+from enum import Enum
+
+
+class GameType(Enum):
+    Genshin = 0
+    HSR = 1
+    ZZZ = 3
+    HI3 = 4
+
+
+class GameChannel(Enum):
+    Overseas = 0
+    China = 1
+
+
+class VoicePackLanguage(Enum):
+    Japanese = "ja-jp"
+    Chinese = "zh-cn"
+    Taiwanese = "zh-tw"
+    Korean = "ko-kr"
+    English = "en-us"
+
+    @staticmethod
+    def from_remote_str(s: str) -> "VoicePackLanguage":
+        """
+        Converts a language string from remote server to a VoicePackLanguage enum.
+        """
+        if s == "ja-jp":
+            return VoicePackLanguage.Japanese
+        elif s == "zh-cn":
+            return VoicePackLanguage.Chinese
+        elif s == "zh-tw":
+            return VoicePackLanguage.Taiwanese
+        elif s == "ko-kr":
+            return VoicePackLanguage.Korean
+        elif s == "en-us":
+            return VoicePackLanguage.English
+        else:
+            raise ValueError(f"Invalid language string: {s}")
+
+    @staticmethod
+    def from_zzz_name(s: str) -> "VoicePackLanguage":
+        """
+        Converts a language string from ZZZ file name to a VoicePackLanguage enum.
+
+        Only English is tested for now.
+        """
+        if s == "Jp":
+            return VoicePackLanguage.Japanese
+        elif s == "Cn":
+            return VoicePackLanguage.Chinese
+        elif s == "Tw":
+            return VoicePackLanguage.Taiwanese
+        elif s == "Kr":
+            return VoicePackLanguage.Korean
+        elif s == "En":
+            return VoicePackLanguage.English
+        else:
+            raise ValueError(f"Invalid language string: {s}")

vollerei/common/functions.py

@@ -0,0 +1,403 @@
+import concurrent.futures
+import json
+import hashlib
+import multivolumefile
+import py7zr
+import zipfile
+from io import IOBase
+from os import PathLike
+from pathlib import Path
+from shutil import move
+from vollerei.abc.launcher.game import GameABC
+from vollerei.common.api import resource
+from vollerei.exceptions.game import (
+    RepairError,
+    GameNotInstalledError,
+    ScatteredFilesNotAvailableError,
+)
+from vollerei.utils import HDiffPatch, HPatchZPatchError, download
+
+
+_hdiff = HDiffPatch()
+
+
+def _extract_files(
+    archive: py7zr.SevenZipFile | zipfile.ZipFile, files, path: PathLike
+):
+    if isinstance(archive, py7zr.SevenZipFile):
+        # .7z archive
+        archive.extract(path, files)
+    else:
+        # .zip archive
+        archive.extractall(path, files)
+
+
+def _open_archive(file: Path | IOBase) -> py7zr.SevenZipFile | zipfile.ZipFile:
+    archive: py7zr.SevenZipFile | zipfile.ZipFile = None
+    try:
+        archive = py7zr.SevenZipFile(file, "r")
+    except py7zr.exceptions.Bad7zFile:
+        # Try to open it as a zip file
+        try:
+            archive = zipfile.ZipFile(file, "r")
+        except zipfile.BadZipFile:
+            raise ValueError("Archive is not a valid 7z or zip file.")
+    return archive
+
+
+def apply_update_archive(
+    game: GameABC, archive_file: Path | IOBase, auto_repair: bool = True
+) -> None:
+    """
+    Applies an update archive to the game, it can be the game update or a
+    voicepack update.
+
+    Because this function is shared for all games, you should use the game's
+    `apply_update_archive()` method instead, which additionally applies required
+    methods for that game.
+    """
+    # Most code here are copied from worthless-launcher.
+    # worthless-launcher uses asyncio for multithreading while this one uses
+    # ThreadPoolExecutor, probably better for this use case.
+
+    # We need `game` for the path and `auto_repair` for the auto repair option.
+
+    # Install HDiffPatch
+    _hdiff.hpatchz()
+
+    # Open archive
+    def reset_if_py7zr(archive):
+        if isinstance(archive, py7zr.SevenZipFile):
+            archive.reset()
+
+    archive = _open_archive(archive_file)
+
+    # Get files list (we don't want to extract all of them)
+    files = archive.namelist()
+    # Don't extract these files (they're useless and if the game isn't patched then
+    # it'll raise 31-4xxx error in Genshin)
+    for file in ["deletefiles.txt", "hdifffiles.txt", "hdiffmap.json"]:
+        try:
+            files.remove(file)
+        except ValueError:
+            pass
+    # Think for me a better name for this variable
+    txtfiles = None
+    if isinstance(archive, py7zr.SevenZipFile):
+        txtfiles = archive.read(["deletefiles.txt", "hdifffiles.txt", "hdiffmap.json"])
+        # Reset archive to extract files
+        archive.reset()
+    try:
+        # miHoYo loves CRLF
+        if txtfiles is not None:
+            deletebytes = txtfiles["deletefiles.txt"].read()
+        else:
+            deletebytes = archive.read("deletefiles.txt")
+        if deletebytes is not str:
+            # Typing
+            deletebytes: bytes
+            deletebytes = deletebytes.decode()
+        deletefiles = deletebytes.split("\r\n")
+    except (IOError, KeyError):
+        pass
+    else:
+        for file_str in deletefiles:
+            file = game.path.joinpath(file_str)
+            if file == game.path:
+                # Don't delete the game folder
+                continue
+            if not file.relative_to(game.path):
+                # File is not in the game folder
+                continue
+            # Delete the file
+            file.unlink(missing_ok=True)
+
+    # hdiffpatch implementation
+    # Read hdifffiles.txt to get the files to patch
+    # Hdifffile format is [(source file, target file)]
+    # While the patch file is named as target file + ".hdiff"
+    hdifffiles: list[tuple[str, str]] = []
+    new_hdiff_map = False
+    if txtfiles is not None:
+        old_hdiff_map = txtfiles.get("hdifffiles.txt")
+        if old_hdiff_map is not None:
+            hdiffbytes = old_hdiff_map.read()
+        else:
+            new_hdiff_map = True
+            hdiffbytes = txtfiles["hdiffmap.json"].read()
+    else:
+        # Archive file must be a zip file
+        if zipfile.Path(archive).joinpath("hdifffiles.txt").is_file():
+            hdiffbytes = archive.read("hdifffiles.txt")
+        else:
+            new_hdiff_map = True
+            hdiffbytes = archive.read("hdiffmap.json")
+    if hdiffbytes is not str:
+        # Typing
+        hdiffbytes: bytes
+        hdiffbytes = hdiffbytes.decode()
+    if new_hdiff_map:
+        mapping = json.loads(hdiffbytes)
+        for diff in mapping["diff_map"]:
+            hdifffiles.append((diff["source_file_name"], diff["target_file_name"]))
+    else:
+        for x in hdiffbytes.split("\r\n"):
+            try:
+                name = json.loads(x.strip())["remoteName"]
+                hdifffiles.append((name, name))
+            except json.JSONDecodeError:
+                pass
+
+    # Patch function
+    def patch(source_file: Path, target_file: Path, patch_file: str):
+        patch_path = game.cache.joinpath(patch_file)
+        # Spaghetti code :(, fuck my eyes.
+        bak_src_file = source_file.rename(
+            source_file.with_suffix(source_file.suffix + ".bak")
+        )
+        try:
+            _hdiff.patch_file(bak_src_file, target_file, patch_path)
+        except HPatchZPatchError:
+            if auto_repair:
+                try:
+                    # The game repairs file by downloading the latest file, in this case we want the target file
+                    # instead of source file. Honestly I haven't tested this but I hope it works.
+                    game.repair_file(target_file)
+                except Exception:
+                    # Let the game download the file.
+                    bak_src_file.rename(file.with_suffix(""))
+                else:
+                    bak_src_file.unlink()
+            else:
+                # Let the game download the file.
+                bak_src_file.rename(file.with_suffix(""))
+            return
+        else:
+            # Remove old file, since we don't need it anymore.
+            bak_src_file.unlink()
+        finally:
+            patch_path.unlink()
+
+    # Multi-threaded patching
+    patch_jobs = []
+    patch_files = []
+    for source_file, target_file in hdifffiles:
+        source_path = game.path.joinpath(source_file)
+        if not source_path.exists():
+            # Not patching since we don't have the file
+            continue
+        target_path = game.path.joinpath(target_file)
+        patch_file: str = target_file + ".hdiff"
+        # Remove hdiff files from files list to extract
+        files.remove(patch_file)
+        # Add file to extract list
+        patch_files.append(patch_file)
+        patch_jobs.append([patch, [source_path, target_path, patch_file]])
+
+    # Extract patch files to temporary dir
+    _extract_files(archive, patch_files, game.cache)
+    reset_if_py7zr(archive)  # For the next extraction
+    # Create new ThreadPoolExecutor for patching
+    patch_executor = concurrent.futures.ThreadPoolExecutor()
+    for job in patch_jobs:
+        patch_executor.submit(job[0], *job[1])
+    patch_executor.shutdown(wait=True)
+
+    # Extract files from archive after we have filtered out the patch files
+    _extract_files(archive, files, game.path)
+
+    # Close the archive
+    archive.close()
+
+
+def install_archive(game: GameABC, archive_file: Path | IOBase) -> None:
+    """
+    Applies an install archive to the game, it can be the game itself or a
+    voicepack one.
+
+    Args:
+        game (GameABC): The game to install the archive for.
+        archive_file (Path | IOBase): The archive file to install, if it's a
+            split archive then this is the first part.
+
+    Because this function is shared for all games, you should use the game's
+    `install_archive()` method instead, which additionally applies required
+    methods for that game.
+    """
+    archive: py7zr.SevenZipFile | zipfile.ZipFile = None
+    archive_path = Path(archive_file)
+    target_archive = None
+    if archive_path.suffix == ".001":
+        archive_path_merged = archive_path.with_suffix("")
+        target_archive = multivolumefile.open(archive_path_merged, mode='rb')
+        if archive_path_merged.suffix == ".zip":
+            # .zip split archive
+            archive = zipfile.ZipFile(target_archive, "r")
+        else:
+            archive = py7zr.SevenZipFile(target_archive, "r")
+    else:
+        archive = _open_archive(archive_file)
+    archive.extractall(game.path)
+    archive.close()
+    if target_archive:
+        target_archive.close()
+
+
+def _repair_file(game: GameABC, file: PathLike, game_info: resource.Main) -> None:
+    # .replace("\\", "/") is needed because Windows uses backslashes :)
+    relative_file = file.relative_to(game.path)
+    url = game_info.major.res_list_url + "/" + str(relative_file).replace("\\", "/")
+    # Backup the file
+    if file.exists():
+        backup_file = file.with_suffix(file.suffix + ".bak")
+        if backup_file.exists():
+            backup_file.unlink()
+        file.rename(backup_file)
+        dest_file = file.with_suffix("")
+    else:
+        dest_file = file
+    try:
+        # Download the file
+        temp_file = game.cache.joinpath(relative_file)
+        dest_file.parent.mkdir(parents=True, exist_ok=True)
+        print(f"Downloading repair file {url} to {temp_file}")
+        download(url, temp_file, overwrite=True, stream=True)
+        # Move the file
+        move(temp_file, dest_file)
+        print("OK")
+    except Exception as e:
+        # Restore the backup
+        print("Failed", e)
+        if file.exists():
+            file.rename(file.with_suffix(""))
+        raise e
+    # Delete the backup
+    if file.exists():
+        file.unlink(missing_ok=True)
+
+
+def repair_files(
+    game: GameABC,
+    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:
+        game (GameABC): The game to repair the files for.
+        files (PathLike): The files to repair.
+        pre_download (bool): Whether to get the pre-download version.
+            Defaults to False.
+    """
+    if not game.is_installed():
+        raise GameNotInstalledError("Game is not installed.")
+    files_path = [Path(file) for file in files]
+    for file in files_path:
+        if not file.is_relative_to(game.path):
+            raise ValueError("File is not in the game folder.")
+    if not game_info:
+        game_info = game.get_remote_game(pre_download=pre_download)
+    if game_info.latest.decompressed_path is None:
+        raise ScatteredFilesNotAvailableError("Scattered files are not available.")
+    executor = concurrent.futures.ThreadPoolExecutor()
+    for file in files_path:
+        executor.submit(_repair_file, file, game=game_info)
+        # self._repair_file(file, game=game)
+    executor.shutdown(wait=True)
+
+
+def repair_game(
+    game: GameABC,
+    pre_download: bool = False,
+) -> None:
+    """
+    Tries to repair the game by reading "pkg_version" file and downloading the
+    mismatched files from the server.
+
+    Because this function is shared for all games, you should use the game's
+    `repair_game()` method instead, which additionally applies required
+    methods for that game.
+    """
+    # Most code here are copied from worthless-launcher.
+    # worthless-launcher uses asyncio for multithreading while this one uses
+    # ThreadPoolExecutor, probably better for this use case.
+    if not game.is_installed():
+        raise GameNotInstalledError("Game is not installed.")
+    game_info = game.get_remote_game(pre_download=pre_download)
+    pkg_version_file = game.path.joinpath("pkg_version")
+    pkg_version: dict[str, dict[str, str]] = {}
+    if not pkg_version_file.is_file():
+        try:
+            game.repair_file(game.path.joinpath("pkg_version"), game_info=game_info)
+        except Exception as e:
+            raise RepairError(
+                "pkg_version file not found, most likely you need to download the full game again."
+            ) from e
+    with pkg_version_file.open("r") as f:
+        for line in f.readlines():
+            line = line.strip()
+            if not line:
+                continue
+            line_json = json.loads(line)
+            pkg_version[line_json["remoteName"]] = {
+                "md5": line_json["md5"],
+                "fileSize": line_json["fileSize"],
+            }
+    read_needed_files: list[Path] = []
+    target_files: list[Path] = []
+    repair_executor = concurrent.futures.ThreadPoolExecutor()
+    for file in game.path.rglob("*"):
+        # Ignore webCaches folder (because it's user data)
+        if file.is_dir():
+            continue
+        if "webCaches" in str(file):
+            continue
+
+        def verify(file_path: Path):
+            nonlocal target_files
+            nonlocal pkg_version
+            relative_path = file_path.relative_to(game.path)
+            relative_path_str = str(relative_path).replace("\\", "/")
+            # print(relative_path_str)
+            # Wtf mihoyo, you build this game for Windows and then use Unix path separator :moyai:
+            try:
+                target_file = pkg_version.pop(relative_path_str)
+                if target_file:
+                    with file_path.open("rb", buffering=0) as f:
+                        file_hash = hashlib.file_digest(f, "md5").hexdigest()
+                    if file_hash == target_file["md5"]:
+                        return
+                    print(
+                        f"Hash mismatch for {target_file['remoteName']} ({file_hash}; expected {target_file['md5']})"
+                    )
+                    target_files.append(file_path)
+            except KeyError:
+                # File not found in pkg_version
+                read_needed_files.append(file_path)
+
+        repair_executor.submit(verify, file)
+    repair_executor.shutdown(wait=True)
+    for file in read_needed_files:
+        try:
+            with file.open("rb", buffering=0) as f:
+                # We only need to read 4 bytes to see if the file is readable or not
+                f.read(4)
+        except Exception:
+            print(f"File '{file}' is corrupted.")
+            target_files.append(file)
+    # value not used for now
+    for key, _ in pkg_version.items():
+        target_file = game.path.joinpath(key)
+        if target_file.is_file():
+            continue
+        print(f"{key} not found.")
+        target_files.append(target_file)
+    if not target_files:
+        return
+    print("Begin repairing files...")
+    game.repair_files(target_files, game_info=game_info)

vollerei/common/telemetry.py

@@ -0,0 +1,31 @@
+import requests
+import concurrent.futures
+from vollerei.utils import write_hosts
+from vollerei.constants import TELEMETRY_HOSTS
+
+
+def _check_telemetry(host: str) -> str | None:
+    try:
+        requests.get(f"https://{host}/", timeout=15)
+    except (requests.ConnectionError, requests.Timeout, requests.HTTPError):
+        return
+    return host
+
+
+def check_telemetry() -> list[str]:
+    futures = []
+    with concurrent.futures.ThreadPoolExecutor() as executor:
+        for host in TELEMETRY_HOSTS:
+            futures.append(executor.submit(_check_telemetry, host))
+    hosts = []
+    for future in concurrent.futures.as_completed(futures):
+        host = future.result()
+        if host:
+            hosts.append(host)
+    return hosts
+
+
+def block_telemetry(telemetry_list: list[str] = None):
+    if not telemetry_list:
+        telemetry_list = check_telemetry()
+    write_hosts(telemetry_list)

vollerei/constants.py

@@ -1,5 +1,22 @@
-# Common
+class LAUNCHER_API:
-telemetry_hosts = [
+    """Launcher API constants."""
+
+    RESOURCE_PATH: str = "hyp/hyp-connect/api/getGamePackages"
+    OS: dict = {
+        "url": "https://sg-hyp-api.hoyoverse.com/",
+        "params": {
+            "launcher_id": "VYTpXlbWo8",
+        },
+    }
+    CN: dict = {
+        "url": "https://hyp-api.mihoyo.com/",
+        "params": {
+            "launcher_id": "jGHBHlcOq1",
+        },
+    }
+
+
+TELEMETRY_HOSTS = [
     # Global
     "log-upload-os.hoyoverse.com",
     "sg-public-data-api.hoyoverse.com",
@@ -8,7 +25,4 @@ telemetry_hosts = [
     "log-upload.mihoyo.com",
     "public-data-api.mihoyo.com",
 ]
-
+HDIFFPATCH_GIT_URL = "https://github.com/sisong/HDiffPatch"
-# HSR
-astra_repo = "https://notabug.org/mkrsym1/astra"
-jadeite_repo = "https://codeberg.org/mkrsym1/jadeite/"

vollerei/exceptions/game.py

@@ -8,6 +8,42 @@ class GameError(VollereiError):
 
 
 class GameNotInstalledError(GameError):
-    """Exception raised when the game is not installed."""
+    """Game is not installed."""
+
+    pass
+
+
+class GameAlreadyUpdatedError(GameError):
+    """Game is already updated."""
+
+    pass
+
+
+class GameAlreadyInstalledError(GameError):
+    """Game is already installed."""
+
+    pass
+
+
+class RepairError(GameError):
+    """Error occurred while repairing the game."""
+
+    pass
+
+
+class ScatteredFilesNotAvailableError(RepairError):
+    """Scattered files are not available."""
+
+    pass
+
+
+class GameNotUpdatedError(GameError):
+    """Game is not updated."""
+
+    pass
+
+
+class PreDownloadNotAvailable(GameError):
+    """Pre-download version is not available."""
 
     pass

vollerei/exceptions/patcher.py

@@ -11,3 +11,21 @@ class VersionNotSupportedError(PatcherError):
     """Exception raised when the game version is not supported."""
 
     pass
+
+
+class PatchingFailedError(PatcherError):
+    """Exception raised when the patching failed."""
+
+    pass
+
+
+class PatchUpdateError(PatcherError):
+    """Exception raised when the patch update failed."""
+
+    pass
+
+
+class UnpatchingFailedError(PatcherError):
+    """Exception raised when the unpatching failed."""
+
+    pass

vollerei/game/genshin/functions.py

@@ -0,0 +1,86 @@
+from vollerei.common.enums import GameChannel
+from vollerei.abc.launcher.game import GameABC
+from vollerei.exceptions.game import GameNotInstalledError
+
+
+def get_channel(game: GameABC) -> GameChannel:
+    """
+    Gets the current game channel.
+
+    Returns:
+        GameChannel: The current game channel.
+    """
+    if game.channel_override:
+        return game.channel_override
+    if not game.is_installed():
+        raise GameNotInstalledError("Game path is not set.")
+    if game.path.joinpath("YuanShen.exe").is_file():
+        return GameChannel.China
+    return GameChannel.Overseas
+
+
+def get_version(game: GameABC) -> tuple[int, int, int]:
+    """
+    Gets the current installed game version.
+
+    Credits to An Anime Team for the code that does the magic:
+    https://github.com/an-anime-team/anime-game-core/blob/main/src/games/genshin/game.rs#L52
+
+    If the above method fails, it'll fallback to read the config.ini file
+    for the version, which is not recommended (as described in
+    `get_version_config()` docs)
+
+    This returns (0, 0, 0) if the version could not be found
+    (usually indicates the game is not installed), and in fact `is_installed()` uses
+    this method to check if the game is installed too.
+
+    Returns:
+        tuple[int, int, int]: The version as a tuple of integers.
+    """
+
+    data_file = game.data_folder().joinpath("globalgamemanagers")
+    if not data_file.exists():
+        return game.get_version_config()
+
+    def bytes_to_int(byte_array: list[bytes]) -> int:
+        bytes_as_int = int.from_bytes(byte_array, byteorder="big")
+        actual_int = bytes_as_int - 48  # 48 is the ASCII code for 0
+        return actual_int
+
+    version_bytes: list[list[bytes]] = [[], [], []]
+    version_ptr = 0
+    correct = True
+    try:
+        with data_file.open("rb") as f:
+            f.seek(4000)
+            for byte in f.read(10000):
+                match byte:
+                    case 0:
+                        version_bytes = [[], [], []]
+                        version_ptr = 0
+                        correct = True
+                    case 46:
+                        version_ptr += 1
+                        if version_ptr > 2:
+                            correct = False
+                    case 95:
+                        if (
+                            correct
+                            and len(version_bytes[0]) > 0
+                            and len(version_bytes[1]) > 0
+                            and len(version_bytes[2]) > 0
+                        ):
+                            return (
+                                bytes_to_int(version_bytes[0]),
+                                bytes_to_int(version_bytes[1]),
+                                bytes_to_int(version_bytes[2]),
+                            )
+                    case _:
+                        if correct and byte in b"0123456789":
+                            version_bytes[version_ptr].append(byte)
+                        else:
+                            correct = False
+    except Exception:
+        pass
+    # Fallback to config.ini
+    return game.get_version_config()

vollerei/game/hsr/constants.py

@@ -0,0 +1,15 @@
+MD5SUMS = {
+    "1.0.5": {
+        "cn": {
+            "StarRailBase.dll": "66c42871ce82456967d004ccb2d7cf77",
+            "UnityPlayer.dll": "0c866c44bb3752031a8c12ffe935b26f",
+        },
+        "os": {
+            "StarRailBase.dll": "8aa3790aafa3dd176678392f3f93f435",
+            "UnityPlayer.dll": "f17b9b7f9b8c9cbd211bdff7771a80c2",
+        },
+    }
+}
+# Patches
+ASTRA_REPO = "https://notabug.org/mkrsym1/astra"
+JADEITE_REPO = "https://codeberg.org/mkrsym1/jadeite/"

vollerei/game/hsr/functions.py

@@ -0,0 +1,98 @@
+from hashlib import md5
+from vollerei.common.enums import GameChannel
+from vollerei.abc.launcher.game import GameABC
+from vollerei.game.hsr.constants import MD5SUMS
+
+
+def get_channel(game: GameABC) -> GameChannel | None:
+    """
+    Gets the current game channel.
+
+    Only works for Star Rail version 1.0.5, other versions will return the
+    overridden channel or GameChannel.Overseas if no channel is overridden.
+
+    This is not needed for game patching, since the patcher will automatically
+    detect the channel.
+
+    Returns:
+        GameChannel: The current game channel.
+    """
+    version = game.version_override or game.get_version()
+    if version == (1, 0, 5):
+        for channel, v in MD5SUMS["1.0.5"].values():
+            for file, md5sum in v.values():
+                if md5(game.path.joinpath(file).read_bytes()).hexdigest() != md5sum:
+                    continue
+                match channel:
+                    case "cn":
+                        return GameChannel.China
+                    case "os":
+                        return GameChannel.Overseas
+    return None
+
+
+def get_version(game: GameABC) -> tuple[int, int, int]:
+    """
+    Gets the current installed game version.
+
+    Credits to An Anime Team for the code that does the magic:
+    https://github.com/an-anime-team/anime-game-core/blob/main/src/games/star_rail/game.rs#L49
+
+    If the above method fails, it'll fallback to read the config.ini file
+    for the version, which is not recommended (as described in
+    `get_version_config()` docs)
+
+    This returns (0, 0, 0) if the version could not be found
+    (usually indicates the game is not installed), and in fact `is_installed()` uses
+    this method to check if the game is installed too.
+
+    Returns:
+        tuple[int, int, int]: The version as a tuple of integers.
+    """
+
+    data_file = game.data_folder().joinpath("data.unity3d")
+    if not data_file.exists():
+        return game.get_version_config()
+
+    def bytes_to_int(byte_array: list[bytes]) -> int:
+        bytes_as_int = int.from_bytes(byte_array, byteorder="big")
+        actual_int = bytes_as_int - 48  # 48 is the ASCII code for 0
+        return actual_int
+
+    version_bytes: list[list[bytes]] = [[], [], []]
+    version_ptr = 0
+    correct = True
+    try:
+        with data_file.open("rb") as f:
+            f.seek(0x7D0)  # 2000 in decimal
+            for byte in f.read(10000):
+                match byte:
+                    case 0:
+                        version_bytes = [[], [], []]
+                        version_ptr = 0
+                        correct = True
+                    case 46:
+                        version_ptr += 1
+                        if version_ptr > 2:
+                            correct = False
+                    case 38:
+                        if (
+                            correct
+                            and len(version_bytes[0]) > 0
+                            and len(version_bytes[1]) > 0
+                            and len(version_bytes[2]) > 0
+                        ):
+                            return (
+                                bytes_to_int(version_bytes[0]),
+                                bytes_to_int(version_bytes[1]),
+                                bytes_to_int(version_bytes[2]),
+                            )
+                    case _:
+                        if correct and byte in b"0123456789":
+                            version_bytes[version_ptr].append(byte)
+                        else:
+                            correct = False
+    except Exception:
+        pass
+    # Fallback to config.ini
+    return game.get_version_config()

vollerei/game/launcher/__init__.py

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

vollerei/game/launcher/api.py

@@ -0,0 +1,32 @@
+from vollerei.common.api import get_game_packages, resource
+from vollerei.common.enums import GameChannel, GameType
+
+
+def get_game_package(
+    game_type: GameType, channel: GameChannel = GameChannel.Overseas
+) -> resource.GameInfo:
+    """
+    Get game package information from the launcher API.
+
+    Doesn't work with HI3 but well, we haven't implemented anything for that game yet.
+
+    Default channel is overseas.
+
+    Args:
+        channel: Game channel to get the resource information from.
+
+    Returns:
+        GameInfo: Game resource information.
+    """
+    find_str: str
+    match game_type:
+        case GameType.HSR:
+            find_str = "hkrpg"
+        case GameType.Genshin:
+            find_str = "hk4e"
+        case GameType.ZZZ:
+            find_str = "nap"
+    game_packages = get_game_packages(channel=channel)
+    for package in game_packages:
+        if find_str in package.game.biz:
+            return package

vollerei/game/launcher/manager.py

@@ -0,0 +1,510 @@
+from configparser import ConfigParser
+from io import IOBase
+from os import PathLike
+from pathlib import Path, PurePath
+from vollerei.abc.launcher.game import GameABC
+from vollerei.common import ConfigFile, functions
+from vollerei.common.api import resource
+from vollerei.common.enums import GameType, VoicePackLanguage, GameChannel
+from vollerei.exceptions.game import (
+    GameAlreadyUpdatedError,
+    GameNotInstalledError,
+    PreDownloadNotAvailable,
+)
+from vollerei.game.launcher import api
+from vollerei.game.hsr import functions as hsr_functions
+from vollerei.game.genshin import functions as genshin_functions
+from vollerei.game.zzz import functions as zzz_functions
+from vollerei import paths
+from vollerei.utils import download
+
+
+class Game(GameABC):
+    """
+    Manages the game installation
+
+    For Star Rail and Zenless Zone Zero:
+
+    Since channel detection isn't implemented yet, most functions assume you're
+    using the overseas version of the game. You can override channel by setting
+    the property `channel_override` to the channel you want to use.
+    """
+
+    def __init__(
+        self, game_type: GameType, path: PathLike = None, cache_path: PathLike = None
+    ):
+        self._path: Path | None = Path(path) if path else None
+        if not cache_path:
+            cache_path = paths.cache_path
+        cache_path = Path(cache_path)
+        self._game_type = game_type
+        self.cache: Path = cache_path.joinpath(f"game/{self._game_type.name.lower()}/")
+        self.cache.mkdir(parents=True, exist_ok=True)
+        self._version_override: tuple[int, int, int] | None = None
+        self._channel_override: GameChannel | None = None
+
+    @property
+    def version_override(self) -> tuple[int, int, int] | None:
+        """
+        Overrides the game version.
+
+        This can be useful if you want to override the version of the game
+        and additionally working around bugs.
+        """
+        return self._version_override
+
+    @version_override.setter
+    def version_override(self, version: tuple[int, int, int] | str | None):
+        if isinstance(version, str):
+            version = tuple(int(i) for i in version.split("."))
+        self._version_override = version
+
+    @property
+    def channel_override(self) -> GameChannel | None:
+        """
+        Overrides the game channel.
+
+        Because game channel detection isn't implemented yet, you may need
+        to use this for some functions to work.
+
+        This can be useful if you want to override the channel of the game
+        and additionally working around bugs.
+        """
+        return self._channel_override
+
+    @channel_override.setter
+    def channel_override(self, channel: GameChannel | str | None):
+        if isinstance(channel, str):
+            channel = GameChannel[channel]
+        self._channel_override = channel
+
+    @property
+    def path(self) -> Path | None:
+        """
+        Paths to the game folder.
+        """
+        return self._path
+
+    @path.setter
+    def path(self, path: PathLike):
+        self._path = Path(path)
+
+    def data_folder(self) -> Path:
+        """
+        Paths to the game data folder.
+
+        Returns:
+            Path: The path to the game data folder.
+        """
+        try:
+            match self._game_type:
+                case GameType.Genshin:
+                    match self.get_channel():
+                        case GameChannel.China:
+                            return self._path.joinpath("YuanShen_Data")
+                        case GameChannel.Overseas:
+                            return self._path.joinpath("GenshinImpact_Data")
+                case GameType.HSR:
+                    return self._path.joinpath("StarRail_Data")
+                case GameType.ZZZ:
+                    return self._path.joinpath("ZenlessZoneZero_Data")
+        except AttributeError:
+            raise GameNotInstalledError("Game path is not set.")
+
+    def is_installed(self) -> bool:
+        """
+        Checks if the game is installed.
+
+        Returns:
+            bool: True if the game is installed, False otherwise.
+        """
+        if self._path is None:
+            return False
+        match self._game_type:
+            case GameType.Genshin:
+                match self.get_channel():
+                    case GameChannel.China:
+                        if not self._path.joinpath("YuanShen.exe").exists():
+                            return False
+                    case GameChannel.Overseas:
+                        if not self._path.joinpath("GenshinImpact.exe").exists():
+                            return False
+            case GameType.HSR:
+                if not self._path.joinpath("StarRail.exe").exists():
+                    return False
+            case GameType.ZZZ:
+                if not self._path.joinpath("ZenlessZoneZero.exe").exists():
+                    return False
+        if not self.data_folder().is_dir():
+            return False
+        if self.get_version() == (0, 0, 0):
+            return False
+        return True
+
+    def get_channel(self) -> GameChannel:
+        """
+        Gets the current game channel.
+
+        Only works for Genshin and Star Rail version 1.0.5, other versions will return
+        the overridden channel or GameChannel.Overseas if no channel is overridden.
+
+        This is not needed for game patching, since the patcher will automatically
+        detect the channel.
+
+        Returns:
+            GameChannel: The current game channel.
+        """
+        match self._game_type:
+            case GameType.HSR:
+                return (
+                    hsr_functions.get_channel(self)
+                    or self._channel_override
+                    or GameChannel.Overseas
+                )
+            case GameType.Genshin:
+                return (
+                    genshin_functions.get_channel(self)
+                    or self._channel_override
+                    or GameChannel.Overseas
+                )
+            case _:
+                return self._channel_override or GameChannel.Overseas
+
+    def get_version_config(self) -> tuple[int, int, int]:
+        """
+        Gets the current installed game version from config.ini.
+
+        Using this is not recommended, as only official launcher creates
+        and uses this file, instead you should use `get_version()`.
+
+        This returns (0, 0, 0) if the version could not be found.
+
+        Returns:
+            tuple[int, int, int]: Game version.
+        """
+        cfg_file = self._path.joinpath("config.ini")
+        if not cfg_file.exists():
+            return (0, 0, 0)
+        cfg = ConfigFile(cfg_file)
+        # Fk u miHoYo
+        if "general" in cfg.sections():
+            version_str = cfg.get("general", "game_version", fallback="0.0.0")
+        elif "General" in cfg.sections():
+            version_str = cfg.get("General", "game_version", fallback="0.0.0")
+        else:
+            return (0, 0, 0)
+        if version_str.count(".") != 2:
+            return (0, 0, 0)
+        try:
+            version = tuple(int(i) for i in version_str.split("."))
+        except Exception:
+            return (0, 0, 0)
+        return version
+
+    def set_version_config(self):
+        """
+        Sets the current installed game version to config.ini.
+
+        Only works for the global version of the game, not the Chinese one (since I don't have
+        them installed to test).
+
+        This method is meant to keep compatibility with the official launcher only.
+        """
+        cfg_file = self._path.joinpath("config.ini")
+        if cfg_file.exists():
+            cfg = ConfigFile(cfg_file)
+            cfg.set("general", "game_version", self.get_version_str())
+            cfg.save()
+        else:
+            cfg_dict = {
+                "general": {
+                    "channel": 1,
+                    "cps": "hyp_hoyoverse",
+                    "game_version": self.get_version_str(),
+                    "sub_channel": 0,
+                    # This probably should be fetched from the server but well
+                    "plugin_n06mjyc2r3_version": "1.1.0",
+                    "uapc": None,  # Honestly what's this?
+                }
+            }
+
+            match self._game_type:
+                case GameType.Genshin:
+                    cfg_dict["general"]["uapc"] = {
+                        "hk4e_global": {"uapc": "f55586a8ce9f_"},
+                        "hyp": {"uapc": "f55586a8ce9f_"},
+                    }
+                case GameType.HSR:
+                    cfg_dict["general"]["uapc"] = {
+                        "hkrpg_global": {"uapc": "f5c7c6262812_"},
+                        "hyp": {"uapc": "f55586a8ce9f_"},
+                    }
+                case GameType.ZZZ:
+                    cfg_dict["general"]["uapc"] = {
+                        "nap_global": {"uapc": "f55586a8ce9f_"},
+                        "hyp": {"uapc": "f55586a8ce9f_"},
+                    }
+            cfg = ConfigParser()
+            cfg.read_dict(cfg_dict)
+            cfg.write(cfg_file.open("w"))
+
+    def get_version(self) -> tuple[int, int, int]:
+        """
+        Gets the current installed game version.
+
+        Credits to An Anime Team for the code that does the magic, see the source
+        in `hsr/functions.py`, `genshin/functions.py` and `zzz/functions.py` for more info
+
+        If the above method fails, it'll fallback to read the config.ini file
+        for the version, which is not recommended (as described in
+        `get_version_config()` docs)
+
+        This returns (0, 0, 0) if the version could not be found
+        (usually indicates the game is not installed), and in fact `is_installed()` uses
+        this method to check if the game is installed too.
+
+        Returns:
+            tuple[int, int, int]: The version as a tuple of integers.
+        """
+        match self._game_type:
+            case GameType.HSR:
+                return hsr_functions.get_version(self)
+            case GameType.Genshin:
+                return genshin_functions.get_version(self)
+            case GameType.ZZZ:
+                return zzz_functions.get_version(self)
+            case _:
+                return self.get_version_config()
+
+    def get_version_str(self) -> str:
+        """
+        Gets the current installed game version as a string.
+
+        Because this method uses `get_version()`, you should read the docs of
+        that method too.
+
+        Returns:
+            str: The version as a string.
+        """
+        return ".".join(str(i) for i in self.get_version())
+
+    def get_installed_voicepacks(self) -> list[VoicePackLanguage]:
+        """
+        Gets the installed voicepacks.
+
+        Returns:
+            list[VoicePackLanguage]: A list of installed voicepacks.
+        """
+        if not self.is_installed():
+            raise GameNotInstalledError("Game is not installed.")
+        voicepacks = []
+        blacklisted_words = ["SFX"]
+        audio_package: Path
+        match self._game_type:
+            case GameType.Genshin:
+                audio_package = self.data_folder().joinpath(
+                    "StreamingAssets/AudioAssets/"
+                )
+                if audio_package.joinpath("AudioPackage").is_dir():
+                    audio_package = audio_package.joinpath("AudioPackage")
+            case GameType.HSR:
+                audio_package = self.data_folder().joinpath(
+                    "Persistent/Audio/AudioPackage/Windows/"
+                )
+            case GameType.ZZZ:
+                audio_package = self.data_folder().joinpath(
+                    "StreamingAssets/Audio/Windows/Full/"
+                )
+        for child in audio_package.iterdir():
+            if child.resolve().is_dir() and child.name not in blacklisted_words:
+                name = child.name
+                if name.startswith("English"):
+                    name = "English"
+                voicepack: VoicePackLanguage
+                try:
+                    if self._game_type == GameType.ZZZ:
+                        voicepack = VoicePackLanguage.from_zzz_name(child.name)
+                    else:
+                        voicepack = VoicePackLanguage[name]
+                    voicepacks.append(voicepack)
+                except (ValueError, KeyError):
+                    pass
+        return voicepacks
+
+    def get_remote_game(
+        self, pre_download: bool = False
+    ) -> resource.Main | resource.PreDownload:
+        """
+        Gets the current game information from remote.
+
+        Args:
+            pre_download (bool): Whether to get the pre-download version.
+                Defaults to False.
+
+        Returns:
+            A `Main` or `PreDownload` object that contains the game information.
+        """
+        channel = self._channel_override or self.get_channel()
+        if pre_download:
+            game = api.get_game_package(
+                game_type=self._game_type, channel=channel
+            ).pre_download
+            if not game:
+                raise PreDownloadNotAvailable("Pre-download version is not available.")
+            return game
+        return api.get_game_package(game_type=self._game_type, channel=channel).main
+
+    def get_update(self, pre_download: bool = False) -> resource.Patch | None:
+        """
+        Gets the current game update.
+
+        Args:
+            pre_download (bool): Whether to get the pre-download version.
+                Defaults to False.
+
+        Returns:
+            A `Patch` object that contains the update information or
+            `None` if the game is not installed or already up-to-date.
+        """
+        if not self.is_installed():
+            return None
+        version = (
+            ".".join(str(x) for x in self._version_override)
+            if self._version_override
+            else self.get_version_str()
+        )
+        for patch in self.get_remote_game(pre_download=pre_download).patches:
+            if patch.version == version:
+                return patch
+        return None
+
+    def repair_file(
+        self,
+        file: PathLike,
+        pre_download: bool = False,
+        game_info: resource.Game = None,
+    ) -> None:
+        """
+        Repairs a game file.
+
+        This will automatically handle backup and restore the file if the repair
+        fails.
+
+        Args:
+            file (PathLike): The file to repair.
+            pre_download (bool): Whether to get the pre-download version.
+                Defaults to False.
+        """
+        return self.repair_files([file], pre_download=pre_download, game_info=game_info)
+
+    def repair_files(
+        self,
+        files: list[PathLike],
+        pre_download: bool = False,
+        game_info: resource.Game = None,
+    ) -> None:
+        """
+        Repairs multiple game files.
+
+        This will automatically handle backup and restore the file if the repair
+        fails.
+
+        Args:
+            files (PathLike): The files to repair.
+            pre_download (bool): Whether to get the pre-download version.
+                Defaults to False.
+            game_info (resource.Game): The game information to use for repair.
+        """
+        functions.repair_files(
+            self, files, pre_download=pre_download, game_info=game_info
+        )
+
+    def repair_game(self) -> None:
+        """
+        Tries to repair the game by reading "pkg_version" file and downloading the
+        mismatched files from the server.
+        """
+        functions.repair_game(self)
+
+    def install_archive(self, archive_file: PathLike | IOBase) -> None:
+        """
+        Applies an install archive to the game, it can be the game itself or a
+        voicepack one.
+
+        `archive_file` can be a path to the archive file or a file-like object,
+        like if you have very high amount of RAM and want to download the archive
+        to memory instead of disk, this can be useful for you.
+
+        Args:
+            archive_file (PathLike | IOBase): The archive file.
+        """
+        if not isinstance(archive_file, IOBase):
+            archive_file = Path(archive_file)
+        functions.install_archive(self, archive_file)
+
+    def apply_update_archive(
+        self, archive_file: PathLike | IOBase, auto_repair: bool = True
+    ) -> None:
+        """
+        Applies an update archive to the game, it can be the game update or a
+        voicepack update.
+
+        `archive_file` can be a path to the archive file or a file-like object,
+        like if you have very high amount of RAM and want to download the update
+        to memory instead of disk, this can be useful for you.
+
+        `auto_repair` is used to determine whether to repair the file if it's
+        broken. If it's set to False, then it'll raise an exception if the file
+        is broken.
+
+        Args:
+            archive_file (PathLike | IOBase): The archive file.
+            auto_repair (bool, optional): Whether to repair the file if it's broken.
+                Defaults to True.
+        """
+        if not self.is_installed():
+            raise GameNotInstalledError("Game is not installed.")
+        if not isinstance(archive_file, IOBase):
+            archive_file = Path(archive_file)
+        # Hello hell again, dealing with HDiffPatch and all the things again.
+        functions.apply_update_archive(self, archive_file, auto_repair=auto_repair)
+
+    def install_update(
+        self, update_info: resource.Patch = None, auto_repair: bool = True
+    ):
+        """
+        Installs an update from a `Patch` object.
+
+        You may want to download the update manually and pass it to
+        `apply_update_archive()` instead for better control, and after that
+        execute `set_version_config()` to set the game version.
+
+        Args:
+            update_info (Diff, optional): The update information. Defaults to None.
+            auto_repair (bool, optional): Whether to repair the file if it's broken.
+                Defaults to True.
+        """
+        if not self.is_installed():
+            raise GameNotInstalledError("Game is not installed.")
+        if not update_info:
+            update_info = self.get_update()
+        if not update_info or update_info.version == self.get_version_str():
+            raise GameAlreadyUpdatedError("Game is already updated.")
+        update_url = update_info.game_pkgs[0].url
+        # Base game update
+        archive_file = self.cache.joinpath(PurePath(update_url).name)
+        download(update_url, archive_file)
+        self.apply_update_archive(archive_file=archive_file, auto_repair=auto_repair)
+        # Get installed voicepacks
+        installed_voicepacks = self.get_installed_voicepacks()
+        # Voicepack update
+        for remote_voicepack in update_info.audio_pkgs:
+            if remote_voicepack.language not in installed_voicepacks:
+                continue
+            # Voicepack is installed, update it
+            archive_file = self.cache.joinpath(PurePath(remote_voicepack.url).name)
+            download(remote_voicepack.url, archive_file)
+            self.apply_update_archive(
+                archive_file=archive_file, auto_repair=auto_repair
+            )
+        self.set_version_config()

vollerei/game/patcher.py

@@ -0,0 +1,223 @@
+from enum import Enum
+from shutil import copy2, rmtree
+from packaging import version
+from vollerei.abc.patcher import PatcherABC
+from vollerei.common import telemetry
+from vollerei.exceptions.game import GameNotInstalledError
+from vollerei.exceptions.patcher import (
+    VersionNotSupportedError,
+    PatcherError,
+    PatchUpdateError,
+)
+from vollerei.game.launcher.manager import Game, GameChannel
+from vollerei.utils import download_and_extract, Git, Xdelta3
+from vollerei.paths import tools_data_path
+from vollerei.hsr.constants import ASTRA_REPO, JADEITE_REPO
+
+
+class PatchType(Enum):
+    """
+    Patch type
+
+    Astra: The old patch which patch the game directly (not recommended).
+    Jadeite: The new patch which patch the game in memory by DLL injection.
+    """
+
+    Astra = 0
+    Jadeite = 1
+
+
+class Patcher(PatcherABC):
+    """
+    Patch helper for HSR and HI3.
+
+    By default this will use Jadeite as it is maintained and more stable.
+    """
+
+    def __init__(self, patch_type: PatchType = PatchType.Jadeite):
+        self._patch_type: PatchType = patch_type
+        self._path = tools_data_path.joinpath("patcher")
+        self._path.mkdir(parents=True, exist_ok=True)
+        self._jadeite = self._path.joinpath("jadeite")
+        self._astra = self._path.joinpath("astra")
+        self._git = Git()
+        self._xdelta3 = Xdelta3()
+
+    @property
+    def patch_type(self) -> PatchType:
+        """
+        Patch type, can be either Astra or Jadeite
+        """
+        return self._patch_type
+
+    @patch_type.setter
+    def patch_type(self, value: PatchType):
+        self._patch_type = value
+
+    def _update_astra(self):
+        self._git.pull_or_clone(ASTRA_REPO, self._astra)
+
+    def _update_jadeite(self):
+        release_info = self._git.get_latest_release(JADEITE_REPO)
+        file = self._git.get_latest_release_dl(release_info)[0]
+        file_version = release_info["tag_name"][1:]  # Remove "v" prefix
+        current_version = None
+        if self._jadeite.joinpath("version").exists():
+            with open(self._jadeite.joinpath("version"), "r") as f:
+                current_version = f.read()
+        if current_version:
+            if version.parse(file_version) <= version.parse(current_version):
+                return
+        download_and_extract(file, self._jadeite)
+        with open(self._jadeite.joinpath("version"), "w") as f:
+            f.write(file_version)
+
+    def update_patch(self):
+        """
+        Update the patch
+        """
+        try:
+            match self._patch_type:
+                case PatchType.Astra:
+                    self._update_astra()
+                case PatchType.Jadeite:
+                    self._update_jadeite()
+        except Exception as e:
+            raise PatchUpdateError("Failed to update patch.") from e
+
+    def _patch_astra(self, game: Game):
+        if game.get_version() != (1, 0, 5):
+            raise VersionNotSupportedError(
+                "Only version 1.0.5 is supported by Astra patch."
+            )
+        self._update_astra()
+        file_type = None
+        match game.get_channel():
+            case GameChannel.China:
+                file_type = "cn"
+            case GameChannel.Overseas:
+                file_type = "os"
+        # Backup and patch
+        for file in ["UnityPlayer.dll", "StarRailBase.dll"]:
+            game.path.joinpath(file).rename(game.path.joinpath(f"{file}.bak"))
+            self._xdelta3.patch_file(
+                self._astra.joinpath(f"{file_type}/diffs/{file}.vcdiff"),
+                game.path.joinpath(f"{file}.bak"),
+                game.path.joinpath(file),
+            )
+        # Copy files
+        for file in self._astra.joinpath(f"{file_type}/files/").rglob("*"):
+            if file.suffix == ".bat":
+                continue
+            if file.is_dir():
+                game.path.joinpath(
+                    file.relative_to(self._astra.joinpath(f"{file_type}/files/"))
+                ).mkdir(parents=True, exist_ok=True)
+            copy2(
+                file,
+                game.path.joinpath(
+                    file.relative_to(self._astra.joinpath(f"{file_type}/files/"))
+                ),
+            )
+
+    def _patch_jadeite(self):
+        """
+        "Patch" the game with Jadeite patch.
+
+        Unlike Astra patch, Jadeite patch does not modify the game files directly
+        but uses DLLs to patch the game in memory and it has an injector to do that
+        automatically.
+        """
+        self._update_jadeite()
+        return self._jadeite
+
+    def _unpatch_astra(self, game: Game):
+        if game.get_version() != (1, 0, 5):
+            raise VersionNotSupportedError(
+                "Only version 1.0.5 is supported by Astra patch."
+            )
+        self._update_astra()
+        file_type = None
+        match game.get_channel():
+            case GameChannel.China:
+                file_type = "cn"
+            case GameChannel.Overseas:
+                file_type = "os"
+        # Restore
+        for file in ["UnityPlayer.dll", "StarRailBase.dll"]:
+            if game.path.joinpath(f"{file}.bak").exists():
+                game.path.joinpath(file).unlink()
+                game.path.joinpath(f"{file}.bak").rename(game.path.joinpath(file))
+        # Remove files
+        for file in self._astra.joinpath(f"{file_type}/files/").rglob("*"):
+            if file.suffix == ".bat":
+                continue
+            file_rel = file.relative_to(self._astra.joinpath(f"{file_type}/files/"))
+            game_path = game.path.joinpath(file_rel)
+            if game_path.is_file():
+                game_path.unlink()
+            elif game_path.is_dir():
+                try:
+                    game_path.rmdir()
+                except OSError:
+                    pass
+
+    def _unpatch_jadeite(self):
+        rmtree(self._jadeite, ignore_errors=True)
+
+    def patch_game(self, game: Game):
+        """
+        Patch the game
+
+        If you use Jadeite (by default), this will just download Jadeite files
+        and won't actually patch the game because Jadeite will do that automatically.
+
+        Args:
+            game (Game): The game to patch
+        """
+        if not game.is_installed():
+            raise PatcherError(GameNotInstalledError("Game is not installed"))
+        match self._patch_type:
+            case PatchType.Astra:
+                self._patch_astra(game)
+            case PatchType.Jadeite:
+                return self._patch_jadeite()
+
+    def unpatch_game(self, game: Game):
+        """
+        Unpatch the game
+
+        If you use Jadeite (by default), this will just delete Jadeite files.
+        Note that Honkai Impact 3rd uses Jadeite too, so executing this will
+        delete the files needed by both games.
+
+        Args:
+            game (Game): The game to unpatch
+        """
+        if not game.is_installed():
+            raise PatcherError(GameNotInstalledError("Game is not installed"))
+        match self._patch_type:
+            case PatchType.Astra:
+                self._unpatch_astra(game)
+            case PatchType.Jadeite:
+                self._unpatch_jadeite()
+
+    def check_telemetry(self) -> list[str]:
+        """
+        Check if telemetry servers are accessible by the user
+
+        Returns:
+            list[str]: A list of telemetry servers that are accessible
+        """
+        return telemetry.check_telemetry()
+
+    def block_telemetry(self, telemetry_list: list[str] = None):
+        """
+        Block the telemetry servers
+
+        If telemetry_list is not provided, it will be checked automatically.
+
+        Args:
+            telemetry_list (list[str], optional): A list of telemetry servers to block.
+        """
+        telemetry.block_telemetry(telemetry_list)

vollerei/game/zzz/functions.py

@@ -0,0 +1,66 @@
+from vollerei.abc.launcher.game import GameABC
+
+
+def get_version(game: GameABC) -> tuple[int, int, int]:
+    """
+    Gets the current installed game version.
+
+    Credits to An Anime Team for the code that does the magic:
+    https://github.com/an-anime-team/anime-game-core/blob/main/src/games/zzz/game.rs#L49
+
+    If the above method fails, it'll fallback to read the config.ini file
+    for the version, which is not recommended (as described in
+    `get_version_config()` docs)
+
+    This returns (0, 0, 0) if the version could not be found
+    (usually indicates the game is not installed), and in fact `is_installed()` uses
+    this method to check if the game is installed too.
+
+    Returns:
+        tuple[int, int, int]: The version as a tuple of integers.
+    """
+
+    data_file = game.data_folder().joinpath("globalgamemanagers")
+    if not data_file.exists():
+        return game.get_version_config()
+
+    def bytes_to_int(byte_array: list[bytes]) -> int:
+        bytes_as_int = int.from_bytes(byte_array, byteorder="big")
+        actual_int = bytes_as_int - 48  # 48 is the ASCII code for 0
+        return actual_int
+
+    version_bytes: list[list[bytes]] = [[], [], []]
+    version_ptr = 0
+    correct = True
+    try:
+        with data_file.open("rb") as f:
+            f.seek(4000)
+            for byte in f.read(10000):
+                match byte:
+                    case 0:
+                        if (
+                            correct
+                            and len(version_bytes[0]) > 0
+                            and len(version_bytes[1]) > 0
+                            and len(version_bytes[2]) > 0
+                        ):
+                            found_version = tuple(
+                                bytes_to_int(i) for i in version_bytes
+                            )
+                            return found_version
+                        version_bytes = [[], [], []]
+                        version_ptr = 0
+                        correct = True
+                    case b".":
+                        version_ptr += 1
+                        if version_ptr > 2:
+                            correct = False
+                    case _:
+                        if correct and byte in b"0123456789":
+                            version_bytes[version_ptr].append(byte)
+                        else:
+                            correct = False
+    except Exception:
+        pass
+    # Fallback to config.ini
+    return game.get_version_config()

vollerei/genshin/__init__.py

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

vollerei/genshin/launcher/__init__.py

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

vollerei/genshin/launcher/api.py

@@ -0,0 +1,20 @@
+from vollerei.common.api import get_game_packages, resource
+from vollerei.common.enums import GameChannel
+
+
+def get_game_package(channel: GameChannel = GameChannel.Overseas) -> resource.GameInfo:
+    """
+    Get game package information from the launcher API.
+
+    Default channel is overseas.
+
+    Args:
+        channel: Game channel to get the resource information from.
+
+    Returns:
+        Resource: Game resource information.
+    """
+    game_packages = get_game_packages(channel=channel)
+    for package in game_packages:
+        if "hk4e" in package.game.biz:
+            return package

vollerei/genshin/launcher/game.py

@@ -0,0 +1,12 @@
+from os import PathLike
+from vollerei.game.launcher.manager import Game as CommonGame
+from vollerei.common.enums import GameType
+
+
+class Game(CommonGame):
+    """
+    Manages the game installation
+    """
+
+    def __init__(self, path: PathLike = None, cache_path: PathLike = None):
+        super().__init__(GameType.Genshin, path, cache_path)

vollerei/hi3/launcher/game.py

@@ -0,0 +1,73 @@
+from configparser import ConfigParser
+from io import IOBase
+from os import PathLike
+from pathlib import Path
+from vollerei.abc.launcher.game import GameABC
+from vollerei.common import ConfigFile, functions
+from vollerei.common.api import resource
+from vollerei.common.enums import VoicePackLanguage
+from vollerei.exceptions.game import (
+    GameAlreadyUpdatedError,
+    GameNotInstalledError,
+    PreDownloadNotAvailable,
+    ScatteredFilesNotAvailableError,
+)
+from vollerei.hi3.launcher.enums import GameChannel
+from vollerei.hsr.launcher import api
+from vollerei import paths
+from vollerei.utils import download
+
+
+class Game(GameABC):
+    """
+    Manages the game installation
+
+    Since channel detection isn't implemented yet, most functions assume you're
+    using the overseas version of the game. You can override channel by setting
+    the property `channel_override` to the channel you want to use.
+    """
+
+    def __init__(self, path: PathLike = None, cache_path: PathLike = None):
+        self._path: Path | None = Path(path) if path else None
+        if not cache_path:
+            cache_path = paths.cache_path
+        cache_path = Path(cache_path)
+        self.cache: Path = cache_path.joinpath("game/hi3/")
+        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

vollerei/hi3/patcher.py

@@ -0,0 +1,6 @@
+from vollerei.hsr.patcher import Patcher, PatchType
+
+
+# Re-exports Patcher and PatchType from HSR because they use the same patcher
+# which is Jadeite.
+__all__ = ["Patcher", "PatchType"]

vollerei/hsr/__init__.py

@@ -0,0 +1,6 @@
+# Re-exports
+from vollerei.hsr.patcher import Patcher, PatchType
+from vollerei.hsr.launcher import Game
+
+
+__all__ = ["Patcher", "PatchType", "Game"]

vollerei/hsr/constants.py

@@ -1,13 +1 @@
-latest_version = (1, 1, 0)
+from vollerei.game.hsr.constants import *  # noqa: F403 because we just want to re-export
-md5sums = {
-    "1.0.5": {
-        "cn": {
-            "StarRailBase.dll": "66c42871ce82456967d004ccb2d7cf77",
-            "UnityPlayer.dll": "0c866c44bb3752031a8c12ffe935b26f",
-        },
-        "os": {
-            "StarRailBase.dll": "8aa3790aafa3dd176678392f3f93f435",
-            "UnityPlayer.dll": "f17b9b7f9b8c9cbd211bdff7771a80c2",
-        },
-    }
-}

vollerei/hsr/launcher/__init__.py

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

vollerei/hsr/launcher/api.py

@@ -0,0 +1,20 @@
+from vollerei.common.api import get_game_packages, resource
+from vollerei.common.enums import GameChannel
+
+
+def get_game_package(channel: GameChannel = GameChannel.Overseas) -> resource.GameInfo:
+    """
+    Get game package information from the launcher API.
+
+    Default channel is overseas.
+
+    Args:
+        channel: Game channel to get the resource information from.
+
+    Returns:
+        Resource: Game resource information.
+    """
+    game_packages = get_game_packages(channel=channel)
+    for package in game_packages:
+        if "hkrpg" in package.game.biz:
+            return package

vollerei/hsr/launcher/game.py

@@ -1,44 +1,16 @@
-from hashlib import md5
 from os import PathLike
-from pathlib import Path
+from vollerei.game.launcher.manager import Game as CommonGame
-from enum import Enum
+from vollerei.common.enums import GameType
-from vollerei.abc.launcher.game import GameABC
-from vollerei.hsr.constants import md5sums
 
 
-class GameChannel(Enum):
+class Game(CommonGame):
-    Overseas = 0
+    """
-    China = 1
+    Manages the game installation
 
+    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.
+    """
 
-class Game(GameABC):
+    def __init__(self, path: PathLike = None, cache_path: PathLike = None):
-    def __init__(self, path: PathLike = None):
+        super().__init__(GameType.HSR, path, cache_path)
-        self._path: Path | None = Path(path) if path else None
-
-    @property
-    def path(self) -> Path | None:
-        return self._path
-
-    def is_installed(self) -> bool:
-        if self._path is None:
-            return False
-        if (
-            not self._path.joinpath("StarRail.exe").exists()
-            or not self._path.joinpath("StarRailBase.dll").exists()
-        ):
-            return False
-
-    def get_channel(self) -> GameChannel:
-        if self.get_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

vollerei/hsr/patcher.py

@@ -1,14 +1,19 @@
-from pathlib import Path
 from enum import Enum
-from shutil import copy2
+from shutil import copy2, rmtree
-from distutils.version import StrictVersion
+from packaging import version
 from vollerei.abc.patcher import PatcherABC
+from vollerei.common import telemetry
+from vollerei.common.enums import GameChannel
 from vollerei.exceptions.game import GameNotInstalledError
-from vollerei.exceptions.patcher import VersionNotSupportedError
+from vollerei.exceptions.patcher import (
-from vollerei.hsr.launcher.game import Game, GameChannel
+    VersionNotSupportedError,
+    PatcherError,
+    PatchUpdateError,
+)
+from vollerei.hsr.launcher.game import Game
 from vollerei.utils import download_and_extract, Git, Xdelta3
 from vollerei.paths import tools_data_path
-from vollerei.constants import astra_repo, jadeite_repo
+from vollerei.hsr.constants import ASTRA_REPO, JADEITE_REPO
 
 
 class PatchType(Enum):
@@ -19,11 +24,17 @@ class PatchType(Enum):
     Jadeite: The new patch which patch the game in memory by DLL injection.
     """
 
-    Astra: int = 0
+    Astra = 0
-    Jadeite: int = 1
+    Jadeite = 1
 
 
 class Patcher(PatcherABC):
+    """
+    Patch helper for HSR.
+
+    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")
@@ -35,6 +46,9 @@ class Patcher(PatcherABC):
 
     @property
     def patch_type(self) -> PatchType:
+        """
+        Patch type, can be either Astra or Jadeite
+        """
         return self._patch_type
 
     @patch_type.setter
@@ -42,28 +56,35 @@ class Patcher(PatcherABC):
         self._patch_type = value
 
     def _update_astra(self):
-        self._git.pull_or_clone(astra_repo, self._astra)
+        self._git.pull_or_clone(ASTRA_REPO, self._astra)
 
     def _update_jadeite(self):
-        file = self._git.get_latest_release_dl(jadeite_repo)[0]
+        release_info = self._git.get_latest_release(JADEITE_REPO)
-        file_version = Path(file).stem[1:]
+        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 StrictVersion(file_version) <= StrictVersion(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):
-        match self._patch_type:
+        """
-            case PatchType.Astra:
+        Update the patch
-                self._update_astra()
+        """
-            case PatchType.Jadeite:
+        try:
-                self._update_jadeite()
+            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):
@@ -77,11 +98,9 @@ class Patcher(PatcherABC):
                 file_type = "cn"
             case GameChannel.Overseas:
                 file_type = "os"
-        # Backup
+        # Backup and patch
         for file in ["UnityPlayer.dll", "StarRailBase.dll"]:
             game.path.joinpath(file).rename(game.path.joinpath(f"{file}.bak"))
-        # Patch
-        for file in ["UnityPlayer.dll", "StarRailBase.dll"]:
             self._xdelta3.patch_file(
                 self._astra.joinpath(f"{file_type}/diffs/{file}.vcdiff"),
                 game.path.joinpath(f"{file}.bak"),
@@ -113,11 +132,93 @@ class Patcher(PatcherABC):
         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 GameNotInstalledError("Game is not 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)

vollerei/paths.py

@@ -1,21 +1,50 @@
 from pathlib import Path
+from os import PathLike
 from platformdirs import PlatformDirs
 
 
-base_paths = PlatformDirs("vollerei", "tretrauit", roaming=True)
+class Paths:
-tools_data_path: Path = None
+    """
-tools_cache_path: Path = None
+    Manages the paths
-launcher_cache_path: Path = None
+    """
-utils_cache_path: Path = None
+
+    base_paths = PlatformDirs("vollerei", "tretrauit", roaming=True)
+    cache_path = base_paths.user_cache_path
+    data_path = base_paths.user_data_path
+    tools_data_path = data_path.joinpath("tools")
+    tools_cache_path = cache_path.joinpath("tools")
+    launcher_cache_path = cache_path.joinpath("launcher")
+    utils_cache_path = cache_path.joinpath("utils")
+
+    @staticmethod
+    def set_base_path(path: PathLike):
+        path = Path(path)
+        Paths.base_paths = path
+        Paths.cache_path = Paths.base_paths.joinpath("cache")
+        Paths.data_path = Paths.base_paths
+        Paths.tools_data_path = Paths.data_path.joinpath("tools")
+        Paths.tools_cache_path = Paths.cache_path.joinpath("tools")
+        Paths.launcher_cache_path = Paths.cache_path.joinpath("launcher")
+        Paths.utils_cache_path = Paths.cache_path.joinpath("utils")
 
 
-def init_paths():
+# Aliases
-    global tools_data_path, tools_cache_path, launcher_cache_path, utils_cache_path
+base_paths = Paths.base_paths
-    tools_data_path = base_paths.site_data_path.joinpath("tools")
+cache_path = Paths.cache_path
-    tools_cache_path = base_paths.site_cache_path.joinpath("tools")
+data_path = Paths.data_path
-    launcher_cache_path = base_paths.site_cache_path.joinpath("launcher")
+tools_data_path = Paths.tools_data_path
-    utils_cache_path = base_paths.site_cache_path.joinpath("utils")
+tools_cache_path = Paths.tools_cache_path
-    tools_data_path.mkdir(parents=True, exist_ok=True)
+launcher_cache_path = Paths.launcher_cache_path
-    tools_cache_path.mkdir(parents=True, exist_ok=True)
+utils_cache_path = Paths.utils_cache_path
-    launcher_cache_path.mkdir(parents=True, exist_ok=True)
+
-    utils_cache_path.mkdir(parents=True, exist_ok=True)
+
+def set_base_path(path: PathLike):
+    Paths.set_base_path(path)
+    global base_paths, cache_path, data_path, tools_data_path, tools_cache_path, launcher_cache_path, utils_cache_path
+    base_paths = Paths.base_paths
+    cache_path = Paths.cache_path
+    data_path = Paths.data_path
+    tools_data_path = Paths.tools_data_path
+    tools_cache_path = Paths.tools_cache_path
+    launcher_cache_path = Paths.launcher_cache_path
+    utils_cache_path = Paths.utils_cache_path

vollerei/utils/__init__.py

@@ -1,17 +1,102 @@
 import requests
+import platform
+import shutil
 from zipfile import ZipFile
 from io import BytesIO
 from pathlib import Path
 
+match platform.system():
+    case "Linux":
+        from vollerei.utils.linux import append_text
+    case _:
+
+        def append_text(text: str, path: Path) -> None:
+            # Fallback to our own implementation
+            # Will NOT work if we don't have permission to write to the file
+            try:
+                with path.open("a") as f:
+                    f.write(text)
+            except FileNotFoundError:
+                with path.open(path, "w") as f:
+                    f.write(text)
+            except (PermissionError, OSError) as e:
+                raise PermissionError(
+                    "You don't have permission to write to the file."
+                ) from e
+
+
 # Re-exports
 from vollerei.utils.git import Git
-from vollerei.utils.xdelta3 import Xdelta3
+from vollerei.utils.xdelta3 import Xdelta3, Xdelta3NotInstalledError, Xdelta3PatchError
+from vollerei.utils.xdelta3.exceptions import Xdelta3Error
+from vollerei.utils.hdiffpatch import (
+    HDiffPatch,
+    HPatchZPatchError,
+    NotInstalledError,
+    PlatformNotSupportedError as HPatchZPlatformNotSupportedError,
+)
 
 
-__all__ = ["Git", "Xdelta3", "download_and_extract"]
+__all__ = [
+    "Git",
+    "Xdelta3",
+    "download_and_extract",
+    "HDiffPatch",
+    "write_hosts",
+    "append_text_to_file",
+    "Xdelta3Error",
+    "Xdelta3NotInstalledError",
+    "Xdelta3PatchError",
+    "HPatchZPatchError",
+    "NotInstalledError",
+    "HPatchZPlatformNotSupportedError",
+]
+
+
+def download(
+    url: str,
+    out: Path,
+    file_len: int = None,
+    overwrite: bool = False,
+    stream: bool = True,
+) -> None:
+    """
+    Download to a path.
+
+    Args:
+        url (str): URL to download from.
+        path (Path): Path to download to.
+    """
+    if overwrite:
+        out.unlink(missing_ok=True)
+    headers = {}
+    mode = "a+b"
+    if out.exists():
+        cur_len = (out.stat()).st_size
+        headers |= {"Range": f"bytes={cur_len}-{file_len if file_len else ''}"}
+    else:
+        mode = "w+b"
+        out.parent.mkdir(parents=True, exist_ok=True)
+        out.touch()
+    # Streaming, so we can iterate over the response.
+    response = requests.get(url=url, headers=headers, stream=stream)
+    if response.status_code == 416:
+        print(f"File already downloaded: {out}")
+        return
+    response.raise_for_status()
+    with open(out, mode) as file:
+        shutil.copyfileobj(response.raw, file)
+    return True
 
 
 def download_and_extract(url: str, path: Path) -> None:
+    """
+    Download and extract a zip file to a path.
+
+    Args:
+        url (str): URL to download from.
+        path (Path): Path to extract to.
+    """
     rsp = requests.get(url, stream=True)
     rsp.raise_for_status()
     with BytesIO() as f:
@@ -20,3 +105,27 @@ def download_and_extract(url: str, path: Path) -> None:
         f.seek(0)
         with ZipFile(f) as z:
             z.extractall(path)
+
+
+def append_text_to_file(path: Path, text: str) -> None:
+    try:
+        with open(path, "a") as f:
+            f.write(text)
+    except FileNotFoundError:
+        with open(path, "w") as f:
+            f.write(text)
+    except (PermissionError, OSError):
+        append_text(text, path)
+
+
+def write_hosts(hosts: list[str]) -> None:
+    hosts_str = ""
+    for line in hosts:
+        hosts_str += f"0.0.0.0 {line}\n"
+    match platform.system():
+        case "Linux":
+            append_text_to_file(Path("/etc/hosts"), hosts_str)
+        case "Windows":
+            append_text_to_file(
+                Path("C:/Windows/System32/drivers/etc/hosts"), hosts_str
+            )

vollerei/utils/git/__init__.py

@@ -1,12 +1,16 @@
+import errno
+from io import BytesIO
+import os
 import subprocess
+import stat
+from zipfile import ZipFile
 import requests
 import json
 from pathlib import Path
 from shutil import which, rmtree
 from urllib.parse import urlparse
-from vollerei.constants import utils_cache_path
+from vollerei.paths import utils_cache_path
 from vollerei.utils.git.exceptions import GitCloneError
-from vollerei.utils import download_and_extract
 
 
 class Git:
@@ -36,7 +40,7 @@ class Git:
         """
         Check if the url is a Gitea server
         """
-        rsp = requests.get(f"https://{netloc}/api/v1/meta")
+        rsp = requests.get(f"https://{netloc}/api/v1/version")
         try:
             data: dict = rsp.json()
         except json.JSONDecodeError:
@@ -61,7 +65,15 @@ class Git:
         return data[0]["sha"]
 
     def _download_and_extract_zip(self, url: str, path: Path) -> None:
-        download_and_extract(url, path)
+        # Copied code so it doesn't depend on vollerei.utils.download_and_extract
+        rsp = requests.get(url, stream=True)
+        rsp.raise_for_status()
+        with BytesIO() as f:
+            for chunk in rsp.iter_content(chunk_size=32768):
+                f.write(chunk)
+            f.seek(0)
+            with ZipFile(f) as z:
+                z.extractall(path)
         path.joinpath(".git/PLEASE_INSTALL_GIT").touch()
 
     def _clone(self, url: str, path: str = None) -> None:
@@ -91,8 +103,7 @@ class Git:
         else:
             raise NotImplementedError
 
-    def get_latest_release_dl(self, url: str) -> list[str]:
+    def get_latest_release(self, url: str) -> dict:
-        dl = []
         if Path(url).suffix == ".git":
             url = url[:-4]
         url_info = urlparse(url)
@@ -103,11 +114,18 @@ class Git:
             )
             rsp.raise_for_status()
             data = rsp.json()
-            for asset in data["assets"]:
+            return data
-                dl.append(asset["browser_download_url"])
         else:
             raise NotImplementedError
 
+    def get_latest_release_dl(self, data: dict) -> list[str]:
+        dl = []
+        if not data.get("assets"):
+            return dl
+        for asset in data["assets"]:
+            dl.append(asset["browser_download_url"])
+        return dl
+
     def pull_or_clone(self, url: str, path: str = None) -> None:
         """
         Pulls or clones a git repository
@@ -130,16 +148,39 @@ class Git:
             rmtree(path)
         try:
             if not path_as_path.exists():
-                raise subprocess.CalledProcessError
+                raise subprocess.CalledProcessError(0, cmd="Vollerei-generated error")
-            origin_url = subprocess.check_output(
+            origin_url = (
-                ["git", "config", "--get", "remote.origin.url"], cwd=path
+                subprocess.check_output(
-            ).decode()
+                    ["git", "config", "--get", "remote.origin.url"], cwd=path
+                )
+                .decode()
+                .strip()
+            )
             if origin_url != url:
-                raise subprocess.CalledProcessError
+                raise subprocess.CalledProcessError(0, cmd="Vollerei-generated error")
             subprocess.check_call(["git", "pull"], cwd=path)
         except subprocess.CalledProcessError:
             if path_as_path.exists():
-                rmtree(path)
+                try:
+
+                    def handle_error(func, path, exc):
+                        excvalue = exc[1]
+                        if (
+                            func in (os.rmdir, os.remove)
+                            and excvalue.errno == errno.EACCES
+                        ):
+                            os.chmod(
+                                path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO
+                            )  # 0777
+                            func(path)
+                        else:
+                            raise
+
+                    rmtree(path, ignore_errors=False, onerror=handle_error)
+                except OSError as e:
+                    raise GitCloneError(
+                        f"Failed to delete existing repository {path_as_path}"
+                    ) from e
             try:
                 subprocess.check_call(["git", "clone", url, path])
             except subprocess.CalledProcessError as e:

vollerei/utils/hdiffpatch/__init__.py

@@ -0,0 +1,135 @@
+from os import PathLike
+import platform
+import subprocess
+from zipfile import ZipFile
+import requests
+from io import BytesIO
+from shutil import which
+from vollerei.constants import HDIFFPATCH_GIT_URL
+from vollerei.paths import tools_data_path
+from vollerei.utils.hdiffpatch.exceptions import (
+    HPatchZPatchError,
+    NotInstalledError,
+    PlatformNotSupportedError,
+)
+
+
+class HDiffPatch:
+    """
+    Quick wrapper around HDiffPatch binaries
+
+    Mostly copied from worthless-launcher
+    """
+
+    def __init__(self):
+        self._hdiff = tools_data_path.joinpath("hdiffpatch")
+        self._hdiff.mkdir(parents=True, exist_ok=True)
+
+    @staticmethod
+    def _get_platform_arch():
+        processor = platform.machine()
+        match platform.system():
+            case "Windows":
+                match processor:
+                    case "i386":
+                        return "windows32"
+                    case "x86_64":
+                        return "windows64"
+                    case "AMD64":
+                        return "windows64"
+                    case "arm":
+                        return "windows_arm32"
+                    case "arm64":
+                        return "windows_arm64"
+            case "Linux":
+                match processor:
+                    case "i386":
+                        return "linux32"
+                    case "x86_64":
+                        return "linux64"
+                    case "arm":
+                        return "linux_arm32"
+                    case "arm64":
+                        return "linux_arm64"
+            case "Darwin":
+                return "macos"
+
+        # TODO: Add support for Android & other architectures
+
+        # Rip BSD they need to use Linux compatibility layer to run this
+        # (or use Wine if they prefer that)
+        raise PlatformNotSupportedError(
+            "Only Windows, Linux and macOS are supported by HDiffPatch"
+        )
+
+    def _get_binary(self, exec_name: str, recurse=None) -> str:
+        if which(exec_name):
+            return exec_name
+        if platform.system() == "Windows" and not exec_name.endswith(".exe"):
+            exec_name += ".exe"
+        if self._hdiff.exists() and any(self._hdiff.iterdir()):
+            file = self._hdiff.joinpath(self._get_platform_arch(), exec_name)
+            if file.exists():
+                if platform.system() != "Windows":
+                    file.chmod(0o755)
+                return str(file)
+        if recurse is None:
+            recurse = 3
+        elif recurse == 0:
+            raise NotInstalledError(
+                "HDiffPatch is not installed and can't be automatically installed"
+            )
+        else:
+            recurse -= 1
+        self.download()
+        return self._get_binary(exec_name=exec_name, recurse=recurse)
+
+    def hpatchz(self) -> str | None:
+        return self._get_binary("hpatchz")
+
+    def patch_file(self, in_file: PathLike, out_file: PathLike, patch_file: PathLike):
+        try:
+            subprocess.check_call(
+                [self.hpatchz(), "-f", str(in_file), str(patch_file), str(out_file)]
+            )
+        except subprocess.CalledProcessError as e:
+            raise HPatchZPatchError("Patch error") from e
+
+    def _get_latest_release_info(self) -> dict:
+        split = HDIFFPATCH_GIT_URL.split("/")
+        repo = split[-1]
+        owner = split[-2]
+        rsp = requests.get(
+            "https://api.github.com/repos/{}/{}/releases/latest".format(owner, repo),
+            params={"Headers": "Accept: application/vnd.github.v3+json"},
+        )
+        rsp.raise_for_status()
+        archive_processor = self._get_platform_arch()
+        for asset in rsp.json()["assets"]:
+            if not asset["name"].endswith(".zip"):
+                continue
+            if archive_processor not in asset["name"]:
+                continue
+            return asset
+
+    def get_latest_release_url(self):
+        asset = self._get_latest_release_info()
+        return asset["browser_download_url"]
+
+    def get_latest_release_name(self):
+        asset = self._get_latest_release_info()
+        return asset["name"]
+
+    def download(self):
+        """
+        Download the latest release of HDiffPatch.
+        """
+        url = self.get_latest_release_url()
+        if not url:
+            raise RuntimeError("Unable to find latest release")
+        file = BytesIO()
+        with requests.get(url, stream=True) as r:
+            for chunk in r.iter_content(chunk_size=32768):
+                file.write(chunk)
+        with ZipFile(file) as z:
+            z.extractall(self._hdiff)

vollerei/utils/hdiffpatch/exceptions.py

@@ -0,0 +1,28 @@
+class HDiffPatchError(Exception):
+    """Base class for HDiffPatch errors"""
+
+    pass
+
+
+class HPatchZError(HDiffPatchError):
+    """Raised when hpatchz fails"""
+
+    pass
+
+
+class HPatchZPatchError(HPatchZError):
+    """Raised when hpatchz patch fails"""
+
+    pass
+
+
+class NotInstalledError(HPatchZError):
+    """Raised when HDiffPatch is not installed"""
+
+    pass
+
+
+class PlatformNotSupportedError(HPatchZError):
+    """Raised when HDiffPatch is not available for your platform"""
+
+    pass

vollerei/utils/linux.py

@@ -0,0 +1,39 @@
+import subprocess
+from pathlib import Path
+
+
+__all__ = ["exec_su", "write_text", "append_text"]
+
+
+def exec_su(args, stdin: str = None):
+    """Execute a command using pkexec (friendly gui)"""
+    if not Path("/usr/bin/pkexec").exists():
+        raise FileNotFoundError("pkexec not found.")
+    proc = subprocess.Popen(
+        args, shell=True, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL
+    )
+    if stdin:
+        proc.stdin.write(stdin.encode())
+        proc.stdin.close()
+    proc.wait()
+    match proc.returncode:
+        case 127:
+            raise OSError("Authentication failed.")
+        case 128:
+            raise RuntimeError("User cancelled the authentication.")
+
+    return proc
+
+
+def write_text(text, path: str | Path):
+    """Write text to a file using pkexec (friendly gui)"""
+    if isinstance(path, Path):
+        path = str(path)
+    exec_su(f'tee "{path}"', stdin=text)
+
+
+def append_text(text, path: str | Path):
+    """Append text to a file using pkexec (friendly gui)"""
+    if isinstance(path, Path):
+        path = str(path)
+    exec_su(f'tee -a "{path}"', stdin=text)

vollerei/utils/xdelta3/__init__.py

@@ -2,9 +2,10 @@ import platform
 import subprocess
 import requests
 from os import PathLike
+from io import BytesIO
 from zipfile import ZipFile
 from shutil import which
-from vollerei.constants import tools_cache_path
+from vollerei.paths import tools_data_path
 from vollerei.utils.xdelta3.exceptions import (
     Xdelta3NotInstalledError,
     Xdelta3PatchError,
@@ -17,7 +18,7 @@ class Xdelta3:
     """
 
     def __init__(self) -> None:
-        self._xdelta3_path = tools_cache_path.joinpath("xdelta3")
+        self._xdelta3_path = tools_data_path.joinpath("xdelta3")
         self._xdelta3_path.mkdir(parents=True, exist_ok=True)
 
     def _get_binary(self, recurse=None) -> str:
@@ -25,7 +26,7 @@ class Xdelta3:
             return "xdelta3"
         if platform.system() == "Windows":
             for path in self._xdelta3_path.glob("*.exe"):
-                return path
+                return str(path)
             if recurse is None:
                 recurse = 3
             elif recurse == 0:
@@ -35,7 +36,7 @@ class Xdelta3:
             else:
                 recurse -= 1
             self.download()
-            return self.get_binary(recurse=recurse)
+            return self._get_binary(recurse=recurse)
         raise Xdelta3NotInstalledError("xdelta3 is not installed")
 
     def get_binary(self) -> str:
@@ -60,11 +61,11 @@ class Xdelta3:
                 url = "https://github.com/jmacd/xdelta-gpl/releases/download/v3.1.0/xdelta3-3.1.0-i686.exe.zip"
             case "i686":
                 url = "https://github.com/jmacd/xdelta-gpl/releases/download/v3.1.0/xdelta3-3.1.0-i686.exe.zip"
+        file = BytesIO()
         with requests.get(url, stream=True) as r:
-            with open(self._xdelta3_path.joinpath("xdelta3.zip"), "wb") as f:
+            for chunk in r.iter_content(chunk_size=32768):
-                for chunk in r.iter_content(chunk_size=32768):
+                file.write(chunk)
-                    f.write(chunk)
+        with ZipFile(file) as z:
-        with ZipFile(self._xdelta3_path.joinpath("xdelta3.zip")) as z:
             z.extractall(self._xdelta3_path)
 
     def patch_file(self, patch: PathLike, target: PathLike, output: PathLike):

vollerei/zzz/__init__.py

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

vollerei/zzz/launcher/__init__.py

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

vollerei/zzz/launcher/api.py

@@ -0,0 +1,20 @@
+from vollerei.common.api import get_game_packages, resource
+from vollerei.common.enums import GameChannel
+
+
+def get_game_package(channel: GameChannel = GameChannel.Overseas) -> resource.GameInfo:
+    """
+    Get game package information from the launcher API.
+
+    Default channel is overseas.
+
+    Args:
+        channel: Game channel to get the resource information from.
+
+    Returns:
+        Resource: Game resource information.
+    """
+    game_packages = get_game_packages(channel=channel)
+    for package in game_packages:
+        if "nap" in package.game.biz:
+            return package

vollerei/zzz/launcher/game.py

@@ -0,0 +1,16 @@
+from os import PathLike
+from vollerei.game.launcher.manager import Game as CommonGame
+from vollerei.common.enums import GameType
+
+
+class Game(CommonGame):
+    """
+    Manages the game installation
+
+    Since channel detection isn't implemented yet, most functions assume you're
+    using the overseas version of the game. You can override channel by setting
+    the property `channel_override` to the channel you want to use.
+    """
+
+    def __init__(self, path: PathLike = None, cache_path: PathLike = None):
+        super().__init__(GameType.ZZZ, path, cache_path)