Wrap game resource info from the server

Also add function to get the diff archive (for faster updating)

Download function soon, although I can't make sure that the download function will work properly (like pause, resume download etc.)
This commit is contained in:
tretrauit 2022-02-16 22:18:56 +07:00
parent 81fbdec553
commit da3ee30ab1
Signed by: tretrauit
GPG Key ID: 862760FF1903319E
14 changed files with 227 additions and 84 deletions

View File

@ -1,38 +1,47 @@
import unittest
import asyncio
import worthless
from worthless.classes import launcher
client = worthless.Launcher()
from worthless.classes import launcher, installer
game_launcher = worthless.Launcher()
game_installer = worthless.Installer()
class LauncherOverseasTest(unittest.TestCase):
def test_get_version_info(self):
version_info = asyncio.run(client.get_version_info())
print("get_version_info test.")
print("get_version_info: ", version_info)
self.assertIsInstance(version_info, dict)
version_info = asyncio.run(game_launcher.get_resource_info())
print("get_resource_info test.")
print("get_resource_info: ", version_info)
print("raw: ", version_info.raw)
self.assertIsInstance(version_info, installer.Resource)
def test_get_launcher_info(self):
launcher_info = asyncio.run(client.get_launcher_info())
launcher_info = asyncio.run(game_launcher.get_launcher_info())
print("get_launcher_info test.")
print("get_launcher_info: ", launcher_info)
print("raw: ", launcher_info.raw)
self.assertIsInstance(launcher_info, launcher.Info)
def test_get_launcher_full_info(self):
launcher_info = asyncio.run(client.get_launcher_full_info())
launcher_info = asyncio.run(game_launcher.get_launcher_full_info())
print("get_launcher_full_info test.")
print("get_launcher_full_info: ", launcher_info)
print("raw: ", launcher_info.raw)
self.assertIsInstance(launcher_info, launcher.Info)
def test_get_launcher_background_url(self):
bg_url = asyncio.run(client.get_launcher_background_url())
bg_url = asyncio.run(game_launcher.get_launcher_background_url())
print("get_launcher_background_url test.")
print("get_launcher_background_url: ", bg_url)
self.assertIsInstance(bg_url, str)
self.assertTrue(bg_url)
def test_get_installer_diff(self):
game_diff = asyncio.run(game_installer.get_game_diff_archive("2.4.0"))
print("get_game_diff_archive test.")
print("get_game_diff_archive: ", game_diff)
print("raw: ", game_diff.raw)
self.assertIsInstance(game_diff, installer.Diff)
if __name__ == '__main__':
unittest.main()

View File

@ -1,3 +1,4 @@
from worthless import launcher
from worthless import launcher, installer
Launcher = launcher.Launcher
Installer = installer.Installer

View File

@ -1,6 +1,6 @@
#!/usr/bin/python3
import gui as gui
from worthless import gui
if __name__ == '__main__':
gui.main()

View File

@ -0,0 +1,6 @@
from worthless.classes.installer import resource, game, latest, diff, voicepack
Resource = resource.Resource
Game = game.Game
Latest = latest.Latest
Diff = diff.Diff
Voicepack = voicepack.Voicepack

View File

@ -0,0 +1,21 @@
from worthless.classes.installer.voicepack import Voicepack
class Diff:
def __init__(self, name, version, path, size, md5, is_recommended_update, voice_packs, raw):
self.name = name
self.version = version
self.path = path
self.size = size
self.md5 = md5
self.is_recommended_update = is_recommended_update
self.voice_packs = voice_packs
self.raw = raw
@staticmethod
def from_dict(data):
voice_packs = []
for v in data['voice_packs']:
voice_packs.append(Voicepack.from_dict(v))
return Diff(data["name"], data["version"], data["path"], data["size"], data["md5"],
data["is_recommended_update"], voice_packs, data)

View File

@ -0,0 +1,16 @@
from worthless.classes.installer.latest import Latest
from worthless.classes.installer.diff import Diff
class Game:
def __init__(self, latest, diffs, raw):
self.latest = latest
self.diffs = diffs
self.raw = raw
@staticmethod
def from_dict(data):
diffs = []
for diff in data['diffs']:
diffs.append(Diff.from_dict(diff))
return Game(Latest.from_dict(data['latest']), diffs, data)

View File

@ -0,0 +1,23 @@
from worthless.classes.installer.voicepack import Voicepack
class Latest:
def __init__(self, name, version, path, size, md5, entry, voice_packs, decompressed_path, segments, raw):
self.name = name
self.version = version
self.path = path
self.size = size
self.md5 = md5
self.entry = entry
self.voice_packs = voice_packs
self.decompressed_path = decompressed_path
self.segments = segments
self.raw = raw
@staticmethod
def from_dict(data):
voice_packs = []
for v in data['voice_packs']:
voice_packs.append(Voicepack.from_dict(v))
return Latest(data["name"], data["version"], data["path"], data["size"], data["md5"], data["entry"],
voice_packs, data["decompressed_path"], data["segments"], data)

View File

@ -0,0 +1,30 @@
from worthless.classes.installer.game import Game
class Resource:
"""Contains the game resource information.
Everything except `game` is not wrapped yet
Attributes:
- :class:`worthless.classes.launcher.background.Background` background: The launcher background information.
- :class:`worthless.classes.launcher.banner.Banner` banner: The launcher banner information.
- :class:`worthless.classes.launcher.iconbutton.IconButton` icon: The launcher icon buttons information.
- :class:`worthless.classes.launcher.qq.QQ` post: The launcher QQ posts information.
- :class:`dict` raw: The launcher raw information.
"""
def __init__(self, game, plugin, web_url, force_update, pre_download_game, deprecated_packages, sdk, raw):
self.game = game
self.plugin = plugin
self.web_url = web_url
self.force_update = force_update
self.pre_download_game = pre_download_game
self.deprecated_packages = deprecated_packages
self.sdk = sdk
self.raw = raw
@staticmethod
def from_dict(data):
return Resource(Game.from_dict(data['game']), data['plugin'], data['web_url'], data['force_update'],
data['pre_download_game'], data['deprecated_packages'], data['sdk'], data)

View File

@ -0,0 +1,12 @@
class Voicepack:
def __init__(self, language, name, path, size, md5, raw):
self.language = language
self.name = name
self.path = path
self.size = size
self.md5 = md5
self.raw = raw
@staticmethod
def from_dict(data):
return Voicepack(data["language"], data["name"], data["path"], data["size"], data["md5"], data)

View File

@ -1,7 +1,8 @@
from worthless.classes.launcher import background, banner, iconbutton, iconotherlink, info, post
from worthless.classes.launcher import background, banner, iconbutton, iconotherlink, info, post, qq
Background = background.Background
Banner = banner.Banner
IconButton = iconbutton.IconButton
IconOtherLink = iconotherlink.IconOtherLink
Info = info.Info
Post = post.Post
QQ = qq.QQ

View File

@ -1,6 +1,6 @@
APP_NAME="worthless"
APP_AUTHOR="tretrauit"
LAUNCHER_API_URL_OS = "https://sdk-os-static.mihoyo.com/hk4e_global/mdk/launcher/api"
LAUNCHER_API_URL_OS = "https://sdk-os-static.hoyoverse.com/hk4e_global/mdk/launcher/api"
LAUNCHER_API_URL_CN = "https://sdk-static.mihoyo.com/hk4e_cn/mdk/launcher/api"
PATCH_GIT_URL = "https://notabug.org/Krock/dawn"
TELEMETRY_URL_LIST = [

View File

@ -3,17 +3,24 @@
import argparse
import appdirs
from pathlib import Path
import constants
from worthless.launcher import Launcher
from worthless.installer import Installer
import worthless.constants as constants
class UI:
def __init__(self, gamedir: str, noconfirm: bool) -> None:
self._noconfirm = noconfirm
self._gamedir = gamedir
self._launcher = Launcher(gamedir)
self._installer = Installer(gamedir)
def _ask(self, title, description):
raise NotImplementedError()
def get_game_version(self):
print(self._installer.get_game_version())
def install_game(self):
# TODO
raise NotImplementedError("Install game is not implemented.")
@ -38,7 +45,7 @@ def main():
help="Specify the temporary directory (default {} and {})".format(default_dirs.user_data_dir,
default_dirs.user_cache_dir))
parser.add_argument("-S", "--install", action="store_true",
help="Install the game (if not already installed, else do nothing)")
help="Install/update the game (if not already installed, else do nothing)")
parser.add_argument("-U", "--install-from-file", action="store_true",
help="Install the game from the game archive (if not already installed, \
else update from archive)")
@ -46,21 +53,28 @@ def main():
help="Patch the game (if not already patched, else do nothing)")
parser.add_argument("-Sy", "--update", action="store_true",
help="Update the game and specified voiceover pack only (or install if not found)")
parser.add_argument("-Sv", "--update-voiceover", action="store_true",
help="Update the voiceover pack only (or install if not found)")
parser.add_argument("-Syu", "--update-all", action="store_true",
help="Update the game and all installed voiceover packs (or install if not found)")
parser.add_argument("-Rs", "--remove", action="store_true", help="Remove the game (if installed)")
parser.add_argument("-Rp", "--remove-patch", action="store_true", help="Revert the game patch (if patched)")
parser.add_argument("-Rv", "--remove-voiceover", action="store_true", help="Remove a Voiceover pack (if installed)")
parser.add_argument("--get-game-version", action="store_true", help="Get the current game version")
parser.add_argument("--no-overseas", action="store_true", help="Don't use overseas server")
parser.add_argument("--noconfirm", action="store_true",
help="Do not ask any for confirmation. (Ignored in interactive mode)")
args = parser.parse_args()
interactive_mode = not args.install and not args.install_from_file and not args.patch and not args.update and not \
args.remove and not args.remove_patch and not args.remove_voiceover
args.remove and not args.remove_patch and not args.remove_voiceover and not args.get_game_version
ui = UI(args.dir, args.noconfirm)
if args.install and args.update:
raise ValueError("Cannot specify both --install and --update arguments.")
if args.get_game_version:
ui.get_game_version()
if args.install:
ui.install_game()

View File

@ -7,27 +7,35 @@ from worthless import constants
from worthless.launcher import Launcher
def _read_version_from_game_file(globalgamemanagers: Path):
with globalgamemanagers.open("rb") as f:
data = f.read().decode("ascii", errors="ignore")
result = re.search(r"([1-9]+\.[0-9]+\.[0-9]+)_[\d]+_[\d]+", data)
if not result:
raise ValueError("Could not find version in game file")
return result.group(1)
class Installer:
def _read_version_from_config(self):
if self._config_file.exists():
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("miHoYo", "game_version")
return cfg.get("General", "game_version")
# https://gitlab.com/KRypt0n_/an-anime-game-launcher/-/blob/main/src/ts/Game.ts#L26
def _read_version_from_game_file(self):
if self._overseas:
globalgamemanagers = self._gamedir.joinpath("./GenshinImpact_Data/globalgamemanagers")
def get_game_version(self):
if self._config_file.exists():
return self._read_version_from_config()
else:
globalgamemanagers = self._gamedir.joinpath("./YuanShen_Data/globalgamemanagers")
if globalgamemanagers.exists():
with globalgamemanagers.open("rb") as f:
data = f.read().decode("ascii")
result = re.search(r"([1-9]+\.[0-9]+\.[0-9]+)_[\d]+_[\d]+", data)
if not result:
raise ValueError("Could not find version in game file")
return result.group(1)
if self._overseas:
globalgamemanagers = self._gamedir.joinpath("./GenshinImpact_Data/globalgamemanagers")
else:
globalgamemanagers = self._gamedir.joinpath("./YuanShen_Data/globalgamemanagers")
if not globalgamemanagers.exists():
return
return _read_version_from_game_file(globalgamemanagers)
def __init__(self, gamedir: str | Path = Path.cwd(), overseas: bool = True, data_dir: str | Path = None):
if isinstance(gamedir, str):
@ -44,9 +52,24 @@ class Installer:
self._config_file = config_file.resolve()
self._version = None
self._overseas = overseas
self._launcher = Launcher(self._gamedir, self._overseas)
if config_file.exists():
self._version = self._read_version_from_config()
elif gamedir.joinpath("./GenshinImpact_Data/globalgamemanagers").exists():
self._version = self._read_version_from_game_file()
self._launcher = Launcher(self._gamedir, overseas=self._overseas)
self._version = self.get_game_version()
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.
"""
if not from_version:
if self._version:
from_version = self._version
else:
from_version = self._version = 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")
for v in game_resource.game.diffs:
if v.version == from_version:
return v

View File

@ -2,7 +2,36 @@ import aiohttp
import locale
from worthless import constants
from pathlib import Path
from worthless.classes import launcher
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:
@ -10,7 +39,7 @@ class Launcher:
Contains functions to get information from server and client like the official launcher.
"""
def __init__(self, gamedir=Path.cwd(), language=None, overseas=True):
def __init__(self, gamedir: str | Path = Path.cwd(), language: str = None, overseas=True):
"""Initialize the launcher API
Args:
@ -23,7 +52,7 @@ class Launcher:
"key": "gcStgarh",
"launcher_id": "10",
}
self._lang = self._get_system_language() if not language else language.lower().replace("_", "-")
self._lang = language.lower().replace("_", "-") if language else _get_system_language()
else:
self._api = constants.LAUNCHER_API_URL_CN
self._params = {
@ -36,42 +65,13 @@ class Launcher:
gamedir = Path(gamedir)
self._gamedir = gamedir.resolve()
@staticmethod
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
@staticmethod
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
async def _get_launcher_info(self, adv=True) -> launcher.Info:
params = self._params | {"filter_adv": str(adv).lower(),
"language": self._lang}
rsp = await self._get(self._api + "/content", params=params)
rsp = await _get(self._api + "/content", params=params)
if rsp["data"]["adv"] is None:
params["language"] = "en-us"
rsp = await self._get(self._api + "/content", params=params)
rsp = await _get(self._api + "/content", params=params)
lc_info = launcher.Info.from_dict(rsp["data"])
return lc_info
@ -94,7 +94,7 @@ class Launcher:
self._lang = language.lower().replace("_", "-")
async def get_version_info(self) -> dict:
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.
@ -105,8 +105,8 @@ class Launcher:
aiohttp.ClientResponseError: An error occurred while fetching the information.
"""
rsp = await self._get(self._api + "/resource", params=self._params)
return rsp
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
@ -143,16 +143,3 @@ class Launcher:
rsp = await self.get_launcher_info()
return rsp.background.background
async def get_system_game_info(self, table_handle, keys, require_all_keys):
# TODO: Implement
raise NotImplementedError("Not implemented yet.")
pass
async def get_system_game_version(self) -> str:
"""Gets the game version from the current system.
:return: str: System game version.
"""
rsp = await self.get_version_info()
return rsp["data"]["system"]["game_version"]