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
149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
import aiohttp
|
|
import locale
|
|
|
|
from aiopath import AsyncPath
|
|
|
|
from worthless import constants
|
|
from pathlib import Path
|
|
from worthless.classes import launcher, installer
|
|
|
|
|
|
async def _get(url, **kwargs) -> dict:
|
|
# Workaround because miHoYo uses retcode for their API instead of HTTP status code
|
|
async with aiohttp.ClientSession() as session:
|
|
rsp = await session.get(url, **kwargs)
|
|
rsp_json = await rsp.json()
|
|
if rsp_json["retcode"] != 0:
|
|
# TODO: Add more information to the error message
|
|
raise aiohttp.ClientResponseError(code=rsp_json["retcode"],
|
|
message=rsp_json["message"],
|
|
history=rsp.history,
|
|
request_info=rsp.request_info)
|
|
return rsp_json
|
|
|
|
|
|
def _get_system_language() -> str:
|
|
"""Gets system language compatible with server parameters.
|
|
|
|
Return:
|
|
System language with format xx-xx.
|
|
"""
|
|
|
|
try:
|
|
lang = locale.getdefaultlocale()[0]
|
|
lowercase_lang = lang.lower().replace("_", "-")
|
|
return lowercase_lang
|
|
except ValueError:
|
|
return "en-us" # Fallback to English if locale is not supported
|
|
|
|
|
|
class Launcher:
|
|
"""
|
|
Contains functions to get information from server and client like the official launcher.
|
|
"""
|
|
|
|
def __init__(self, gamedir: str | Path = Path.cwd(), language: str = None, overseas=True):
|
|
"""Initialize the launcher API
|
|
|
|
Args:
|
|
gamedir (Path): Path to the game directory.
|
|
"""
|
|
self._overseas = overseas
|
|
if overseas:
|
|
self._api = constants.LAUNCHER_API_URL_OS
|
|
self._params = {
|
|
"key": "gcStgarh",
|
|
"launcher_id": "10",
|
|
}
|
|
self._lang = language.lower().replace("_", "-") if language else _get_system_language()
|
|
else:
|
|
self._api = constants.LAUNCHER_API_URL_CN
|
|
self._params = {
|
|
"key": "eYd89JmJ",
|
|
"launcher_id": "18",
|
|
"channel_id": "1"
|
|
}
|
|
self._lang = "zh-cn" # Use chinese language because this is chinese version
|
|
if isinstance(gamedir, str | AsyncPath):
|
|
gamedir = Path(gamedir)
|
|
self._gamedir = gamedir.resolve()
|
|
|
|
async def _get_launcher_info(self, adv=True) -> launcher.Info:
|
|
params = self._params | {"filter_adv": str(adv).lower(),
|
|
"language": self._lang}
|
|
rsp = await _get(self._api + "/content", params=params)
|
|
if rsp["data"]["adv"] is None:
|
|
params["language"] = "en-us"
|
|
rsp = await _get(self._api + "/content", params=params)
|
|
lc_info = launcher.Info.from_dict(rsp["data"])
|
|
return lc_info
|
|
|
|
async def override_gamedir(self, gamedir: str | Path) -> None:
|
|
"""Overrides game directory with another directory.
|
|
|
|
Args:
|
|
gamedir (str): New directory to override with.
|
|
"""
|
|
if isinstance(gamedir, str):
|
|
gamedir = Path(gamedir).resolve()
|
|
self._gamedir = gamedir
|
|
|
|
async def override_language(self, language: str) -> None:
|
|
"""Overrides system detected language with another language.
|
|
|
|
Args:
|
|
language (str): Language to override with.
|
|
"""
|
|
|
|
self._lang = language.lower().replace("_", "-")
|
|
|
|
async def get_resource_info(self) -> installer.Resource:
|
|
"""Gets version info from the server.
|
|
|
|
This function gets version info including audio pack and their download url from the server.
|
|
|
|
Returns:
|
|
A dict containing version info from the server.
|
|
Raises:
|
|
aiohttp.ClientResponseError: An error occurred while fetching the information.
|
|
"""
|
|
|
|
rsp = await _get(self._api + "/resource", params=self._params)
|
|
return installer.Resource.from_dict(rsp["data"])
|
|
|
|
async def get_launcher_info(self) -> launcher.Info:
|
|
"""Gets short launcher info from the server
|
|
|
|
This function only gets background image and the FAQ url from the server.
|
|
|
|
Returns:
|
|
A dict containing short launcher info from the server.
|
|
Raises:
|
|
aiohttp.ClientResponseError: An error occurred while fetching the information.
|
|
"""
|
|
|
|
return await self._get_launcher_info(adv=True)
|
|
|
|
async def get_launcher_full_info(self) -> launcher.Info:
|
|
"""Gets full launcher info from the server.
|
|
|
|
Returns:
|
|
A dict containing full launcher info from the server.
|
|
Raises:
|
|
aiohttp.ClientResponseError: An error occurred while fetching the information.
|
|
"""
|
|
|
|
return await self._get_launcher_info(adv=False)
|
|
|
|
async def get_launcher_background_url(self) -> str:
|
|
"""Gets launcher background image url from the server.
|
|
|
|
Returns:
|
|
Background image url.
|
|
Raises:
|
|
aiohttp.ClientResponseError: An error occurred while fetching the background image.
|
|
"""
|
|
|
|
rsp = await self.get_launcher_info()
|
|
return rsp.background.background
|