tretrauit
a5659f7ff3
chore: rename gui.py to cli.py fix: internal downloader can resume download now. feat: add verify_game, verify_from_pkg_version, clear_cache to installer.py. feat: add clear_cache to patcher.py. fix: linux now check for pkexec before executing it. fix: add get_name to voicepack.py, latest.py, diff.py to get name from path (since the developer didn't set a name to these files in the sdk url) chore: remove deprecation message in read_version_from_config in installer.py misc: use chunk from self._download_chunk instead of being hardcoded to 8192. fix: is_telemetry_blocked will only wait 15s for a connection. chore: move appdirs to constants.py This commit refactor almost all functions to be compatible with asyncio, also restructured CLI to use asyncio.run on main function instead of executing it randomly. Also prioritize the use of asyncio.gather, sometimes making tasks faster
531 lines
22 KiB
Python
531 lines
22 KiB
Python
import asyncio
|
|
import re
|
|
import shutil
|
|
import platform
|
|
import aiohttp
|
|
import zipfile
|
|
import json
|
|
import hashlib
|
|
from pathlib import Path
|
|
from configparser import ConfigParser
|
|
from aiopath import AsyncPath
|
|
from worthless import constants
|
|
from worthless.launcher import Launcher
|
|
from worthless.launcherconfig import LauncherConfig
|
|
|
|
|
|
async def _download_file(file_url: str, file_name: str, file_path: Path | str, file_len: int = None, overwrite=False,
|
|
chunks=8192):
|
|
"""
|
|
Download file name to temporary directory,
|
|
:param file_url:
|
|
:param file_name:
|
|
:return:
|
|
"""
|
|
headers = {}
|
|
file_path = AsyncPath(file_path).joinpath(file_name)
|
|
if overwrite:
|
|
await file_path.unlink(missing_ok=True)
|
|
if await file_path.exists():
|
|
cur_len = (await file_path.stat()).st_size
|
|
headers |= {
|
|
"Range": f"bytes={cur_len}-{file_len if file_len else ''}"
|
|
}
|
|
else:
|
|
await file_path.touch()
|
|
async with aiohttp.ClientSession() as session:
|
|
rsp = await session.get(file_url, headers=headers, timeout=None)
|
|
if rsp.status == 416:
|
|
return
|
|
rsp.raise_for_status()
|
|
while True:
|
|
chunk = await rsp.content.read(chunks)
|
|
if not chunk:
|
|
break
|
|
async with file_path.open("ab") as f:
|
|
await f.write(chunk)
|
|
|
|
|
|
class HDiffPatch:
|
|
def __init__(self, git_url=None, data_dir=None):
|
|
if not git_url:
|
|
git_url = constants.HDIFFPATCH_GIT_URL
|
|
self._git_url = git_url
|
|
if not data_dir:
|
|
self._appdirs = constants.APPDIRS
|
|
self.temp_path = Path(self._appdirs.user_cache_dir).joinpath("HDiffPatch")
|
|
self.data_path = Path(self._appdirs.user_data_dir).joinpath("Tools/HDiffPatch")
|
|
else:
|
|
if not isinstance(data_dir, Path):
|
|
data_dir = Path(data_dir)
|
|
self.data_path = Path(data_dir).joinpath("Tools/HDiffPatch")
|
|
self.temp_path = data_dir.joinpath("Temp/HDiffPatch")
|
|
self.temp_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
@staticmethod
|
|
def _get_platform_arch():
|
|
match platform.system():
|
|
case "Windows":
|
|
match platform.architecture()[0]:
|
|
case "32bit":
|
|
return "windows32"
|
|
case "64bit":
|
|
return "windows64"
|
|
case "Linux":
|
|
match platform.architecture()[0]:
|
|
case "32bit":
|
|
return "linux32"
|
|
case "64bit":
|
|
return "linux64"
|
|
case "Darwin":
|
|
return "macos"
|
|
|
|
# Rip BSD they need to use Linux compatibility layer to run this (or use Wine if they prefer that)
|
|
raise RuntimeError("Unsupported platform")
|
|
|
|
def _get_hdiffpatch_exec(self, exec_name):
|
|
if shutil.which(exec_name):
|
|
return exec_name
|
|
if not self.data_path.exists():
|
|
return None
|
|
if not any(self.data_path.iterdir()):
|
|
return None
|
|
platform_arch_path = self.data_path.joinpath(self._get_platform_arch())
|
|
if platform_arch_path.joinpath(exec_name).exists():
|
|
return str(platform_arch_path.joinpath(exec_name))
|
|
return None
|
|
|
|
def get_hpatchz_executable(self):
|
|
hpatchz_name = "hpatchz" + (".exe" if platform.system() == "Windows" else "")
|
|
return self._get_hdiffpatch_exec(hpatchz_name)
|
|
|
|
async def patch_file(self, in_file, out_file, patch_file, error=False, wait=False):
|
|
hpatchz = self.get_hpatchz_executable()
|
|
if not hpatchz:
|
|
raise RuntimeError("hpatchz executable not found")
|
|
proc = await asyncio.create_subprocess_exec(hpatchz, "-f", in_file, patch_file, out_file)
|
|
if not wait:
|
|
return proc
|
|
await proc.wait()
|
|
if error and proc.returncode != 0:
|
|
raise RuntimeError(f"Patching failed, return code is {proc.returncode}")
|
|
return proc
|
|
|
|
def get_hdiffz_executable(self):
|
|
hdiffz_name = "hdiffz" + (".exe" if platform.system() == "Windows" else "")
|
|
return self._get_hdiffpatch_exec(hdiffz_name)
|
|
|
|
async def _get_latest_release_info(self):
|
|
async with aiohttp.ClientSession() as session:
|
|
split = self._git_url.split("/")
|
|
repo = split[-1]
|
|
owner = split[-2]
|
|
rsp = await session.get("https://api.github.com/repos/{}/{}/releases/latest".format(owner, repo),
|
|
params={"Headers": "Accept: application/vnd.github.v3+json"})
|
|
rsp.raise_for_status()
|
|
for asset in (await rsp.json())["assets"]:
|
|
if asset["name"].endswith(".zip") and "linux" not in asset["name"] and "windows" not in asset["name"] \
|
|
and "macos" not in asset["name"] and "android" not in asset["name"]:
|
|
return asset
|
|
|
|
async def get_latest_release_url(self):
|
|
asset = await self._get_latest_release_info()
|
|
return asset["browser_download_url"]
|
|
|
|
async def get_latest_release_name(self):
|
|
asset = await self._get_latest_release_info()
|
|
return asset["name"]
|
|
|
|
async def download_latest_release(self, extract=True):
|
|
url = await self.get_latest_release_url()
|
|
name = await self.get_latest_release_name()
|
|
if not url:
|
|
raise RuntimeError("Unable to find latest release")
|
|
await _download_file(url, name, self.temp_path, overwrite=True)
|
|
if not extract:
|
|
return
|
|
with zipfile.ZipFile(self.temp_path.joinpath(name), 'r') as f:
|
|
await asyncio.to_thread(f.extractall, path=self.data_path)
|
|
|
|
|
|
class Installer:
|
|
def __init__(self, gamedir: str | Path | AsyncPath = AsyncPath.cwd(),
|
|
overseas: bool = True, data_dir: str | Path | AsyncPath = None):
|
|
if isinstance(gamedir, str | Path):
|
|
gamedir = AsyncPath(gamedir)
|
|
self._gamedir = gamedir
|
|
if not data_dir:
|
|
self._appdirs = constants.APPDIRS
|
|
self.temp_path = AsyncPath(self._appdirs.user_cache_dir).joinpath("Installer")
|
|
else:
|
|
if isinstance(data_dir, str | AsyncPath):
|
|
data_dir = AsyncPath(data_dir)
|
|
self.temp_path = data_dir.joinpath("Temp/Installer/")
|
|
Path(self.temp_path).mkdir(parents=True, exist_ok=True)
|
|
config_file = self._gamedir.joinpath("config.ini")
|
|
self._config_file = config_file
|
|
self._download_chunk = 8192
|
|
self._overseas = overseas
|
|
self._version = None
|
|
self._launcher = Launcher(self._gamedir, overseas=self._overseas)
|
|
self._hdiffpatch = HDiffPatch(data_dir=data_dir)
|
|
self._config = LauncherConfig(self._config_file, self._version)
|
|
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):
|
|
"""
|
|
Download file name to temporary directory,
|
|
:param file_url:
|
|
:param file_name:
|
|
:return:
|
|
"""
|
|
await _download_file(file_url, file_name, self.temp_path, file_len=file_len, overwrite=overwrite,
|
|
chunks=self._download_chunk)
|
|
|
|
async def read_version_from_config(self):
|
|
if not await self._config_file.exists():
|
|
raise FileNotFoundError(f"Config file {self._config_file} not found")
|
|
cfg = ConfigParser()
|
|
await asyncio.to_thread(cfg.read, str(self._config_file))
|
|
return cfg.get("General", "game_version")
|
|
|
|
async def read_version_from_game_file(self, globalgamemanagers: AsyncPath | Path | bytes) -> str:
|
|
"""
|
|
Reads the version from the globalgamemanagers file. (Data/globalgamemanagers)
|
|
|
|
Uses `An Anime Game Launcher` method to read the version:
|
|
https://gitlab.com/KRypt0n_/an-anime-game-launcher/-/blob/main/src/ts/Game.ts#L26
|
|
|
|
:return: Game version (ex 1.0.0)
|
|
"""
|
|
if isinstance(globalgamemanagers, Path | AsyncPath):
|
|
globalgamemanagers = AsyncPath(globalgamemanagers)
|
|
data = await globalgamemanagers.read_text("ascii", errors="ignore")
|
|
else:
|
|
data = globalgamemanagers.decode("ascii", errors="ignore")
|
|
result = self._game_version_re.search(data)
|
|
if not result:
|
|
raise ValueError("Could not find version in game file")
|
|
return result.group(1)
|
|
|
|
@staticmethod
|
|
def voiceover_lang_translate(lang: str, base_language="game") -> str:
|
|
"""
|
|
Translates the voiceover language to the language code used by the game.
|
|
:param lang: Language to translate
|
|
:param base_language: Base language type (game/locale/both)
|
|
:return: Language code
|
|
"""
|
|
if base_language == "game" or base_language == "both":
|
|
match lang.lower():
|
|
case "english(us)":
|
|
return "en-us"
|
|
case "japanese":
|
|
return "ja-jp"
|
|
case "chinese":
|
|
return "zh-cn"
|
|
case "korean":
|
|
return "ko-kr"
|
|
if base_language == "locale" or base_language == "both":
|
|
match lang.lower().replace("_", "-"):
|
|
case "en-us":
|
|
return "English(US)"
|
|
case "ja-jp":
|
|
return "Japanese"
|
|
case "zh-cn":
|
|
return "Chinese"
|
|
case "ko-kr":
|
|
return "Korean"
|
|
# If nothing else matches
|
|
return lang
|
|
|
|
@staticmethod
|
|
async def get_voiceover_archive_language(voiceover_archive: str | Path | AsyncPath) -> str:
|
|
if isinstance(voiceover_archive, str | Path):
|
|
voiceover_archive = Path(voiceover_archive).resolve()
|
|
if not voiceover_archive.exists():
|
|
raise FileNotFoundError(f"Voiceover archive {voiceover_archive} not found")
|
|
with zipfile.ZipFile(voiceover_archive, 'r') as f:
|
|
for file in zipfile.Path(f).iterdir():
|
|
if file.name.endswith("_pkg_version"):
|
|
return file.name.split("_")[1]
|
|
|
|
@staticmethod
|
|
async def get_voiceover_archive_type(voiceover_archive: str | Path) -> bool:
|
|
"""
|
|
Gets voiceover archive type.
|
|
:param voiceover_archive:
|
|
:return: True if this is a full archive, else False.
|
|
"""
|
|
vo_lang = Installer.get_voiceover_archive_language(voiceover_archive)
|
|
with zipfile.ZipFile(voiceover_archive, 'r') as f:
|
|
archive_path = zipfile.Path(f)
|
|
files = (await asyncio.to_thread(f.read, "Audio_{}_pkg_version".format(vo_lang))).decode().split("\n")
|
|
for file in files:
|
|
if file.strip() and not archive_path.joinpath(json.loads(file)["remoteName"]).exists():
|
|
return False
|
|
return True
|
|
|
|
def set_download_chunk(self, chunk: int):
|
|
self._download_chunk = chunk
|
|
|
|
def get_game_data_name(self):
|
|
if self._overseas:
|
|
return "GenshinImpact_Data/"
|
|
else:
|
|
return "YuanShen_Data/"
|
|
|
|
def get_game_data_path(self) -> AsyncPath:
|
|
return self._gamedir.joinpath(self.get_game_data_name())
|
|
|
|
async def get_game_archive_version(self, game_archive: str | Path):
|
|
if not game_archive.exists():
|
|
raise FileNotFoundError(f"Game archive {game_archive} not found")
|
|
with zipfile.ZipFile(game_archive, 'r') as f:
|
|
return await self.read_version_from_game_file(
|
|
await asyncio.to_thread(f.read, self.get_game_data_name() + "globalgamemanagers")
|
|
)
|
|
|
|
async def get_game_version(self) -> str | None:
|
|
globalgamemanagers = self.get_game_data_path().joinpath("./globalgamemanagers")
|
|
if not await globalgamemanagers.exists():
|
|
try:
|
|
return await self.read_version_from_config()
|
|
except FileNotFoundError:
|
|
return
|
|
return await self.read_version_from_game_file(globalgamemanagers)
|
|
|
|
async def get_installed_voiceovers(self) -> list[str]:
|
|
"""
|
|
Returns a list of installed voiceovers.
|
|
|
|
:return: List of installed voiceovers
|
|
"""
|
|
voiceovers = []
|
|
async for file in self.get_game_data_path()\
|
|
.joinpath("StreamingAssets/Audio/GeneratedSoundBanks/Windows").iterdir():
|
|
if await file.is_dir():
|
|
voiceovers.append(file.name)
|
|
return voiceovers
|
|
|
|
async def update_game(self, game_archive: str | Path | AsyncPath):
|
|
if not await self.get_game_data_path().exists():
|
|
raise FileNotFoundError(f"Game not found in {self._gamedir}")
|
|
if isinstance(game_archive, str | Path):
|
|
game_archive = Path(game_archive).resolve()
|
|
if not game_archive.exists():
|
|
raise FileNotFoundError(f"Update archive {game_archive} not found")
|
|
|
|
archive = zipfile.ZipFile(game_archive, 'r')
|
|
|
|
if not self._hdiffpatch.get_hpatchz_executable():
|
|
await self._hdiffpatch.download_latest_release()
|
|
|
|
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 ingame)
|
|
for file in ["deletefiles.txt", "hdifffiles.txt"]:
|
|
try:
|
|
files.remove(file)
|
|
except ValueError:
|
|
pass
|
|
|
|
# hdiffpatch implementation
|
|
hdifffiles = []
|
|
for x in (await asyncio.to_thread(archive.read, "hdifffiles.txt")).decode().split("\n"):
|
|
if x:
|
|
hdifffiles.append(json.loads(x)["remoteName"])
|
|
patch_jobs = []
|
|
for file in hdifffiles:
|
|
current_game_file = self._gamedir.joinpath(file)
|
|
if not await current_game_file.exists():
|
|
# Not patching since we don't have the file
|
|
continue
|
|
|
|
patch_file = str(file) + ".hdiff"
|
|
|
|
async def extract_and_patch(old_file, diff_file):
|
|
await asyncio.to_thread(archive.extract, diff_file, self.temp_path)
|
|
patch_path = self.temp_path.joinpath(diff_file)
|
|
old_suffix = old_file.suffix
|
|
old_file = old_file.rename(old_file.with_suffix(".bak"))
|
|
proc = await self._hdiffpatch.patch_file(old_file, old_file.with_suffix(old_suffix),
|
|
patch_path, wait=True)
|
|
patch_path.unlink()
|
|
if proc.returncode != 0:
|
|
# Let the game download the file.
|
|
old_file.rename(old_file.with_suffix(old_suffix))
|
|
return
|
|
old_file.unlink()
|
|
|
|
files.remove(patch_file)
|
|
patch_jobs.append(extract_and_patch(current_game_file, patch_file))
|
|
|
|
await asyncio.gather(*patch_jobs)
|
|
|
|
deletefiles = archive.read("deletefiles.txt").decode().split("\n")
|
|
for file in deletefiles:
|
|
current_game_file = self._gamedir.joinpath(file)
|
|
if not await current_game_file.exists():
|
|
continue
|
|
if current_game_file.is_file():
|
|
current_game_file.unlink(missing_ok=True)
|
|
|
|
await asyncio.to_thread(archive.extractall, self._gamedir, members=files)
|
|
archive.close()
|
|
# Update game version on local variable.
|
|
self._version = await self.get_game_version()
|
|
self.set_version_config()
|
|
|
|
def set_version_config(self, version: str = None):
|
|
if not version:
|
|
version = self._version
|
|
self._config.set_game_version(version)
|
|
self._config.save()
|
|
|
|
async def download_full_game(self):
|
|
resource = await self._launcher.get_resource_info()
|
|
if resource is None:
|
|
raise RuntimeError("Failed to fetch game resource info.")
|
|
archive_name = resource.game.latest.path.split("/")[-1]
|
|
await self._download_file(resource.game.latest.path, archive_name, resource.game.latest.size)
|
|
|
|
async def download_full_voiceover(self, language: str):
|
|
archive = await self._launcher.get_resource_info()
|
|
if archive is None:
|
|
raise RuntimeError("Failed to fetch game resource info.")
|
|
translated_lang = self.voiceover_lang_translate(language)
|
|
for vo in archive.game.latest.voice_packs:
|
|
if vo.language == translated_lang:
|
|
await self._download_file(vo.path, vo.get_name(), vo.size)
|
|
|
|
async def uninstall_game(self):
|
|
await asyncio.to_thread(shutil.rmtree, self._gamedir, ignore_errors=True)
|
|
|
|
async def _extract_game_file(self, archive: str | Path | AsyncPath):
|
|
if isinstance(archive, str | AsyncPath):
|
|
archive = Path(archive).resolve()
|
|
if not archive.exists():
|
|
raise FileNotFoundError(f"'{archive}' not found")
|
|
with zipfile.ZipFile(archive, 'r') as f:
|
|
await asyncio.to_thread(f.extractall, path=self._gamedir)
|
|
|
|
async def apply_voiceover(self, voiceover_archive: str | Path):
|
|
# Since Voiceover packages are unclear about diff package or full package
|
|
# we will try to extract the voiceover package and apply it to the game
|
|
# making this function universal for both cases
|
|
if not await self.get_game_data_path().exists():
|
|
raise FileNotFoundError(f"Game not found in {self._gamedir}")
|
|
await self._extract_game_file(voiceover_archive)
|
|
|
|
async def install_game(self, game_archive: str | Path | AsyncPath, force_reinstall: bool = False):
|
|
"""Installs the game to the current directory
|
|
|
|
If `force_reinstall` is True, the game will be uninstalled then reinstalled.
|
|
"""
|
|
if await self.get_game_data_path().exists():
|
|
if not force_reinstall:
|
|
raise ValueError(f"Game is already installed in {self._gamedir}")
|
|
await self.uninstall_game()
|
|
|
|
await self._gamedir.mkdir(parents=True, exist_ok=True)
|
|
await self._extract_game_file(game_archive)
|
|
self._version = await self.get_game_version()
|
|
self.set_version_config()
|
|
|
|
async def _get_game_resource(self, from_version: str = None):
|
|
if not from_version:
|
|
if self._version:
|
|
from_version = self._version
|
|
else:
|
|
from_version = self._version = await self.get_game_version()
|
|
if not from_version:
|
|
raise ValueError("No game version found")
|
|
game_resource = await self._launcher.get_resource_info()
|
|
if not game_resource:
|
|
raise ValueError("Could not fetch game resource")
|
|
return game_resource
|
|
|
|
async def download_game_update(self, from_version: str = None):
|
|
version_info = await self._get_game_resource()
|
|
if self._version == version_info.game.latest.version:
|
|
raise ValueError("Game is already up to date.")
|
|
diff_archive = await self.get_game_diff_archive(from_version)
|
|
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_voiceover_update(self, language: str, from_version: str = None):
|
|
diff_archive = await self.get_voiceover_diff_archive(language, from_version)
|
|
if diff_archive is None:
|
|
raise ValueError("Voiceover 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 get_voiceover_diff_archive(self, lang: str, from_version: str = None):
|
|
"""Gets a diff archive from `from_version` to the latest one
|
|
|
|
If from_version is not specified, it will be taken from the game version.
|
|
"""
|
|
game_resource = await self._get_game_resource()
|
|
if not game_resource:
|
|
raise ValueError("Could not fetch game resource")
|
|
translated_lang = self.voiceover_lang_translate(lang)
|
|
for v in game_resource.game.diffs:
|
|
if v.version != from_version:
|
|
continue
|
|
for vo in v.voice_packs:
|
|
if vo.language != translated_lang:
|
|
continue
|
|
return vo
|
|
|
|
async def get_game_diff_archive(self, from_version: str = None):
|
|
"""Gets a diff archive from `from_version` to the latest one
|
|
|
|
If from_version is not specified, it will be taken from the game version.
|
|
"""
|
|
game_resource = await self._get_game_resource()
|
|
for v in game_resource.game.diffs:
|
|
if v.version == from_version:
|
|
return v
|
|
|
|
async def verify_from_pkg_version(self, pkg_version: AsyncPath, ignore_mismatch=False):
|
|
contents = await pkg_version.read_text()
|
|
|
|
async def calculate_md5(file_to_calculate):
|
|
async with AsyncPath(file_to_calculate).open("rb") as f:
|
|
file_hash = hashlib.md5()
|
|
while chunk := await f.read(self._download_chunk):
|
|
file_hash.update(chunk)
|
|
return file_hash.hexdigest()
|
|
|
|
async def verify_file(file_to_verify, md5):
|
|
file_md5 = await calculate_md5(file_to_verify)
|
|
if file_md5 == md5:
|
|
return None
|
|
if ignore_mismatch:
|
|
return file_to_verify, md5, file_md5
|
|
raise ValueError(f"MD5 does not match for {file_to_verify}, expected md5: {md5}, actual md5: {file_md5}")
|
|
|
|
verify_jobs = []
|
|
for content in contents.split("\r\n"):
|
|
if not content.strip():
|
|
continue
|
|
info = json.loads(content)
|
|
verify_jobs.append(verify_file(self._gamedir.joinpath(info["remoteName"]), info["md5"]))
|
|
|
|
verify_result = await asyncio.gather(*verify_jobs)
|
|
failed_files = []
|
|
for file in verify_result:
|
|
if file is not None:
|
|
failed_files.append(file)
|
|
|
|
return None if not failed_files else failed_files
|
|
|
|
async def verify_game(self, pkg_version: str | Path | AsyncPath = None, ignore_mismatch=False):
|
|
if pkg_version is None:
|
|
pkg_version = self._gamedir.joinpath("pkg_version")
|
|
return await self.verify_from_pkg_version(pkg_version, ignore_mismatch)
|
|
|
|
async def clear_cache(self):
|
|
await asyncio.to_thread(shutil.rmtree, self.temp_path, ignore_errors=True)
|