587 lines
24 KiB
Python
587 lines
24 KiB
Python
import asyncio
|
|
import re
|
|
import shutil
|
|
import platform
|
|
import aiohttp
|
|
import zipfile
|
|
import json
|
|
import hashlib
|
|
import secrets
|
|
import logging
|
|
from pathlib import Path
|
|
from configparser import ConfigParser
|
|
from worthless import constants
|
|
from worthless.launcher import Launcher
|
|
from worthless.enums import GameVariant
|
|
from worthless.launcherconfig import LauncherConfig
|
|
|
|
logger = logging.getLogger("worthless.GameManager")
|
|
|
|
|
|
async def _download_file(file_url: str, file_name: str, file_path: Path | str, file_len: int = None, overwrite=False,
|
|
chunks=None, threads_num=None) -> Path:
|
|
"""
|
|
Download file name to temporary directory,
|
|
:param file_url: The url to download the file from
|
|
:param file_name: The file name to download into
|
|
:return: Downloaded file as a Path object
|
|
"""
|
|
if not chunks:
|
|
chunks = 8192
|
|
if not threads_num:
|
|
threads_num = 8
|
|
file_path = Path(file_path).joinpath(file_name)
|
|
|
|
async def _download(session: aiohttp.ClientSession, from_bytes: int, to_bytes: int, threaded: bool = None) -> Path:
|
|
headers = {
|
|
"Range": f"bytes={from_bytes}-{to_bytes if to_bytes else ''}"
|
|
}
|
|
if not threaded:
|
|
p = file_path
|
|
else:
|
|
p = file_path.parent.joinpath(secrets.token_urlsafe(16))
|
|
p.touch(exist_ok=True)
|
|
rsp = await session.get(file_url, headers=headers, timeout=None)
|
|
if rsp.status == 416:
|
|
# Not an error, so yeah.
|
|
return p
|
|
rsp.raise_for_status()
|
|
while True:
|
|
chunk = await rsp.content.read(chunks)
|
|
if not chunk:
|
|
break
|
|
with p.open("ab") as file:
|
|
await asyncio.to_thread(file.write, chunk)
|
|
return p
|
|
if overwrite:
|
|
file_path.unlink(missing_ok=True)
|
|
if file_path.exists():
|
|
cur_len = file_path.stat().st_size
|
|
else:
|
|
file_path.touch()
|
|
cur_len = 0
|
|
if not file_len or threads_num == 1:
|
|
async with aiohttp.ClientSession() as s:
|
|
return await _download(session=s, from_bytes=cur_len, to_bytes=file_len)
|
|
|
|
download_bytes = file_len - cur_len
|
|
# if bytes * threads is smaller than file_len then we will not get the full file.
|
|
download_bytes_t = int(download_bytes / threads_num) + 1
|
|
download_jobs = []
|
|
current_bytes = cur_len
|
|
async with aiohttp.ClientSession() as s:
|
|
for thread in range(threads_num):
|
|
next_bytes = current_bytes + download_bytes_t
|
|
if next_bytes > file_len:
|
|
next_bytes = file_len
|
|
download_jobs.append(_download(session=s, from_bytes=current_bytes, to_bytes=next_bytes, threaded=True))
|
|
all_bytes = await asyncio.gather(*download_jobs)
|
|
# Merge bytes into the file
|
|
with file_path.open("ab") as f:
|
|
for bytes_path in all_bytes:
|
|
f.write(bytes_path.read_bytes())
|
|
bytes_path.unlink()
|
|
|
|
|
|
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"
|
|
|
|
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())
|
|
file = platform_arch_path.joinpath(exec_name)
|
|
if file.exists():
|
|
file.chmod(0o755)
|
|
return str(file)
|
|
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, threads_num=1)
|
|
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 GameManager:
|
|
def __init__(self, game_dir: str | Path = None,
|
|
variant: GameVariant = None, data_dir: str | Path = None):
|
|
if not game_dir:
|
|
game_dir = Path.cwd()
|
|
if not isinstance(game_dir, Path):
|
|
game_dir = Path(game_dir)
|
|
self._game_dir = game_dir
|
|
if not data_dir:
|
|
self._cache_path = Path(constants.APPDIRS.user_cache_dir).joinpath("Installer")
|
|
else:
|
|
if not isinstance(data_dir, Path):
|
|
data_dir = Path(data_dir)
|
|
self.temp_path = data_dir.joinpath("Temp/Installer/")
|
|
Path(self.temp_path).mkdir(parents=True, exist_ok=True)
|
|
config_file = self._game_dir.joinpath("config.ini")
|
|
self._config_file = config_file
|
|
self._download_chunk = 8192
|
|
self._variant = variant
|
|
self._version = None
|
|
self._launcher = Launcher(self._game_dir, variant=variant)
|
|
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)
|
|
|
|
def read_version_from_config(self):
|
|
if not self._config_file.exists():
|
|
raise FileNotFoundError(f"Config file {self._config_file} not found")
|
|
cfg = ConfigParser()
|
|
cfg.read(str(self._config_file))
|
|
return cfg.get("General", "game_version")
|
|
|
|
def read_version_from_game_file(self, globalgamemanagers: 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 (e.g. 1.0.0)
|
|
"""
|
|
if isinstance(globalgamemanagers, Path):
|
|
data = globalgamemanagers.read_text(encoding="ascii", errors="ignore")
|
|
else:
|
|
data = globalgamemanagers.decode(encoding="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) -> 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
|
|
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 = GameManager.get_voiceover_archive_language(voiceover_archive)
|
|
with zipfile.ZipFile(voiceover_archive, 'r') as f:
|
|
archive_path = zipfile.Path(f)
|
|
files = 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):
|
|
match self._variant:
|
|
case GameVariant.INTERNATIONAL:
|
|
return "GenshinImpact_Data/"
|
|
case GameVariant.CHINESE:
|
|
return "YuanShen_Data/"
|
|
case GameVariant.BILIBILI:
|
|
return "YuanShen_Data/"
|
|
|
|
def get_game_data_path(self) -> Path:
|
|
return self._game_dir.joinpath(self.get_game_data_name())
|
|
|
|
def get_game_archive_version(self, game_archive: str | Path):
|
|
game_archive = Path(game_archive)
|
|
if not game_archive.is_file():
|
|
raise FileNotFoundError(f"Game archive {game_archive} not found")
|
|
with zipfile.ZipFile(game_archive, 'r') as f:
|
|
return self.read_version_from_game_file(
|
|
f.read(self.get_game_data_name() + "globalgamemanagers")
|
|
)
|
|
|
|
def get_game_version(self) -> str | None:
|
|
globalgamemanagers = self.get_game_data_path().joinpath("./globalgamemanagers")
|
|
if not globalgamemanagers.exists():
|
|
try:
|
|
return self.read_version_from_config()
|
|
except FileNotFoundError:
|
|
return
|
|
return self.read_version_from_game_file(globalgamemanagers)
|
|
|
|
def get_installed_voiceovers(self) -> list[str]:
|
|
"""
|
|
Returns a list of installed voiceovers.
|
|
|
|
:return: List of installed voiceovers
|
|
"""
|
|
voiceovers = []
|
|
for file in self.get_game_data_path() \
|
|
.joinpath("StreamingAssets/Audio/GeneratedSoundBanks/Windows").iterdir():
|
|
if file.is_dir():
|
|
voiceovers.append(file.name)
|
|
return voiceovers
|
|
|
|
async def update_game(self, game_archive: str | Path):
|
|
if not self.get_game_data_path().exists():
|
|
raise FileNotFoundError(f"Game not found in {self._game_dir}")
|
|
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._game_dir.joinpath(file)
|
|
if not 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):
|
|
diff_path = self.temp_path.joinpath(diff_file)
|
|
if diff_path.is_file():
|
|
diff_path.unlink(missing_ok=True)
|
|
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 = await 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.
|
|
await old_file.rename(old_file.with_suffix(old_suffix))
|
|
return
|
|
await 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 = Path(self._game_dir.joinpath(file))
|
|
if not 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._game_dir, members=files)
|
|
archive.close()
|
|
# Update game version on local variable.
|
|
self._version = 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, 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_voiceover(self, language: str, pre_download=False):
|
|
game = await self._get_game(pre_download)
|
|
translated_lang = self.voiceover_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 uninstall_game(self):
|
|
shutil.rmtree(self._game_dir, ignore_errors=True)
|
|
|
|
def _extract_game_file(self, archive: str | Path):
|
|
if isinstance(archive, str):
|
|
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._game_dir)
|
|
|
|
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 self.get_game_data_path().exists():
|
|
raise FileNotFoundError(f"Game not found in {self._game_dir}")
|
|
self._extract_game_file(voiceover_archive)
|
|
|
|
async def install_game(self, game_archive: str | Path, force_reinstall: bool = False):
|
|
"""Installs the game to the current directory
|
|
|
|
If `force_reinstall` is True, the game will be uninstalled then reinstalled.
|
|
"""
|
|
if self.get_game_data_path().exists():
|
|
if not force_reinstall:
|
|
raise ValueError(f"Game is already installed in {self._game_dir}")
|
|
self.uninstall_game()
|
|
|
|
self._game_dir.mkdir(parents=True, exist_ok=True)
|
|
self._extract_game_file(game_archive)
|
|
self._version = self.get_game_version()
|
|
self.set_version_config()
|
|
|
|
async def _get_game_version(self):
|
|
if self._version:
|
|
from_version = self._version
|
|
else:
|
|
from_version = self._version = self.get_game_version()
|
|
return from_version
|
|
|
|
async def _get_game_resource(self):
|
|
game_resource = await self._launcher.get_resource_info()
|
|
if not game_resource:
|
|
raise ValueError("Could not fetch game resource")
|
|
return game_resource
|
|
|
|
async def _get_game(self, pre_download=False):
|
|
game_resource = await self._get_game_resource()
|
|
game = game_resource.game
|
|
if pre_download:
|
|
game = game_resource.pre_download_game
|
|
return game
|
|
|
|
async def download_game_update(self, from_version: str = None, pre_download=False):
|
|
if not from_version:
|
|
from_version = await self._get_game_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_voiceover_update(self, language: str, from_version: str = None, pre_download=False):
|
|
if not from_version:
|
|
from_version = await self._get_game_version()
|
|
diff_archive = await self.get_voiceover_diff_archive(language, from_version, pre_download)
|
|
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, pre_download=False):
|
|
"""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.
|
|
"""
|
|
if not from_version:
|
|
from_version = await self._get_game_version()
|
|
game = await self._get_game(pre_download=pre_download)
|
|
translated_lang = self.voiceover_lang_translate(lang)
|
|
for v in game.diffs:
|
|
if v.version != from_version:
|
|
continue
|
|
for vo in v.voice_packs:
|
|
if vo.language == translated_lang:
|
|
return vo
|
|
|
|
async def get_game_diff_archive(self, from_version: str = None, pre_download=False):
|
|
"""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.
|
|
"""
|
|
if not from_version:
|
|
from_version = await self._get_game_version()
|
|
game = await self._get_game(pre_download=pre_download)
|
|
for v in game.diffs:
|
|
if v.version == from_version:
|
|
return v
|
|
|
|
async def verify_from_pkg_version(self, pkg_version: Path, ignore_mismatch=False):
|
|
contents = pkg_version.read_text()
|
|
|
|
def calculate_md5(file_to_calculate):
|
|
file_to_calculate = Path(file_to_calculate)
|
|
if not file_to_calculate.exists():
|
|
return ""
|
|
with file_to_calculate.open("rb") as f:
|
|
file_hash = hashlib.md5()
|
|
while chunk := f.read(self._download_chunk):
|
|
file_hash.update(chunk)
|
|
return file_hash.hexdigest()
|
|
|
|
def verify_file(file_to_verify, md5):
|
|
file_md5 = 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(asyncio.to_thread(verify_file, self._game_dir.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 = None, ignore_mismatch=False):
|
|
if pkg_version is None:
|
|
pkg_version = self._game_dir.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)
|