refactor: use ark_fbs

.github/actions/setup-python/action.yml

@@ -5,7 +5,7 @@ inputs:
   python-version:
     description: Python version
     required: false
-    default: "3.12"
+    default: "3.13"
 
 runs:
   using: "composite"

AGENTS.md

@@ -41,13 +41,12 @@ Common flags:
 - `storage/asset/gamedata/`: decoded game data
 - `storage/asset/raw/`: raw extracted assets
 - `OpenArknightsFBS/FBS/`: flatbuffer schemas
-- `bin/flatc` (`bin/flatc.exe` on Windows): flatc binary
 
 ## Config (Env)
 
 - `ENVIRONMENT`, `LOG_LEVEL`, `TIMEOUT`
 - `TOKEN`, `BACKEND_ENDPOINT` (for upload-related tasks, e.g. `ItemDemand`)
-- `FLATC_PATH`, `SENTRY_DSN`
+- `SENTRY_DSN`
 
 Use `BACKEND_ENDPOINT` (not `ENDPOINT`).
 

Dockerfile

@@ -1,6 +1,6 @@
 # syntax=docker/dockerfile:1
 
-FROM python:3.12-bookworm AS requirements-stage
+FROM python:3.13-bookworm AS requirements-stage
 
 WORKDIR /tmp
 
@@ -12,15 +12,15 @@ COPY ./pyproject.toml ./uv.lock* /tmp/
 
 RUN uv export --format requirements.txt -o requirements.txt --no-editable --no-hashes --no-dev --no-emit-project
 
-FROM python:3.12-bookworm AS build-stage
+FROM python:3.13-bookworm AS build-stage
 
 WORKDIR /wheel
 
 COPY --from=requirements-stage /tmp/requirements.txt /wheel/requirements.txt
 
 RUN pip wheel --wheel-dir=/wheel --no-cache-dir --requirement /wheel/requirements.txt
 
-FROM python:3.12-bookworm AS metadata-stage
+FROM python:3.13-bookworm AS metadata-stage
 
 WORKDIR /tmp
 
@@ -29,7 +29,7 @@ RUN --mount=type=bind,source=./.git/,target=/tmp/.git/ \
   || git rev-parse --short HEAD > /tmp/VERSION \
   && echo "Building version: $(cat /tmp/VERSION)"
 
-FROM python:3.12-slim-bookworm
+FROM python:3.13-slim-bookworm
 
 WORKDIR /app
 
@@ -55,6 +55,4 @@ COPY --from=metadata-stage /tmp/VERSION /app/VERSION
 
 COPY . /app/
 
-RUN chmod -R +x /app/bin
-
 ENTRYPOINT ["python", "-m", "torappu"]

README.md

@@ -13,7 +13,7 @@ An unpacker for Arknights assets with a focus on resource extraction and analysi
 
 ## Requirements
 
-- Python 3.12+
+- Python 3.13+
 - Dependencies as specified in pyproject.toml
 
 ## Installation
@@ -76,7 +76,6 @@ docker run -e TOKEN=your_token -v $(pwd)/storage:/app/storage torappu [CLIENT_VE
   - `core/`: Core functionality
 - `OpenArknightsFBS/`: FlatBuffer schema definitions
 - `assets/`: Asset resources
-- `bin/`: Binary tools (includes flatc for FlatBuffer compilation)
 - `scripts/`: Utility scripts
 - `storage/`: Storage for extracted assets
 

pyproject.toml

@@ -31,7 +31,7 @@ dev = ["ruff>=0.15.0,<1", "pre-commit>=4.0.0,<5"]
 
 [tool.ruff]
 line-length = 88
-target-version = "py312"
+target-version = "py313"
 
 [tool.ruff.format]
 line-ending = "lf"
@@ -75,7 +75,7 @@ keep-runtime-typing = true
 
 [tool.pyright]
 extraPaths = ["./"]
-pythonVersion = "3.12"
+pythonVersion = "3.13"
 pythonPlatform = "All"
 executionEnvironments = [{ root = "./" }]
 

torappu/config.py

@@ -1,19 +1,7 @@
-from pathlib import Path
 from typing import Literal
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from torappu.consts import MACOS, WINDOWS
-
-
-def get_flatc_path():
-    if WINDOWS:
-        return Path("bin/flatc.exe")
-    elif MACOS:
-        return Path("bin/macos/flatc")
-    else:
-        return Path("bin/flatc")
-
 
 class Config(BaseSettings):
     model_config = SettingsConfigDict(
@@ -31,7 +19,6 @@ class Config(BaseSettings):
     timeout: int = 10
 
     backend_endpoint: str | None = None
-    flatc_path: Path = get_flatc_path()
 
     sentry_dsn: str | None = None
 

torappu/consts.py

@@ -17,6 +17,8 @@
 GAMEDATA_DIR = STORAGE_DIR / "asset" / "gamedata"
 HOT_UPDATE_LIST_DIR = STORAGE_DIR / "hot_update_list"
 
+RESOURCE_MANIFEST_IDX_NAME = "resource_manifest_idx.json"
+
 HEADERS = {
     "user-agent": "Dalvik/2.1.0 (Linux; U; Android 6.0.1; vivo X9L Build/MMB29M)"
 }

torappu/core/client.py

@@ -1,15 +1,15 @@
 import asyncio
 import json
-import subprocess
 from hashlib import md5
 from io import BytesIO
 from pathlib import Path
-from tempfile import TemporaryDirectory
 from zipfile import ZipFile
 
 import anyio
 import httpx
 import UnityPy
+from ark_fbs import Options as FBOptions
+from ark_fbs import Schema as FBSchema
 from tenacity import retry, wait_random_exponential
 from UnityPy.classes import MonoBehaviour
 
@@ -20,12 +20,19 @@
     HG_CN_BASEURL,
     HOT_UPDATE_LIST_DIR,
     PRE_RESOLVE_PATHS,
+    RESOURCE_MANIFEST_IDX_NAME,
     STORAGE_DIR,
 )
 from torappu.core.utils.path import hg_normalize_url
 from torappu.log import logger
 from torappu.models import ABInfo, Diff, HotUpdateInfo, Version
 
+resource_manifest_schema: FBSchema = FBSchema.from_fbs_file(
+    "assets/ResourceManifest.fbs",
+    include_paths=["assets"],
+    options=FBOptions(),
+)
+
 
 class Client:
     def __init__(
@@ -52,12 +59,7 @@ async def init(self):
             self.prev_hot_update_list = None
         if self.hot_update_list.manifest_name is not None:
             idx_path = await self.fetch_asset_bundle(self.hot_update_list.manifest_name)
-            self.load_idx(
-                idx_path,
-                GAMEDATA_DIR.joinpath(
-                    self.version.res_version, self.hot_update_list.manifest_name
-                ),
-            )
+            self.load_idx(idx_path, self.hot_update_list.manifest_name)
         else:
             await self.load_torappu_index()
 
@@ -246,30 +248,25 @@ async def load_torappu_index(self):
                 for item in torappu_index.assetToBundleList  # type: ignore
             }
 
-    def load_idx(self, idx_path: str, decoded_path: Path):
-        tmp_dir = TemporaryDirectory()
-        tmp_path = Path(tmp_dir.name)
+    def load_idx(self, idx_path: str, manifest_name: str):
         idx = Path(idx_path).read_bytes()
-        flatbuffer_data_path = tmp_path / "idx.bin"
-        flatbuffer_data_path.write_bytes(idx[128:])
-        params = [
-            self.config.flatc_path,
-            "-o",
-            decoded_path.resolve(),
-            "--no-warnings",
-            "--json",
-            "--strict-json",
-            "--natural-utf8",
-            "--defaults-json",
-            "--raw-binary",
-            "assets/ResourceManifest.fbs",
-            "--",
-            flatbuffer_data_path,
-        ]
-        subprocess.run(params)
-        flatbuffer_data_path.unlink()
-        json_path = decoded_path / "idx.json"
-        jsons = json.loads(json_path.read_text(encoding="utf-8"))
+        flatbuffer_data = idx[128:]
+        decoded_path = GAMEDATA_DIR.joinpath(
+            self.version.res_version, RESOURCE_MANIFEST_IDX_NAME
+        )
+        decoded_path.parent.mkdir(parents=True, exist_ok=True)
+
+        try:
+            jsons = json.loads(resource_manifest_schema.binary_to_json(flatbuffer_data))
+            decoded_path.write_text(
+                json.dumps(jsons, ensure_ascii=False, separators=(",", ":")),
+                encoding="utf-8",
+            )
+        except Exception as exc:
+            raise RuntimeError(
+                f"failed to decode idx flatbuffer from {idx_path!r}"
+            ) from exc
+
         self.asset_to_bundle = {
             item["assetName"]: jsons["bundles"][item["bundleIndex"]]["name"]
             for item in jsons["assetToBundleList"]

torappu/core/tasks/gamedata.py

@@ -3,13 +3,13 @@
 import json
 import os
 import platform
-import subprocess
 from pathlib import Path
-from tempfile import TemporaryDirectory
 from typing import ClassVar
 
 import bson
 import UnityPy
+from ark_fbs import Options as FBOptions
+from ark_fbs import Schema as FBSchema
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import unpad
 from UnityPy.classes import TextAsset
@@ -95,6 +95,7 @@
 plaintexts = ["levels/levels_meta.json", "data_version.txt"]
 signed_list = ["excel", "_table", "[uc]lua"]
 chat_mask = "UITpAi82pHAWwnzqHRMCwPonJLIB3WCl"
+flatbuffer_schema_cache: dict[str, FBSchema] = {}
 
 
 class Task(BaseTask):
@@ -138,35 +139,33 @@ def _get_flatbuffer_output_name(self, relative_path: str, fb_name: str) -> str:
             return source_name
         return fb_name
 
+    def _get_flatbuffer_schema(self, fb_name: str) -> FBSchema:
+        schema = flatbuffer_schema_cache.get(fb_name)
+        if schema is not None:
+            return schema
+
+        schema_path = FBS_DIR / f"{fb_name}.fbs"
+        schema = FBSchema.from_fbs_file(
+            str(schema_path),
+            include_paths=[str(FBS_DIR)],
+            options=FBOptions(),
+        )
+        flatbuffer_schema_cache[fb_name] = schema
+        return schema
+
     @run_sync
     def _decode_flatbuffer(self, path: str, obj: TextAsset, fb_name: str):
-        tmp_dir = TemporaryDirectory()
-        tmp_path = Path(tmp_dir.name)
         relative_path = path.replace("dyn/gamedata/", "")
         output_name = self._get_flatbuffer_output_name(relative_path, fb_name)
+        flatbuffer_data = m_script_to_bytes(obj.m_Script)[128:]
+        try:
+            schema = self._get_flatbuffer_schema(fb_name)
+            jsons = json.loads(schema.binary_to_json(flatbuffer_data))
+        except Exception as exc:
+            raise RuntimeError(
+                f"failed to decode flatbuffer {fb_name!r} for asset {path!r}"
+            ) from exc
 
-        flatbuffer_data_path = tmp_path.joinpath(f"{output_name}.bytes")
-        output_path = tmp_path.joinpath(os.path.dirname(relative_path))
-        flatbuffer_data_path.write_bytes(m_script_to_bytes(obj.m_Script)[128:])
-
-        params = [
-            self.client.config.flatc_path,
-            "-o",
-            output_path.resolve(),
-            "--no-warnings",
-            "--json",
-            "--strict-json",
-            "--natural-utf8",
-            "--defaults-json",
-            "--raw-binary",
-            f"{FBS_DIR}/{fb_name}.fbs",
-            "--",
-            flatbuffer_data_path.resolve(),
-        ]
-        subprocess.run(params)
-        flatbuffer_data_path.unlink()
-        json_path = output_path / f"{output_name}.json"
-        jsons = json.loads(json_path.read_text(encoding="utf-8"))
         if fb_name == "activity_table":
             for k, v in jsons["dynActs"].items():
                 if "base64" in v:
@@ -190,8 +189,6 @@ def _decode_flatbuffer(self, path: str, obj: TextAsset, fb_name: str):
             encoding="utf-8",
         )
 
-        tmp_dir.cleanup()
-
     @run_sync
     def _decrypt(self, path: str, obj: TextAsset, is_signed: bool):
         key: bytes = chat_mask[:16].encode()