From 2a9c512943e10a5faa8705d2fb9765164ac23103 Mon Sep 17 00:00:00 2001 From: ashlen Date: Sun, 11 Aug 2024 20:00:38 +0200 Subject: [PATCH] Remove unused asset features --- arkprts/assets/base.py | 19 +---- arkprts/assets/bundle.py | 156 ++++++++++++--------------------------- arkprts/assets/git.py | 99 ++++++------------------- arkprts/auth.py | 3 +- arkprts/models/base.py | 4 +- arkprts/network.py | 15 ++++ requirements.txt | 1 - setup.py | 4 +- 8 files changed, 93 insertions(+), 208 deletions(-) diff --git a/arkprts/assets/base.py b/arkprts/assets/base.py index 172bf59..2b4b7fd 100644 --- a/arkprts/assets/base.py +++ b/arkprts/assets/base.py @@ -38,10 +38,10 @@ class Assets(abc.ABC): def __init__( self, *, - default_server: netn.ArknightsServer = "en", + default_server: netn.ArknightsServer | None = None, json_loads: typing.Callable[[bytes], typing.Any] = json.loads, ) -> None: - self.default_server = default_server + self.default_server = default_server or "en" self.loaded = False self.excel_cache = {} self.json_loads = json_loads @@ -75,10 +75,6 @@ async def update_assets(self) -> None: def get_file(self, path: str, *, server: netn.ArknightsServer | None = None) -> bytes: """Get an extracted asset file. If server is None any server is allowed with preference for default server.""" - @abc.abstractmethod - async def aget_file(self, path: str, *, server: netn.ArknightsServer | None = None) -> bytes: - """Get an extracted asset file without requiring load.""" - def get_excel(self, name: str, *, server: netn.ArknightsServer | None = None) -> models.DDict: """Get a gamedata table file.""" path = f"gamedata/excel/{name}.json" @@ -90,17 +86,6 @@ def get_excel(self, name: str, *, server: netn.ArknightsServer | None = None) -> return models.DDict(data) - async def aget_excel(self, name: str, *, server: netn.ArknightsServer | None = None) -> models.DDict: - """Get a gamedata table file without requiring load.""" - path = f"gamedata/excel/{name}.json" - if data := self.excel_cache.setdefault(server or self.default_server, {}).get(path): - return models.DDict(data) - - data = self.json_loads(await self.aget_file(path, server=server)) - self.excel_cache[server or self.default_server][path] = data - - return models.DDict(data) - def __getitem__(self, name: str) -> models.DDict: """Get a gamedata table file.""" return self.get_excel(name) diff --git a/arkprts/assets/bundle.py b/arkprts/assets/bundle.py index 222a780..4832ac1 100644 --- a/arkprts/assets/bundle.py +++ b/arkprts/assets/bundle.py @@ -7,9 +7,7 @@ from __future__ import annotations import asyncio -import concurrent.futures import fnmatch -import functools import io import json import logging @@ -20,7 +18,6 @@ import subprocess import tempfile import typing -import warnings import zipfile from arkprts import network as netn @@ -56,7 +53,7 @@ def unzip_only_file(stream: io.BytesIO | bytes) -> bytes: def resolve_unity_asset_cache(filename: str, server: netn.ArknightsServer) -> pathlib.Path: """Resolve a path to a cached arknights ab file.""" - path = pathlib.Path(tempfile.gettempdir()) / "ArknightsUnity" / filename + path = netn.TEMP_DIR / "ArknightsAB" / filename path.parent.mkdir(parents=True, exist_ok=True) return path.with_suffix(".ab") @@ -117,7 +114,8 @@ def run_flatbuffers( ] result = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) # noqa: S603, UP022 if result.returncode != 0: - file = pathlib.Path(tempfile.mktemp(".log")) + file = pathlib.Path(tempfile.mktemp(".log", dir=netn.TEMP_DIR / "flatbufferlogs")) + file.parent.mkdir(parents=True, exist_ok=True) file.write_bytes(result.stdout + b"\n\n\n\n" + result.stderr) raise ValueError( f"flatc failed with code {result.returncode}: {file} `{shlex.join(args)}` (random exit code likely means a faulty FBS file was provided)", @@ -132,7 +130,7 @@ def resolve_fbs_schema_directory(server: typing.Literal["cn", "yostar"]) -> path if path: return pathlib.Path(path) - core_path = pathlib.Path(tempfile.gettempdir()) / "ArknightsFBS" + core_path = netn.APPDATA_DIR / "ArknightsFBS" core_path.mkdir(parents=True, exist_ok=True) path = core_path / server / "OpenArknightsFBS" / "FBS" os.environ[f"FLATBUFFERS_SCHEMA_DIR_{server.upper()}"] = str(path) @@ -178,7 +176,7 @@ def decrypt_fbs_file( if rsa: data = data[128:] - tempdir = pathlib.Path(tempfile.gettempdir()) / "TempArknightsFBS" + tempdir = netn.TEMP_DIR / "ArknightsFBS" tempdir.mkdir(parents=True, exist_ok=True) fbs_path = tempdir / (table_name + ".bytes") @@ -235,23 +233,15 @@ def normalize_json(data: bytes, *, indent: int = 4, lenient: bool = True) -> byt DYNP = r"assets/torappu/dynamicassets/" -def unpack_assets( +def find_ab_assets( asset: UnityPyAsset, - target_container: str | None = None, - # target_path: str | None = None, *, - server: netn.ArknightsServer | None = None, + server: netn.ArknightsServer, normalize: bool = False, ) -> typing.Iterable[tuple[str, bytes]]: """Yield relative paths and data for a unity asset.""" for container, obj in asset.container.items(): - if target_container and container != target_container: - continue - if obj.type.name == "TextAsset": - if server is None: - raise TypeError("Server required for text decryption") - if match := re.match(DYNP + r"(.+\.txt)", container): data = obj.read() yield (match[1], data.script) @@ -293,23 +283,28 @@ def unpack_assets( continue -def guess_asset_path(path: str, hot_update_list: typing.Any) -> typing.Sequence[str]: - """Return a sequence of all files thought to be needed to be downloaded for an asset to be available.""" - # images have to be added later - match = re.match(r"(gamedata/\w+).json", path) - if not match: - return [] +def extract_ab( + ab_path: PathLike, + save_directory: PathLike, + *, + server: netn.ArknightsServer, + normalize: bool = False, +) -> typing.Sequence[pathlib.Path]: + """Extract an AB file and save files. Returns a list of found files.""" + ab_path = pathlib.Path(ab_path) + save_directory = pathlib.Path(save_directory) + asset = load_unity_file(ab_path.read_bytes()) - filename = match[0] + paths: list[pathlib.Path] = [] + for unpacked_rel_path, unpacked_data in find_ab_assets(asset, server=server, normalize=normalize): + savepath = save_directory / server / unpacked_rel_path + savepath.parent.mkdir(exist_ok=True, parents=True) + savepath.write_bytes(unpacked_data) - asset_paths: list[str] = [] - for info in hot_update_list["abInfos"]: - # just supporting excel for now - match = re.match(DYNP + filename + r"(?:[a-fA-F0-9]{6})?\.ab", info["name"]) - if match: - asset_paths.append(info["name"]) + LOGGER.debug("Extracted asset %s from %s for server %s", unpacked_rel_path, ab_path.name, server) + paths.append(savepath) - return asset_paths + return paths def get_outdated_hashes(hot_update_now: typing.Any, hot_update_before: typing.Any) -> typing.Sequence[str]: @@ -348,11 +343,10 @@ def __init__( except OSError as e: raise ImportError("Cannot use BundleAssets without a flatc executable") from e - super().__init__(default_server=default_server or "en", json_loads=json_loads) + super().__init__(default_server=default_server or (network and network.default_server), json_loads=json_loads) - temporary_directory = pathlib.Path(tempfile.gettempdir()) - self.directory = pathlib.Path(directory or temporary_directory / "ArknightsResources") - self.network = network or netn.NetworkSession(default_server=default_server) + self.directory = pathlib.Path(directory or netn.APPDATA_DIR / "ArknightsResources") + self.network = network or netn.NetworkSession(default_server=self.default_server) async def _download_asset(self, path: str, *, server: netn.ArknightsServer | None = None) -> bytes: """Download a raw zipped unity asset.""" @@ -381,7 +375,6 @@ def _get_current_hot_update_list(self, server: netn.ArknightsServer) -> typing.A if not path.exists(): return None - path.parent.mkdir(exist_ok=True, parents=True) with path.open("r") as file: return json.load(file) @@ -389,45 +382,15 @@ async def _download_unity_file( self, path: str, *, - save: bool = True, server: netn.ArknightsServer | None = None, - ) -> bytes: - """Download an asset and return it unzipped.""" + ) -> pathlib.Path: + """Download an asset and return its path.""" LOGGER.debug("Downloading and unzipping asset %s for server %s", path, server) zipped_data = await self._download_asset(path, server=server) data = unzip_only_file(zipped_data) - if save: - p = resolve_unity_asset_cache(path, server=server or self.default_server) - p.write_bytes(data) - - return data - - def _parse_and_save( - self, - data: bytes, - *, - target_container: str | None = None, - server: netn.ArknightsServer | None = None, - normalize: bool = False, - ) -> typing.Iterable[tuple[str, bytes]]: - """Download and extract an asset.""" - server = server or self.default_server - - asset = load_unity_file(data) - - fetched_any = False - for fetched_any, (unpacked_rel_path, unpacked_data) in enumerate( - unpack_assets(asset, target_container, server=server, normalize=normalize), - 1, - ): - savepath = self.directory / server / unpacked_rel_path - savepath.parent.mkdir(exist_ok=True, parents=True) - savepath.write_bytes(unpacked_data) - - yield (unpacked_rel_path, unpacked_data) - - if not fetched_any: - warnings.warn(f"Unpacking yielded no results (container: {target_container}) ") + cache_path = resolve_unity_asset_cache(path, server=server or self.default_server) + cache_path.write_bytes(data) + return cache_path async def update_assets( self, @@ -449,9 +412,9 @@ async def update_assets( return hot_update_list = await self._get_hot_update_list(server) - requested_names = [info["name"] for info in hot_update_list["abInfos"] if fnmatch.fnmatch(info["name"], allow)] - old_hot_update_list = self._get_current_hot_update_list(server) + + requested_names = [info["name"] for info in hot_update_list["abInfos"] if fnmatch.fnmatch(info["name"], allow)] if old_hot_update_list and not force: outdated_names = set(get_outdated_hashes(hot_update_list, old_hot_update_list)) requested_names = [name for name in requested_names if name in outdated_names] @@ -459,24 +422,17 @@ async def update_assets( if any("gamedata" in name for name in requested_names): await update_fbs_schema() - datas = await asyncio.gather(*(self._download_unity_file(name, server=server) for name in requested_names)) - loop = asyncio.get_event_loop() - # this should be a ProcessPoolExecutor but pickling is a problem in classes - with concurrent.futures.ThreadPoolExecutor() as executor: - futures = [ - loop.run_in_executor( - executor, - functools.partial(self._parse_and_save, d, server=server, normalize=normalize), - ) - for d in datas - ] - for name, f in zip(requested_names, asyncio.as_completed(futures)): - try: - for path, _ in await f: - LOGGER.debug("Extracted asset %s from %s for server %s", path, name, server) - except Exception as e: - LOGGER.exception("Failed to extract asset %s for server %s", name, server, exc_info=e) + # download and extract assets + ab_file_paths = await asyncio.gather( + *(self._download_unity_file(name, server=server) for name in requested_names), + ) + for path in ab_file_paths: + try: + extract_ab(path, self.directory, server=server, normalize=normalize) + except Exception as e: + LOGGER.exception("Failed to extract asset %s for server %s", path.name, server, exc_info=e) + # save new hot_update_list hot_update_list_path = self.directory / server / "hot_update_list.json" hot_update_list_path.parent.mkdir(parents=True, exist_ok=True) with hot_update_list_path.open("w") as file: @@ -487,23 +443,3 @@ async def update_assets( def get_file(self, path: str, *, server: netn.ArknightsServer | None = None) -> bytes: """Get an extracted asset file. If server is None any server is allowed with preference for default server.""" return (self.directory / (server or self.default_server) / path).read_bytes() - - async def aget_file(self, path: str, *, server: netn.ArknightsServer | None = None, save: bool = True) -> bytes: - """Get an extracted asset file without requiring load.""" - server = server or self.default_server - hot_update_list = await self._get_hot_update_list(server) - asset_paths = guess_asset_path(path, hot_update_list) - if not asset_paths: - raise ValueError("No viable asset path found, please load all assets and use get_file.") - - for potential_asset_path in asset_paths: - asset = load_unity_file(await self._download_unity_file(potential_asset_path, server=server)) - for output_path, data in unpack_assets(asset, path, server=server): - if save: - savepath = self.directory / server / output_path - savepath.parent.mkdir(exist_ok=True, parents=True) - savepath.write_bytes(data) - - return data - - raise ValueError("File not found, please load all assets and use get_file.") diff --git a/arkprts/assets/git.py b/arkprts/assets/git.py index d78c283..4b4264e 100644 --- a/arkprts/assets/git.py +++ b/arkprts/assets/git.py @@ -29,23 +29,19 @@ LOGGER: logging.Logger = logging.getLogger("arkprts.assets.git") +PathLike = typing.Union[pathlib.Path, str] + CN_GAMEDATA_REPOSITORY = "Kengxxiao/ArknightsGameData" # master -GLOBAL_GAMEDATA_REPOSITORY = "Kengxxiao/ArknightsGameData_YoStar" # main -TW_GAMEDATA_REPOSITORY = "aelurum/ArknightsGameData" # zh-tw fork # master_v2 -RESOURCES_REPOSITORY = "Aceship/Arknight-Images" # main -ALT_RESOURCES_REPOSITORY = "yuanyan3060/ArknightsGameResource" # contains zh-cn files # main - -GAMEDATA_LANGUAGE: typing.Mapping[netn.ArknightsServer, str] = { - "en": "en_US", - "jp": "ja_JP", - "kr": "ko_KR", - "cn": "zh_CN", - "bili": "zh_CN", - "tw": "zh_TW", +YOSTAR_GAMEDATA_REPOSITORY = "Kengxxiao/ArknightsGameData_YoStar" # main + +LANGUAGE_PATH: typing.Mapping[netn.ArknightsServer, PathLike] = { + "cn": "ArknightsGameData/zh_CN", + "bili": "ArknightsGameData/zh_CN", + "en": "ArknightsGameData_YoStar/en_US", + "jp": "ArknightsGameData_YoStar/ja_JP", + "kr": "ArknightsGameData_YoStar/ko_KR", } -PathLike = typing.Union[pathlib.Path, str] - async def download_github_file(repository: str, path: str, *, branch: str = "HEAD") -> bytes: """Download a file from github.""" @@ -206,80 +202,33 @@ async def update_repository( class GitAssets(base.Assets): """Game assets client downloaded through 3rd party git repositories.""" - gamedata_directory: pathlib.Path - resources_directory: pathlib.Path - gamedata_repository: str - resources_repository: str + parent_directory: pathlib.Path + """Parent directory of ArknightsGameData and ArnightsGameData_YoStar""" def __init__( self, - gamedata_directory: PathLike | None = None, - resources_directory: PathLike | None = None, - gamedata_repository: str | None = None, - resources_repository: str | None = None, + parent_directory: PathLike | None = None, *, default_server: netn.ArknightsServer = "en", json_loads: typing.Callable[[bytes], typing.Any] = json.loads, ) -> None: super().__init__(default_server=default_server, json_loads=json_loads) - default_directory = pathlib.Path(tempfile.gettempdir()) - - if gamedata_repository: - self.gamedata_repository = gamedata_repository - elif self.default_server == "cn": - self.gamedata_repository = CN_GAMEDATA_REPOSITORY - elif self.default_server == "tw": - self.gamedata_repository = TW_GAMEDATA_REPOSITORY - else: - self.gamedata_repository = GLOBAL_GAMEDATA_REPOSITORY + self.parent_directory = pathlib.Path(parent_directory or netn.APPDATA_DIR) - self.resources_repository = resources_repository or RESOURCES_REPOSITORY - - self.gamedata_directory = pathlib.Path( - gamedata_directory or default_directory / self.gamedata_repository.split("/")[1], - ) - self.resources_directory = pathlib.Path( - resources_directory or default_directory / self.resources_repository.split("/")[1], - ) - - async def update_assets( - self, - resources: bool = False, - *, - force: bool = False, - ) -> None: - """Update game data. - - Only gamedata for the default server is downloaded by default. - """ - await update_repository( - self.gamedata_repository, - self.gamedata_directory, - allow="gamedata/excel/*", - force=force, - ) - - if resources: - await update_repository(self.resources_repository, self.resources_directory, force=force) + async def update_assets(self, *, force: bool = False) -> None: + """Update game data.""" + for repo in (CN_GAMEDATA_REPOSITORY, YOSTAR_GAMEDATA_REPOSITORY): + await update_repository( + repo, + self.parent_directory / repo.split("/")[1], + allow="gamedata/excel/*", + force=force, + ) self.loaded = True def get_file(self, path: str, *, server: netn.ArknightsServer | None = None) -> bytes: """Get an extracted asset file.""" - if "gamedata" in path: - directory = self.gamedata_directory / GAMEDATA_LANGUAGE[server or self.default_server] - else: - directory = self.resources_directory - + directory = self.parent_directory / LANGUAGE_PATH[server or self.default_server] return (directory / path).read_bytes() - - async def aget_file(self, path: str, *, server: netn.ArknightsServer | None = None) -> bytes: - """Get an extracted asset file without requiring load.""" - if "gamedata" in path: - repository = self.gamedata_repository - path = f"{GAMEDATA_LANGUAGE[server or self.default_server]}/{path}" - else: - repository = self.resources_repository - - return await download_github_file(repository, path) diff --git a/arkprts/auth.py b/arkprts/auth.py index 8ad28cc..6302885 100644 --- a/arkprts/auth.py +++ b/arkprts/auth.py @@ -81,7 +81,6 @@ import pathlib import random import string -import tempfile import time import typing import urllib.parse @@ -826,7 +825,7 @@ def __init__( elif isinstance(cache, (pathlib.Path, str)): self.cache_path = pathlib.Path(cache).expanduser() elif cache is None: - self.cache_path = pathlib.Path(tempfile.gettempdir()) / "arkprts_auth_cache.json" + self.cache_path = netn.APPDATA_DIR / "arkprts_auth_cache.json" else: self.cache_path = None self.upcoming_auth = list(cache) diff --git a/arkprts/models/base.py b/arkprts/models/base.py index 2a5045a..28a4677 100644 --- a/arkprts/models/base.py +++ b/arkprts/models/base.py @@ -68,12 +68,14 @@ def _fix_amiya(cls, value: typing.Any, info: pydantic.ValidationInfo) -> typing. """Flatten Amiya to only keep her selected form if applicable.""" if value and value.get("tmpl"): value["variations"] = { - tmplid: cls(value["client"], **{**value, **tmpl}) for tmplid, tmpl in value["tmpl"].items() + tmplid: cls(**{**value, **tmpl, "tmpl": {}, "charId": tmplid}) # pyright: ignore + for tmplid, tmpl in value["tmpl"].items() } # tmplId present in battle replays, sometimes the tmpl for amiya guard is not actually present current_tmpl = value["currentTmpl"] if "currentTmpl" in value else value["tmplId"] current = value["tmpl"].get(current_tmpl, next(iter(value["tmpl"].values()))) value.update(current) + value["charId"] = current_tmpl return value diff --git a/arkprts/network.py b/arkprts/network.py index 8f9ea19..89ae72d 100644 --- a/arkprts/network.py +++ b/arkprts/network.py @@ -27,6 +27,10 @@ import asyncio import json import logging +import os +import pathlib +import platform +import tempfile import typing import aiohttp @@ -65,6 +69,17 @@ "Connection": "Keep-Alive", } +if platform.system() == "Windows": + _localappdata = pathlib.Path(os.getenv("LOCALAPPDATA", "~/AppData/Local")) +elif platform.system() == "Darwin": + _localappdata = pathlib.Path("~/Library/Application Support/") +else: + _localappdata = pathlib.Path(os.getenv("XDG_DATA_HOME", "~/.local/share")) + +APPDATA_DIR = _localappdata.expanduser() / "arkprts" +TEMP_DIR = pathlib.Path(tempfile.gettempdir()) / "arkprts" + + # aiohttp uses a very noisy library _charset_normalizer_logger = logging.getLogger("charset_normalizer") _charset_normalizer_logger.setLevel(logging.INFO) diff --git a/requirements.txt b/requirements.txt index a44f367..468b4a5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,4 +5,3 @@ rsa pycryptodome UnityPy bson -Pillow diff --git a/setup.py b/setup.py index a42ddfb..dbaf340 100644 --- a/setup.py +++ b/setup.py @@ -13,10 +13,10 @@ package_data={"arkprts": ["py.typed"]}, install_requires=["aiohttp", "pydantic==2.*"], extras_require={ - "all": ["rsa", "pycryptodome", "UnityPy", "Pillow", "bson"], + "all": ["rsa", "pycryptodome", "UnityPy", "bson"], "rsa": ["rsa"], "aes": ["pycryptodome"], - "assets": ["UnityPy", "pycryptodome", "Pillow", "bson"], + "assets": ["UnityPy", "pycryptodome", "bson"], }, long_description=pathlib.Path("README.md").read_text(encoding="utf-8"), long_description_content_type="text/markdown",