tretrauit
/
worthless-launcher
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

refactor: move gamemanager.py to game & rename to Game 

Also add back some staticmethods, because it'll be useful if you only import the Game class
This commit is contained in:
tretrauit 2022-09-20 16:35:56 +07:00
parent 811403bdfa
commit 96d1c7e8d3
Signed by: tretrauit
GPG Key ID: CDDE1C97EE305DAF
11 changed files with 206 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
worthless/__init__.py
View File

@ -1,4 +1,4 @@
from worthless import launcher, gamemanager
from worthless import launcher, game
Launcher = launcher.Launcher
GameManager = gamemanager.GameManager
Game = game.Game

2
worthless/cli.py
View File

@ -6,7 +6,7 @@ import asyncio
import appdirs
from pathlib import Path
from worthless.launcher import Launcher
from worthless.gamemanager import GameManager as Installer
from worthless.game.game import GameManager as Installer
from worthless.patcher import Patcher
import worthless.constants as constants

2
worthless/game/__init__.py Normal file
View File

@ -0,0 +1,2 @@
from worthless.game.game import *
from worthless.game.helper import Helper

8
worthless/enums.py → worthless/game/enums.py
View File

@ -1,13 +1,13 @@
from enum import Enum
class GameVariant(Enum):
class Variant(Enum):
INTERNATIONAL = "international"
CHINESE = "chinese"
BILIBILI = "bilibili"
class GameVersionLocation(Enum):
class VersionLocation(Enum):
GAME_FILE = "globalgamemanagers"
LAUNCHER_CONFIG = "config.ini"
@ -24,13 +24,13 @@ class VoicepackArchiveLanguage(Enum):
Chinese = "Chinese"
class GameInstallStatus(Enum):
class InstallStatus(Enum):
REMOVING_OLD_GAME = 0
EXTRACTING = 1
COMPLETED = 2
class GameUpdateStatus(Enum):
class UpdateStatus(Enum):
DOWNLOADING_PATCHER = 0
PREPARING_PATCH = 1
PATCHING = 2

170
worthless/gamemanager.py → worthless/game/game.py
View File

@ -8,21 +8,20 @@ import logging
from os import PathLike
from pathlib import Path
import worthless.helper as helper
from worthless import constants
from worthless.launcher import Launcher
from worthless.classes import installer
from worthless.enums import (
GameVariant,
from worthless.game.enums import (
Variant,
VoicepackArchiveType,
GameUpdateStatus,
GameInstallStatus,
UpdateStatus,
InstallStatus,
VoicepackArchiveLanguage,
)
from worthless.hdiffpatch import HDiffPatch
from worthless.launcherconfig import LauncherConfig
from worthless.game.hdiffpatch import HDiffPatch
from worthless.game.launcherconfig import LauncherConfig
_logger = logging.getLogger("worthless.GameManager")
_logger = logging.getLogger("worthless.Game")
def voicepack_lang_translate(lang: str, base_language="game") -> str | None:
@ -101,19 +100,23 @@ def get_voicepack_archive_type(archive: PathLike) -> VoicepackArchiveType:
return VoicepackArchiveType.FULL
class GameManager:
class Game:
"""
Manages game & voicepacks installation.
This class handles installing and updating game & voicepacks installation.
It also includes some download function for your convenience, although you are advised
to use your own download function (e.g. integrate with GUI, etc.)
Args:
game_dir: Game directory
variant: Game variant
cache_dir: Cache directory, if not specified it'll automatically detect your
system cache directory and use that.
"""
def __init__(
self,
game_dir: PathLike = None,
variant: GameVariant = None,
variant: Variant = None,
cache_dir: PathLike = None,
):
if not game_dir:
@ -130,7 +133,6 @@ class GameManager:
"Installer"
)
Path(self._cache_path).mkdir(parents=True, exist_ok=True)
self._download_chunk = 8192
self._variant = variant
self._version = None
self._launcher = Launcher(variant=variant)
@ -141,29 +143,6 @@ class GameManager:
)
self._game_version_re = re.compile(r"([1-9]+\.[0-9]+\.[0-9]+)_\d+_\d+")
async def _download_file(
self, file_url: str, file_name: str, file_len: int = None, overwrite=False
) -> Path:
"""
Download file name to temporary directory.
This function is a wrapper for helper.download_file
Args:
file_url: The file url to download
file_name: The file name to download into
Returns:
A Path object containing downloaded file
"""
return await helper.download_file(
file_url,
file_name,
self._cache_path,
file_len=file_len,
overwrite=overwrite,
chunks=self._download_chunk,
)
def _read_version_from_config(self) -> str | None:
"""
Reads the version from config.ini
@ -218,7 +197,7 @@ class GameManager:
return ""
with file_to_calculate.open("rb") as f:
file_hash = hashlib.md5()
while chunk := f.read(self._download_chunk):
while chunk := f.read(8196):
file_hash.update(chunk)
return file_hash.hexdigest()
@ -251,15 +230,13 @@ class GameManager:
if file is not None:
failed_files.append(file)
return None if not failed_files else failed_files
return failed_files
@property
def version(self) -> str | None:
"""
Gets the current game version
This property is a shorthand for get_current_game_version()
Returns:
Version as string "x.x.x" or None if not found.
"""
@ -268,7 +245,7 @@ class GameManager:
return self._version
@property
def variant(self) -> GameVariant:
def variant(self) -> Variant:
"""
Gets the current game variant
@ -281,14 +258,30 @@ class GameManager:
self._version = self.get_current_game_variant()
return self._variant
def set_download_chunk(self, chunk: int):
"""
Sets the download chunk for the internal download function
@property
def cache_path(self):
return self._cache_path
Args:
chunk: The chunk to set into
@staticmethod
def voicepack_lang_translate(lang: str, base_language="game") -> str | None:
"""
self._download_chunk = chunk
This function is an alias to worthless.game.voicepack_lang_translate
"""
return voicepack_lang_translate(lang=lang, base_language=base_language)
@staticmethod
def get_voicepack_archive_language(archive: PathLike) -> VoicepackArchiveLanguage:
"""
This function is an alias to worthless.game.get_voicepack_archive_language
"""
return get_voicepack_archive_language(archive=archive)
@staticmethod
def get_voicepack_archive_type(archive: PathLike) -> VoicepackArchiveType:
"""
This function is an alias to worthless.game.get_voicepack_archive_type
"""
return get_voicepack_archive_type(archive=archive)
def get_game_data_name(self) -> str:
"""
@ -298,11 +291,11 @@ class GameManager:
A string containing game data path (e.g. GenshinImpact_Data)
"""
match self.variant:
case GameVariant.INTERNATIONAL:
case Variant.INTERNATIONAL:
return "GenshinImpact_Data"
case GameVariant.CHINESE:
case Variant.CHINESE:
return "YuanShen_Data"
case GameVariant.BILIBILI:
case Variant.BILIBILI:
return "YuanShen_Data"
def get_game_data_path(self) -> Path:
@ -314,7 +307,7 @@ class GameManager:
"""
return self._game_dir.joinpath(self.get_game_data_name())
def get_archive_game_version(self, game_archive: str | Path) -> str | None:
def get_archive_game_version(self, game_archive: PathLike) -> str | None:
"""
Gets the game version in the archive
@ -329,7 +322,7 @@ class GameManager:
f.read(self.get_game_data_name() + "globalgamemanagers")
)
def get_current_game_variant(self) -> GameVariant:
def get_current_game_variant(self) -> Variant:
"""
Gets the current game variant
@ -337,14 +330,14 @@ class GameManager:
GameVariant
"""
if self._game_dir.joinpath("GenshinImpact.exe").is_file():
return GameVariant.INTERNATIONAL
return Variant.INTERNATIONAL
if not self._game_dir.joinpath("YuanShen.exe").is_file():
raise FileNotFoundError("Game installation not found.")
# We can't depend on get_game_data_name() because it depends on self._variant
# which depends on this function.
if self._game_dir.joinpath("YuanShen_Data/Plugins/PCGameSDK.dll").is_file():
return GameVariant.BILIBILI
return GameVariant.CHINESE
return Variant.BILIBILI
return Variant.CHINESE
def get_current_game_version(self) -> str | None:
"""
@ -377,10 +370,10 @@ class GameManager:
voicepacks.append(file.name)
return voicepacks
async def update_game(self, game_archive: str | Path, callback=None):
async def update_game(self, game_archive: PathLike, callback=None):
if not self.get_game_data_path().exists():
raise FileNotFoundError(f"Game not found in {self._game_dir}")
if isinstance(game_archive, str | Path):
if not isinstance(game_archive, Path):
game_archive = Path(game_archive).resolve()
if not game_archive.exists():
raise FileNotFoundError(f"Update archive {game_archive} not found")
@ -391,14 +384,14 @@ class GameManager:
callback(status=status, file=f)
if not self._hdiffpatch.get_hpatchz_executable():
call(GameUpdateStatus.DOWNLOADING_PATCHER)
call(UpdateStatus.DOWNLOADING_PATCHER)
await self._hdiffpatch.download_latest_release()
archive = zipfile.ZipFile(game_archive, "r")
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-game)
call(GameUpdateStatus.PREPARING_PATCH)
call(UpdateStatus.PREPARING_PATCH)
for file in ["deletefiles.txt", "hdifffiles.txt"]:
try:
files.remove(file)
@ -416,7 +409,7 @@ class GameManager:
hdifffiles.append(json.loads(x)["remoteName"])
patch_jobs = []
for file in hdifffiles:
call(GameUpdateStatus.PREPARING_PATCH, f=file)
call(UpdateStatus.PREPARING_PATCH, f=file)
current_game_file = self._game_dir.joinpath(file)
if not current_game_file.exists():
# Not patching since we don't have the file
@ -425,7 +418,7 @@ class GameManager:
patch_file = str(file) + ".hdiff"
async def extract_and_patch(old_file, diff_file):
call(GameUpdateStatus.PATCHING, f=old_file)
call(UpdateStatus.PATCHING, f=old_file)
diff_path = self._cache_path.joinpath(diff_file)
if diff_path.is_file():
diff_path.unlink(missing_ok=True)
@ -447,23 +440,23 @@ class GameManager:
patch_jobs.append(extract_and_patch(current_game_file, patch_file))
await asyncio.gather(*patch_jobs)
call(GameUpdateStatus.PREPARING_REMOVE_UNUSED)
call(UpdateStatus.PREPARING_REMOVE_UNUSED)
deletefiles = archive.read("deletefiles.txt").decode().split("\n")
for file in deletefiles:
current_game_file = Path(self._game_dir.joinpath(file))
if not current_game_file.exists():
continue
if current_game_file.is_file():
call(GameUpdateStatus.REMOVING_UNUSED, f=file)
call(UpdateStatus.REMOVING_UNUSED, f=file)
current_game_file.unlink(missing_ok=True)
call(GameUpdateStatus.EXTRACTING)
call(UpdateStatus.EXTRACTING)
await asyncio.to_thread(archive.extractall, self._game_dir, members=files)
archive.close()
# Update game version on local variable.
self._version = self.get_current_game_version()
self.set_version_config()
call(GameUpdateStatus.COMPLETED)
call(UpdateStatus.COMPLETED)
def set_version_config(self, version: str = None):
"""
@ -477,20 +470,8 @@ class GameManager:
self._game_config.set_game_version(version)
self._game_config.save()
async def download_full_game(self, pre_download=False):
game = await self._get_game(pre_download)
archive_name = game.latest.path.split("/")[-1]
await self._download_file(game.latest.path, archive_name, game.latest.size)
async def download_full_voicepack(self, language: str, pre_download=False):
game = await self._get_game(pre_download)
translated_lang = voicepack_lang_translate(language)
for vo in game.latest.voice_packs:
if vo.language == translated_lang:
await self._download_file(vo.path, vo.get_name(), vo.size)
def _extract_game_file(self, archive: PathLike):
if isinstance(archive, str):
if not isinstance(archive, Path):
archive = Path(archive).resolve()
if not archive.exists():
raise FileNotFoundError(f"'{archive}' not found")
@ -527,11 +508,11 @@ class GameManager:
if self.get_game_data_path().exists():
if not force_reinstall:
raise ValueError(f"Game is already installed in {self._game_dir}")
call(GameInstallStatus.REMOVING_OLD_GAME)
call(InstallStatus.REMOVING_OLD_GAME)
self.delete_game()
self._game_dir.mkdir(parents=True, exist_ok=True)
call(GameInstallStatus.EXTRACTING)
call(InstallStatus.EXTRACTING)
self._extract_game_file(game_archive)
self._version = self.get_current_game_version()
self.set_version_config()
@ -566,7 +547,7 @@ class GameManager:
if v.version == from_version:
return v
async def verify_game(self, pkg_version: str | Path = None, ignore_mismatch=None):
async def verify_game(self, pkg_version: PathLike = None, ignore_mismatch=None):
"""
Verifies the current game installation
@ -597,30 +578,3 @@ class GameManager:
Clears the GameManager cache (e.g. downloaded game files)
"""
shutil.rmtree(self._cache_path, ignore_errors=True)
async def download_game_update(self, from_version: str = None, pre_download=False):
from_version = from_version if from_version else self.version
game = await self._get_game(pre_download=pre_download)
if self.version == game.latest.version:
raise ValueError("Game is already up to date.")
diff_archive = await self.get_game_diff_archive(from_version, pre_download)
if diff_archive is None:
raise ValueError(
"Game diff archive is not available for this version, please reinstall."
)
await self._download_file(
diff_archive.path, diff_archive.name, diff_archive.size
)
async def download_voicepack_update(
self, language: str, from_version: str = None, pre_download=False
):
from_version = from_version if from_version else self.version
diff_archive = await self.get_voicepack_diff_archive(
language, from_version, pre_download
)
if diff_archive is None:
raise ValueError("Voiceover diff archive is not available for this version")
await self._download_file(
diff_archive.path, diff_archive.name, diff_archive.size
)

2
worthless/hdiffpatch.py → worthless/game/hdiffpatch.py
View File

@ -12,7 +12,7 @@ class HDiffPatch:
"""
Contains legacy HDiffPatch support for worthless
You should not use this class directly, since it's automatically used by GameManager
You should not use this class directly, since it's automatically used by Game when needed
"""
def __init__(self, git_url=None, data_dir=None):

108
worthless/game/helper.py Normal file
View File

@ -0,0 +1,108 @@
from os import PathLike
from pathlib import Path
from worthless import helper
from worthless.game import Game, Variant
class Helper(Game):
"""
Quick and dirty extra functions for Game
Since this is quick and dirty, you are recommended to write your own method instead.
Args:
game: A worthless.game.Game instance, if not specified it'll initialize a new one
using super().__init__ and use the following arguments for that instance creation
game_dir: Game directory
variant: Game variant
cache_dir: Cache directory, if not specified it'll automatically detect your
system cache directory and use that.
"""
def __init__(
self,
game: Game = None,
game_dir: PathLike = None,
variant: Variant = None,
cache_dir: PathLike = None,
):
if not game:
super().__init__(game_dir, variant, cache_dir)
game = self
self._download_chunk = 8192
self._game = game
async def _download_file(
self, file_url: str, file_name: str, file_len: int = None, overwrite=False
) -> Path:
"""
Download file name to temporary directory.
This function is a wrapper for helper.download_file
Args:
file_url: The file url to download
file_name: The file name to download into
Returns:
A Path object containing downloaded file
"""
return await helper.download_file(
file_url,
file_name,
self._game.cache_path,
file_len=file_len,
overwrite=overwrite,
chunks=self._download_chunk,
)
def set_download_chunk(self, chunk: int):
"""
Sets the download chunk for the internal download function
Args:
chunk: The chunk to set into
"""
self._download_chunk = chunk
async def download_full_game(self, pre_download=False) -> Path:
game = await self._game._get_game(pre_download)
archive_name = game.latest.path.split("/")[-1]
return await self._download_file(
game.latest.path, archive_name, game.latest.size
)
async def download_full_voicepack(self, language: str, pre_download=False) -> Path:
game = await self._game._get_game(pre_download)
translated_lang = self._game.voicepack_lang_translate(language)
for vo in game.latest.voice_packs:
if vo.language == translated_lang:
return await self._download_file(vo.path, vo.get_name(), vo.size)
async def download_game_update(
self, from_version: str = None, pre_download=False
) -> Path:
from_version = from_version if from_version else self.version
game = await self._game._get_game(pre_download=pre_download)
if self._game.version == game.latest.version:
raise ValueError("Game is already up to date.")
diff_archive = await self._game.get_game_diff_archive(from_version, pre_download)
if diff_archive is None:
raise ValueError(
"Game diff archive is not available for this version, please reinstall."
)
return await self._download_file(
diff_archive.path, diff_archive.name, diff_archive.size
)
async def download_voicepack_update(
self, language: str, from_version: str = None, pre_download=False
) -> Path:
from_version = from_version if from_version else self.version
diff_archive = await self._game.get_voicepack_diff_archive(
language, from_version, pre_download
)
if diff_archive is None:
raise ValueError("Voiceover diff archive is not available for this version")
return await self._download_file(
diff_archive.path, diff_archive.name, diff_archive.size
)

24
worthless/launcherconfig.py → worthless/game/launcherconfig.py
View File

@ -1,16 +1,18 @@
from configparser import ConfigParser
from os import PathLike
from pathlib import Path
from worthless.enums import GameVariant
from worthless.game.enums import Variant
class LauncherConfig:
"""
Provides config.ini for official launcher compatibility
You should not use this class directly, since it's automatically used by Game when needed
"""
@staticmethod
def create_config(game_version, variant: GameVariant):
def create_config(game_version, variant: Variant):
"""
Creates `config.ini`
https://notabug.org/Krock/dawn/src/master/updater/update_gi.sh#L212
@ -19,15 +21,15 @@ class LauncherConfig:
channel = 1
cps = "mihoyo"
match variant:
case GameVariant.INTERNATIONAL:
case Variant.INTERNATIONAL:
channel = 1
sub_channel = 0
cps = "mihoyo"
case GameVariant.CHINESE:
case Variant.CHINESE:
channel = 1
sub_channel = 1
cps = "mihoyo"
case GameVariant.BILIBILI:
case Variant.BILIBILI:
channel = 14
sub_channel = 0
cps = "bilibili"
@ -41,10 +43,10 @@ class LauncherConfig:
return config
def __init__(
self, config_path: PathLike, game_version=None, variant: GameVariant = None
self, config_path: PathLike, game_version=None, variant: Variant = None
):
if not variant:
variant = GameVariant.INTERNATIONAL
variant = Variant.INTERNATIONAL
if not isinstance(config_path, Path):
config_path = Path(config_path)
if not game_version:
@ -59,14 +61,14 @@ class LauncherConfig:
def set_game_version(self, game_version):
self._config.set("General", "game_version", game_version)
def set_variant(self, variant: GameVariant):
def set_variant(self, variant: Variant):
sub_channel = 0
match variant:
case GameVariant.INTERNATIONAL:
case Variant.INTERNATIONAL:
sub_channel = 0
case GameVariant.CHINESE:
case Variant.CHINESE:
sub_channel = 1
case GameVariant.BILIBILI:
case Variant.BILIBILI:
sub_channel = 0
self._config.set("General", "sub_channel", str(sub_channel))

12
worthless/launcher.py
View File

@ -2,7 +2,7 @@ import aiohttp
import locale
from worthless import constants
from worthless.classes import launcher, installer
from worthless.enums import GameVariant
from worthless.game.enums import Variant
def _get_system_language() -> str:
@ -28,14 +28,14 @@ class Launcher:
def __init__(
self,
language: str = None,
variant: GameVariant = None,
variant: Variant = None,
):
"""Initialize the launcher API"""
if not variant:
variant = GameVariant.INTERNATIONAL
variant = Variant.INTERNATIONAL
self._variant = variant
match variant:
case GameVariant.INTERNATIONAL:
case Variant.INTERNATIONAL:
self._api = constants.LAUNCHER_API_URL_OS
self._params = {
"key": "gcStgarh",
@ -46,7 +46,7 @@ class Launcher:
if language
else _get_system_language()
)
case GameVariant.CHINESE:
case Variant.CHINESE:
self._api = constants.LAUNCHER_API_URL_CN
self._params = {
"key": "eYd89JmJ",
@ -56,7 +56,7 @@ class Launcher:
self._lang = (
"zh-cn" # Use chinese language because this is chinese version
)
case GameVariant.BILIBILI:
case Variant.BILIBILI:
self._api = constants.LAUNCHER_API_URL_CN
self._params = {
"key": "KAtdSsoQ",

10
worthless/linux.py
View File

@ -1,6 +1,6 @@
import asyncio
from os import PathLike
from pathlib import Path
from aiopath import AsyncPath
class LinuxUtils:
@ -12,7 +12,7 @@ class LinuxUtils:
@staticmethod
async def _exec_command(args):
"""Execute a command using pkexec (friendly gui)"""
if not await AsyncPath("/usr/bin/pkexec").exists():
if not Path("/usr/bin/pkexec").is_file():
raise FileNotFoundError("pkexec not found.")
rsp = await asyncio.create_subprocess_shell(args)
await rsp.wait()
@ -24,13 +24,13 @@ class LinuxUtils:
return rsp
async def write_text_to_file(self, text, file_path: str | Path | AsyncPath):
async def write_text_to_file(self, text, file_path: PathLike):
"""Write text to a file using pkexec (friendly gui)"""
if isinstance(file_path, Path | AsyncPath):
if isinstance(file_path, Path):
file_path = str(file_path)
await self._exec_command('echo -e "{}" | pkexec tee {}'.format(text, file_path))
async def append_text_to_file(self, text, file_path: str | Path | AsyncPath):
async def append_text_to_file(self, text, file_path: PathLike):
"""Append text to a file using pkexec (friendly gui)"""
if isinstance(file_path, Path):
file_path = str(file_path)

6
worthless/patcher.py
View File

@ -1,16 +1,14 @@
import os
import platform
import tarfile
from pathlib import Path
import shutil
import aiohttp
import asyncio
from aiopath import AsyncPath
from pathlib import Path
from worthless import constants
from worthless.launcher import Launcher
from worthless.gamemanager import GameManager as Installer
from worthless.game import Game as Installer
match platform.system():
case "Linux":