worthless-launcher/worthless/launcher.py
tretrauit a5659f7ff3
refactor: convert all task-intensive functions to async.
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
2022-06-25 01:13:47 +07:00

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