From fbdc97c1648d57a44e2453cccbc7b3dc5836d941 Mon Sep 17 00:00:00 2001 From: ritenseki Date: Wed, 5 Aug 2026 01:10:07 +0800 Subject: [PATCH] tools: rebuild the asset dump toolchain in-repo The previous toolchain (C:\Users\riten\Desktop\UmaCrack\) was deleted, taking with it the only way to read authoritative field names out of the game bundles. Reimplemented from the decryption logic already in Assets/Scripts. - uma_common.py key constants, bundle XOR, MetaDb (path->hash, UnityPy load) - dump_meta.py decrypt meta -> meta_plain.db; find / info / songs - dump_bundle.py list / tree / json / save / gameobjects for any bundle - dump_cutt_typetree.py authoritative LiveTimelineWorkSheet schema per song - scan_live_tracks.py all 59 songs -> out/scan_{keys,groups}.csv, scan_summary.txt cutt bundles ship full TypeTrees, so field names and keyframe counts read back out of them are ground truth. All 70 worksheets across all 59 songs share one 128-field schema. Needs UnityPy + apsw-sqlite3mc (not plain apsw); see tools/requirements.txt. Generated output under tools/out/ is gitignored (meta_plain.db alone is 151 MB). Co-Authored-By: Claude Opus 5 --- tools/.gitignore | 4 + tools/dump_bundle.py | 158 +++++++++++++ tools/dump_cutt_typetree.py | 187 ++++++++++++++++ tools/dump_meta.py | 171 ++++++++++++++ tools/requirements.txt | 11 + tools/scan_live_tracks.py | 199 ++++++++++++++++ tools/uma_common.py | 436 ++++++++++++++++++++++++++++++++++++ 7 files changed, 1166 insertions(+) create mode 100644 tools/.gitignore create mode 100644 tools/dump_bundle.py create mode 100644 tools/dump_cutt_typetree.py create mode 100644 tools/dump_meta.py create mode 100644 tools/requirements.txt create mode 100644 tools/scan_live_tracks.py create mode 100644 tools/uma_common.py diff --git a/tools/.gitignore b/tools/.gitignore new file mode 100644 index 0000000..235bfbf --- /dev/null +++ b/tools/.gitignore @@ -0,0 +1,4 @@ +# generated output of the asset tools (meta_plain.db, scan_*.csv, ...) +out/ +__pycache__/ +*.pyc diff --git a/tools/dump_bundle.py b/tools/dump_bundle.py new file mode 100644 index 0000000..90cc2f6 --- /dev/null +++ b/tools/dump_bundle.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +"""Decrypt any game asset bundle and inspect it with UnityPy. + +Given an asset path from the meta DB (e.g. ``3d/effect/live/pfb_eff_live1157``) +this locates the hashed file under ``dat/``, undoes the XOR cipher and hands +the result to UnityPy. + +Usage +----- + PY=~/.venvs/umatools/bin/python + + # list every object in the bundle + $PY tools/dump_bundle.py list 3d/env/live/live1014/pfb_env_live1014_controller000 + + # object names of one type only + $PY tools/dump_bundle.py list cutt/cutt_son1157/type_01 --type MonoBehaviour + + # TypeTree schema of an object (by path_id, or the first of a type) + $PY tools/dump_bundle.py tree cutt/cutt_son1157/type_01 --type MonoBehaviour --depth 2 + + # deserialized object as JSON + $PY tools/dump_bundle.py json cutt/cutt_son1157/type_01 -o /tmp/ws.json + + # write the decrypted bundle itself (loadable by UnityPy/AssetStudio) + $PY tools/dump_bundle.py save cutt/cutt_son1157/type_01 -o /tmp/type_01.unity3d + + # game object hierarchy of a prefab bundle + $PY tools/dump_bundle.py gameobjects 3d/env/live/live1014/pfb_env_live1014_controller000 + +``--deps`` also loads the bundle's prerequisites, which is needed when an +object references assets living in another bundle. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import uma_common as uc # noqa: E402 + + +def pick(env, args): + """Objects selected by --type / --path-id, in file order.""" + objs = list(env.objects) + if args.path_id is not None: + objs = [o for o in objs if o.path_id == args.path_id] + if args.type: + objs = [o for o in objs if o.type.name.lower() == args.type.lower()] + return objs + + +def obj_name(obj) -> str: + try: + d = obj.read_typetree() + return d.get("m_Name") or "" + except Exception: + return "" + + +def node_lines(node, depth=0, max_depth=99): + yield "%s%-46s %s" % (" " * depth, node.m_Type, node.m_Name) + if depth >= max_depth: + return + for ch in getattr(node, "m_Children", []) or []: + yield from node_lines(ch, depth + 1, max_depth) + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("cmd", choices=("list", "tree", "json", "save", "gameobjects")) + ap.add_argument("name", help="asset path as stored in the meta DB") + ap.add_argument("--type", default=None, help="filter by Unity class name (MonoBehaviour, Material, ...)") + ap.add_argument("--path-id", type=int, default=None) + ap.add_argument("--depth", type=int, default=99, help="depth limit for `tree`") + ap.add_argument("--deps", action="store_true", help="also load the bundle's dependencies") + ap.add_argument("-o", "--output", default=None) + uc.add_common_args(ap) + args = ap.parse_args(argv) + + uc.require("apsw-sqlite3mc", "UnityPy") + + with uc.MetaDb(args.game_path, args.region) as db: + entry = db.get(args.name) + if entry is None: + print("no meta entry named %s" % args.name, file=sys.stderr) + return 1 + + if args.cmd == "save": + out = args.output or (os.path.basename(entry.name) + ".unity3d") + with open(out, "wb") as fh: + fh.write(db.bundle_bytes(entry)) + print("wrote %s (%d bytes, key=%d)" % (out, os.path.getsize(out), entry.key)) + return 0 + + env = db.load(entry, with_deps=args.deps) + + if args.cmd == "list": + counts: dict[str, int] = {} + for o in pick(env, args): + counts[o.type.name] = counts.get(o.type.name, 0) + 1 + print("%-22s %-22d %s" % (o.type.name, o.path_id, obj_name(o))) + print("-- " + ", ".join("%s=%d" % kv for kv in sorted(counts.items())), file=sys.stderr) + return 0 + + if args.cmd == "gameobjects": + for o in env.objects: + if o.type.name != "GameObject": + continue + try: + d = o.read_typetree() + except Exception: + continue + print("%-22d %-40s active=%s components=%d" % ( + o.path_id, d.get("m_Name", ""), d.get("m_IsActive"), len(d.get("m_Component", [])))) + return 0 + + objs = pick(env, args) + if not objs: + print("no matching object", file=sys.stderr) + return 1 + + if args.cmd == "tree": + for o in objs: + node = getattr(getattr(o, "serialized_type", None), "node", None) + print("=" * 90) + print("%s path_id=%d %s" % (o.type.name, o.path_id, obj_name(o))) + print("=" * 90) + if node is None: + print(" (bundle carries no TypeTree for this object)") + continue + for line in node_lines(node, 0, args.depth): + print(line) + return 0 + + # json + data = [] + for o in objs: + try: + data.append(o.read_typetree()) + except Exception as exc: + print("skip %s %d: %s" % (o.type.name, o.path_id, exc), file=sys.stderr) + payload = data[0] if len(data) == 1 else data + text = json.dumps(payload, indent=1, ensure_ascii=False, default=str) + if args.output: + with open(args.output, "w", encoding="utf-8") as fh: + fh.write(text) + print("wrote %s (%d object(s))" % (args.output, len(data))) + else: + print(text) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/dump_cutt_typetree.py b/tools/dump_cutt_typetree.py new file mode 100644 index 0000000..45d510f --- /dev/null +++ b/tools/dump_cutt_typetree.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Dump the real serialized field names of `LiveTimelineWorkSheet` from a +live (cutt) asset bundle, together with how much data each track carries. + +Why this exists +--------------- +The C# classes in ``Assets/Scripts/umamusume/Gallop/Live/Cutt/`` must declare +fields whose names match the game's serialization *exactly* (case sensitive), +otherwise Unity silently deserializes nothing. The game's cutt bundles ship +with full TypeTrees, so the authoritative names can simply be read back out. + +Where the worksheets live +------------------------- +Each song has a bundle group ``cutt/cutt_son/``: + + cutt_son prefab bundle (no worksheets) + data LiveTimelineData (no worksheets) + son_camera WorkSheet "son_Camera" + type_01 .. type_10 WorkSheet "Type_01" .. (only on newer songs) + +A MonoBehaviour is treated as a worksheet when its TypeTree contains the field +``cameraPosKeys``. + +Usage +----- + PY=~/.venvs/umatools/bin/python + + # every worksheet of a song: field table with keyframe counts + $PY tools/dump_cutt_typetree.py 1157 + + # just the schema (field type + name), no bundle data needed + $PY tools/dump_cutt_typetree.py 1157 --fields + + # only tracks that actually contain keys, one worksheet + $PY tools/dump_cutt_typetree.py 1157 --sheet Type_01 --nonzero + + # nested schema of one track -- this is how you learn the key class fields + $PY tools/dump_cutt_typetree.py 1157 --tree laserList + + # first keyframe of a track, as JSON (real values) + $PY tools/dump_cutt_typetree.py 1157 --sample laserList + + # whole worksheet as JSON + $PY tools/dump_cutt_typetree.py 1157 --sheet Type_01 --json out.json + + # any bundle path also works instead of a song id + $PY tools/dump_cutt_typetree.py cutt/cutt_son1001/son1001_camera +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import uma_common as uc # noqa: E402 + + +def bundles_for(db, target: str) -> list: + """Resolve a song id or an explicit asset path into meta entries.""" + if re.fullmatch(r"\d+", target): + entries = db.like("cutt/cutt_son%s/%%" % target) + if not entries: + raise SystemExit("no cutt bundles for song id %s" % target) + # the prefab/data bundles never hold worksheets - skip them cheaply + return [e for e in entries if not e.name.endswith(("/data", "/cutt_son%s" % target))] + e = db.get(target) + if e is None: + raise SystemExit("no meta entry named %s" % target) + return [e] + + +def node_lines(node, depth: int = 0, max_depth: int = 99): + yield "%s%-46s %s" % (" " * depth, node.m_Type, node.m_Name) + if depth >= max_depth: + return + for ch in getattr(node, "m_Children", []) or []: + yield from node_lines(ch, depth + 1, max_depth) + + +def field_nodes(root): + """The WorkSheet's own fields, minus the four MonoBehaviour base fields.""" + skip = {"m_GameObject", "m_Enabled", "m_Script", "m_Name"} + return [ch for ch in root.m_Children if ch.m_Name not in skip] + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("target", help="song id (e.g. 1157) or full asset path") + ap.add_argument("--sheet", default=None, help="only this worksheet (e.g. Type_01, son1157_Camera)") + ap.add_argument("--fields", action="store_true", help="print the schema only, skip deserialization") + ap.add_argument("--nonzero", action="store_true", help="hide tracks with no keyframes") + ap.add_argument("--tree", metavar="FIELD", help="print the nested TypeTree of one field") + ap.add_argument("--tree-depth", type=int, default=99, help="depth limit for --tree") + ap.add_argument("--sample", metavar="FIELD", help="print the first keyframe of one field as JSON") + ap.add_argument("--json", metavar="PATH", help="write the whole worksheet as JSON") + ap.add_argument("--groups", action="store_true", help="also list the group names of each track") + uc.add_common_args(ap) + args = ap.parse_args(argv) + + uc.require("apsw-sqlite3mc", "UnityPy") + + with uc.MetaDb(args.game_path, args.region) as db: + entries = bundles_for(db, args.target) + found_any = False + for entry in entries: + try: + env = db.load(entry) + except FileNotFoundError as exc: + print("skip %s: %s" % (entry.name, exc), file=sys.stderr) + continue + for name, obj, root in uc.worksheets(env): + if args.sheet and name.lower() != args.sheet.lower(): + continue + found_any = True + fields = field_nodes(root) + print("=" * 100) + print("%s (%s) %d fields" % (name, entry.name, len(fields))) + print("=" * 100) + + if args.tree: + node = next((f for f in fields if f.m_Name.lower() == args.tree.lower()), None) + if node is None: + print(" no such field: %s" % args.tree) + continue + for line in node_lines(node, 0, args.tree_depth): + print(line) + continue + + if args.fields: + for i, f in enumerate(fields): + print("%3d %-50s %s" % (i, f.m_Type, f.m_Name)) + continue + + tree = obj.read_typetree() + + if args.sample: + key = next((f.m_Name for f in fields if f.m_Name.lower() == args.sample.lower()), None) + if key is None: + print(" no such field: %s" % args.sample) + continue + val = tree.get(key) + if isinstance(val, list) and val: + val = val[0] + if isinstance(val, dict) and isinstance(val.get("thisList"), list): + val = dict(val, thisList=val["thisList"][:1]) + if isinstance(val, dict) and isinstance(val.get("keys"), dict): + kl = val["keys"] + if isinstance(kl.get("thisList"), list): + val = dict(val, keys=dict(kl, thisList=kl["thisList"][:1])) + print(json.dumps(val, indent=2, ensure_ascii=False, default=str)) + continue + + if args.json: + with open(args.json, "w", encoding="utf-8") as fh: + json.dump(tree, fh, indent=1, ensure_ascii=False, default=str) + print("wrote %s" % args.json) + continue + + print("%3s %-50s %-46s %7s %7s" % ("#", "serialized type", "field name", "groups", "keys")) + print("-" * 100) + total = 0 + for i, f in enumerate(fields): + val = tree.get(f.m_Name) + keys = uc.count_keyframes(val) + groups = uc.count_groups(val) + total += keys + if args.nonzero and keys == 0 and groups == 0: + continue + print("%3d %-50s %-46s %7d %7d" % (i, f.m_Type, f.m_Name, groups, keys)) + if args.groups: + for gn in uc.group_names(val): + print("%sname: %s" % (" " * 8, gn)) + print("-" * 100) + print("total keyframes: %d" % total) + if not found_any: + print("no LiveTimelineWorkSheet found", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/dump_meta.py b/tools/dump_meta.py new file mode 100644 index 0000000..673cc39 --- /dev/null +++ b/tools/dump_meta.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""Decrypt and query the game's `meta` database. + +The meta DB maps an asset path (``cutt/cutt_son1157/type_01``) to the hashed +file name under ``dat/`` plus the per-bundle XOR key. It is a SQLite database +encrypted with SQLite3 Multiple Ciphers (ChaCha20); see ``uma_common.py`` for +the key derivation. + +Usage +----- + PY=~/.venvs/umatools/bin/python + + # write a plaintext copy you can open with any sqlite client + $PY tools/dump_meta.py plain # -> tools/out/meta_plain.db + $PY tools/dump_meta.py plain -o /tmp/meta.db + + # search asset paths (SQL LIKE pattern, % is the wildcard) + $PY tools/dump_meta.py find 'cutt/cutt_son1157/%' + $PY tools/dump_meta.py find '%live1014%' --limit 20 + + # everything known about one asset, including where it lives on disk + $PY tools/dump_meta.py info cutt/cutt_son1157/type_01 + + # list the live song ids that have cutt bundles + $PY tools/dump_meta.py songs + +Table ``a`` columns (see Assets/Scripts/UmaDatabase/UmaDatabaseEntry.cs): + i row id n asset path d ';'-joined dependencies + g ? l length c checksum + h hashed filename m asset type e per-bundle cipher key +""" + +from __future__ import annotations + +import argparse +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import uma_common as uc # noqa: E402 + +DEFAULT_OUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "out") + + +def cmd_plain(args) -> int: + """Copy the decrypted meta DB into a plaintext SQLite file. + + ``sqlite3_backup`` and ``VACUUM INTO`` both refuse / re-encrypt here (the + ChaCha20 codec reserves bytes per page, so source and target page layouts + are incompatible), so the schema and rows are copied explicitly. + """ + uc.require("apsw-sqlite3mc") + import apsw + + out = args.output or os.path.join(DEFAULT_OUT_DIR, "meta_plain.db") + os.makedirs(os.path.dirname(os.path.abspath(out)), exist_ok=True) + for suffix in ("", "-journal", "-wal", "-shm"): + if os.path.exists(out + suffix): + os.remove(out + suffix) + + with uc.MetaDb(args.game_path, args.region) as db: + dest = apsw.Connection(out) + try: + schema = list( + db.con.execute( + "SELECT type,name,sql FROM sqlite_master " + "WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite\\_%' ESCAPE '\\'" + ) + ) + tables = [(n, sql) for t, n, sql in schema if t == "table"] + indexes = [(n, sql) for t, n, sql in schema if t == "index"] + total = 0 + for name, sql in tables: + dest.execute(sql) + rows = list(db.con.execute('SELECT * FROM "%s"' % name)) + if rows: + ph = ",".join("?" * len(rows[0])) + with dest: + dest.executemany('INSERT INTO "%s" VALUES (%s)' % (name, ph), rows) + total += len(rows) + print(" %-16s %8d rows" % (name, len(rows))) + for name, sql in indexes: + try: + dest.execute(sql) + except Exception as exc: # a stale index definition must not kill the dump + print(" index %s skipped (%s)" % (name, exc), file=sys.stderr) + finally: + dest.close() + print("wrote %s (%d rows total, %.1f MB)" % (out, total, os.path.getsize(out) / 1048576.0)) + return 0 + + +def cmd_find(args) -> int: + pattern = args.pattern if "%" in args.pattern else "%" + args.pattern + "%" + with uc.MetaDb(args.game_path, args.region) as db: + rows = db.like(pattern) + if args.limit: + rows = rows[: args.limit] + for e in rows: + on_disk = "" if os.path.isfile(e.path(db.game_path)) else " [NOT DOWNLOADED]" + print("%-60s %-34s key=%-22d%s" % (e.name, e.url, e.key, on_disk)) + print("-- %d entr%s" % (len(rows), "y" if len(rows) == 1 else "ies"), file=sys.stderr) + return 0 + + +def cmd_info(args) -> int: + with uc.MetaDb(args.game_path, args.region) as db: + e = db.get(args.name) + if e is None: + print("no such asset: %s" % args.name, file=sys.stderr) + return 1 + p = e.path(db.game_path) + print("name : %s" % e.name) + print("type : %s" % e.type) + print("hash (url) : %s" % e.url) + print("path : %s%s" % (p, "" if os.path.isfile(p) else " [NOT DOWNLOADED]")) + if os.path.isfile(p): + print("size : %d bytes" % os.path.getsize(p)) + print("checksum : %s" % e.checksum) + print("key : %d (%s)" % (e.key, "encrypted" if e.is_encrypted else "plaintext")) + if e.is_encrypted: + print("fkey : %s" % uc.build_fkey(e.key).hex()) + deps = e.deps() + print("dependencies : %d" % len(deps)) + for d in deps: + print(" %s" % d) + return 0 + + +def cmd_songs(args) -> int: + with uc.MetaDb(args.game_path, args.region) as db: + ids = db.song_ids() + for sid in ids: + entries = db.like("cutt/cutt_son%s/%%" % sid) + missing = sum(0 if os.path.isfile(e.path(db.game_path)) else 1 for e in entries) + print("son%-8s %2d bundles%s" % (sid, len(entries), " (%d missing)" % missing if missing else "")) + print("-- %d songs" % len(ids), file=sys.stderr) + return 0 + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = ap.add_subparsers(dest="cmd", required=True) + + p = sub.add_parser("plain", help="write a decrypted copy of the meta DB") + p.add_argument("-o", "--output", default=None, help="output path (default: tools/out/meta_plain.db)") + uc.add_common_args(p) + p.set_defaults(func=cmd_plain) + + p = sub.add_parser("find", help="search asset paths with a LIKE pattern") + p.add_argument("pattern") + p.add_argument("--limit", type=int, default=0) + uc.add_common_args(p) + p.set_defaults(func=cmd_find) + + p = sub.add_parser("info", help="show one asset's meta row and disk location") + p.add_argument("name") + uc.add_common_args(p) + p.set_defaults(func=cmd_info) + + p = sub.add_parser("songs", help="list live song ids that have cutt bundles") + uc.add_common_args(p) + p.set_defaults(func=cmd_songs) + + args = ap.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/requirements.txt b/tools/requirements.txt new file mode 100644 index 0000000..4e8e1c8 --- /dev/null +++ b/tools/requirements.txt @@ -0,0 +1,11 @@ +# Python dependencies for the UmaViewer asset tools (tools/*.py). +# +# python3 -m venv ~/.venvs/umatools +# ~/.venvs/umatools/bin/python -m pip install -r tools/requirements.txt +# +# NOTE: it must be `apsw-sqlite3mc`, NOT plain `apsw`. Only that wheel bundles +# SQLite3 Multiple Ciphers, which is what decrypts the game's `meta` database +# (the same library the C# side loads as sqlite3mc_x64.dll). + +UnityPy>=1.20 +apsw-sqlite3mc>=3.46 diff --git a/tools/scan_live_tracks.py b/tools/scan_live_tracks.py new file mode 100644 index 0000000..c82a13c --- /dev/null +++ b/tools/scan_live_tracks.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Scan every live song and report how many keyframes each timeline track has. + +This is the "which track should I implement next" tool. It walks all +``cutt/cutt_son/`` bundle groups, deserializes every LiveTimelineWorkSheet +inside them, and counts the keyframes of each of the ~124 track fields. + +Outputs (default directory: tools/out/) +--------------------------------------- + scan_keys.csv song x track matrix, cell = number of keyframes + scan_groups.csv song x track matrix, cell = number of track groups + (list length for grouped tracks, keyframe count for + plain key lists) + scan_summary.txt per-track totals, sorted by how many songs use the track + scan.json everything above in machine readable form, plus the + per-song worksheet inventory and the serialized type of + each field + +Aggregation +----------- +A song usually ships several worksheets (``son_Camera`` plus ``Type_01`` .. +``Type_10``); the type sheets are variations of one another. By default a +song's value for a track is the MAX over its worksheets, so ten near-identical +sheets do not inflate the number. ``--agg sum`` and ``--per-sheet`` are +available if you want the raw numbers. + +Usage +----- + PY=~/.venvs/umatools/bin/python + + $PY tools/scan_live_tracks.py # everything, to tools/out/ + $PY tools/scan_live_tracks.py --songs 1157 1001 # only these songs + $PY tools/scan_live_tracks.py --per-sheet # one row per worksheet + $PY tools/scan_live_tracks.py --out-dir /tmp/x --quiet + + # what the summary looks like: + # track field songs total keys max example songs + # blinkLightList 56 12345 410 1157, 1156, ... +""" + +from __future__ import annotations + +import argparse +import csv +import json +import os +import sys +import time +from collections import OrderedDict + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import uma_common as uc # noqa: E402 + +DEFAULT_OUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "out") + +BASE_FIELDS = {"m_GameObject", "m_Enabled", "m_Script", "m_Name"} +# scalar header fields that are not tracks +SCALAR_FIELDS = { + "version", "targetCameraIndex", "enableAtRuntime", "enableAtEdit", + "TotalTimeLength", "Lyrics", "SheetType", "SheetVariationId", "IsVariationSheet", +} + + +def scan_song(db, song_id: str, verbose: bool = True): + """Return ``{sheet_name: {field: (groups, keys)}}`` plus field types.""" + sheets = OrderedDict() + types: dict[str, str] = {} + entries = db.like("cutt/cutt_son%s/%%" % song_id) + for entry in entries: + if entry.name.endswith(("/data", "/cutt_son%s" % song_id)): + continue # never carry worksheets, skip the decrypt cost + try: + env = db.load(entry) + except FileNotFoundError as exc: + if verbose: + print(" skip %s (%s)" % (entry.name, exc), file=sys.stderr) + continue + except Exception as exc: + print(" FAILED %s: %s" % (entry.name, exc), file=sys.stderr) + continue + for name, obj, root in uc.worksheets(env): + try: + tree = obj.read_typetree() + except Exception as exc: + print(" FAILED typetree %s/%s: %s" % (entry.name, name, exc), file=sys.stderr) + continue + row: dict[str, tuple[int, int]] = {} + for ch in root.m_Children: + f = ch.m_Name + if f in BASE_FIELDS or f in SCALAR_FIELDS: + continue + types.setdefault(f, ch.m_Type) + val = tree.get(f) + row[f] = (uc.count_groups(val), uc.count_keyframes(val)) + sheets["%s|%s" % (song_id, name)] = row + return sheets, types + + +def write_matrix(path: str, rows: "OrderedDict[str, dict]", fields: list, idx: int) -> None: + with open(path, "w", newline="", encoding="utf-8") as fh: + w = csv.writer(fh) + w.writerow(["song"] + fields) + for row_name, values in rows.items(): + w.writerow([row_name] + [values.get(f, (0, 0))[idx] for f in fields]) + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--songs", nargs="*", default=None, help="song ids to scan (default: all)") + ap.add_argument("--out-dir", default=DEFAULT_OUT_DIR) + ap.add_argument("--agg", choices=("max", "sum"), default="max", help="how to fold a song's worksheets") + ap.add_argument("--per-sheet", action="store_true", help="keep one row per worksheet instead of per song") + ap.add_argument("--quiet", action="store_true") + uc.add_common_args(ap) + args = ap.parse_args(argv) + + uc.require("apsw-sqlite3mc", "UnityPy") + os.makedirs(args.out_dir, exist_ok=True) + + t0 = time.time() + rows: "OrderedDict[str, dict]" = OrderedDict() + types: dict[str, str] = {} + inventory: dict[str, list] = {} + + with uc.MetaDb(args.game_path, args.region) as db: + song_ids = args.songs or db.song_ids() + for i, sid in enumerate(song_ids, 1): + if not args.quiet: + print("[%2d/%d] son%s" % (i, len(song_ids), sid), file=sys.stderr) + sheets, t = scan_song(db, sid, verbose=not args.quiet) + types.update(t) + inventory[sid] = [k.split("|", 1)[1] for k in sheets] + if not sheets: + continue + if args.per_sheet: + rows.update(sheets) + else: + merged: dict[str, tuple[int, int]] = {} + for values in sheets.values(): + for f, (g, k) in values.items(): + pg, pk = merged.get(f, (0, 0)) + merged[f] = (max(pg, g), max(pk, k)) if args.agg == "max" else (pg + g, pk + k) + rows["son%s" % sid] = merged + + # keep the game's own field order (types dict is insertion ordered) + fields = list(types.keys()) + + keys_csv = os.path.join(args.out_dir, "scan_keys.csv") + groups_csv = os.path.join(args.out_dir, "scan_groups.csv") + write_matrix(groups_csv, rows, fields, 0) + write_matrix(keys_csv, rows, fields, 1) + + stats = [] + for f in fields: + vals = [(name, v.get(f, (0, 0))[1]) for name, v in rows.items()] + used = [n for n, k in vals if k > 0] + stats.append( + { + "field": f, + "type": types[f], + "songs": len(used), + "total": sum(k for _, k in vals), + "max": max([k for _, k in vals] or [0]), + "examples": used[:6], + } + ) + stats.sort(key=lambda s: (-s["songs"], -s["total"])) + + summary = os.path.join(args.out_dir, "scan_summary.txt") + with open(summary, "w", encoding="utf-8") as fh: + fh.write("LiveTimelineWorkSheet track usage over %d rows (%s)\n" % (len(rows), args.agg)) + fh.write("%-46s %-50s %6s %9s %7s\n" % ("field", "serialized type", "songs", "total", "max")) + fh.write("-" * 125 + "\n") + for s in stats: + fh.write("%-46s %-50s %6d %9d %7d\n" % (s["field"], s["type"], s["songs"], s["total"], s["max"])) + + js = os.path.join(args.out_dir, "scan.json") + with open(js, "w", encoding="utf-8") as fh: + json.dump( + { + "rows": {n: {f: list(v) for f, v in vals.items()} for n, vals in rows.items()}, + "field_types": types, + "stats": stats, + "worksheets": inventory, + "aggregation": "per-sheet" if args.per_sheet else args.agg, + }, + fh, + indent=1, + ) + + print("scanned %d rows / %d fields in %.1fs" % (len(rows), len(fields), time.time() - t0)) + for p in (keys_csv, groups_csv, summary, js): + print(" %s" % p) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/uma_common.py b/tools/uma_common.py new file mode 100644 index 0000000..9141268 --- /dev/null +++ b/tools/uma_common.py @@ -0,0 +1,436 @@ +"""Shared library for the UmaViewer Python asset tools. + +This is a Python re-implementation of the decryption logic that lives in the +C# side of this repository: + + * ``Assets/Scripts/Config.cs`` -- the key constants + * ``Assets/Scripts/UmaAssetBundleStream.cs`` -- asset bundle XOR cipher + * ``Assets/Scripts/UmaDatabase/UmaDatabaseEntry.cs``-- per-bundle key expansion + * ``Assets/Scripts/UmaDatabase/UmaDatabaseController.cs`` -- meta DB cipher + +Nothing here writes to the game folder; every file is opened read-only. + +---------------------------------------------------------------------------- +Setup +---------------------------------------------------------------------------- + python3 -m venv ~/.venvs/umatools + ~/.venvs/umatools/bin/python -m pip install -r tools/requirements.txt + +Then run any tool with that interpreter, e.g. + + ~/.venvs/umatools/bin/python tools/dump_cutt_typetree.py 1157 + +---------------------------------------------------------------------------- +Game folder +---------------------------------------------------------------------------- +The tools look for the game's ``Persistent`` folder (the one containing +``meta`` and ``dat/``) in this order: + + 1. ``--game-path`` command line option + 2. ``$UMA_GAME_PATH`` environment variable + 3. the paths in :data:`DEFAULT_GAME_PATHS` + +---------------------------------------------------------------------------- +How the encryption works +---------------------------------------------------------------------------- +meta database + A SQLite database encrypted with SQLite3 Multiple Ciphers, cipher index 3 + (= ChaCha20 / "sqleet" scheme, non-legacy defaults). The passphrase is + ``DBKey[i] XOR DBBaseKey[i % 13]`` (32 raw bytes, see :func:`meta_key`). + We hand those raw bytes to sqlite3mc through ``PRAGMA hexkey`` which is + the exact equivalent of the C# ``sqlite3_key(db, bytes, 32)`` call. + +asset bundles + ``dat//``. The first 256 bytes are stored in + the clear; every byte at file offset ``p >= 256`` is XORed with + ``FKey[p % 88]`` where ``FKey`` is built from the 11-byte ``ABKey`` and the + per-bundle 64-bit key stored in the meta table (column ``e``): + + FKey[i * 8 + j] = ABKey[i] XOR little_endian_int64(key)[j] + + A key of 0 means the bundle is not encrypted. +""" + +from __future__ import annotations + +import os +import shutil +import struct +import sys +import tempfile +from dataclasses import dataclass +from typing import Iterable, Iterator, Optional + +# -------------------------------------------------------------------------- +# dependency check (friendly message instead of a bare ImportError) +# -------------------------------------------------------------------------- + +_MISSING: list[str] = [] + +try: + import apsw # provided by the `apsw-sqlite3mc` wheel +except ImportError: # pragma: no cover + apsw = None + _MISSING.append("apsw-sqlite3mc") + +try: + import UnityPy +except ImportError: # pragma: no cover + UnityPy = None + _MISSING.append("UnityPy") + + +def require(*modules: str) -> None: + """Abort with an actionable message if a needed dependency is absent.""" + needed = [m for m in modules if m in _MISSING] + if not needed: + return + sys.stderr.write( + "\nMissing Python package(s): %s\n\n" + "Install them into a virtualenv, e.g.\n" + " python3 -m venv ~/.venvs/umatools\n" + " ~/.venvs/umatools/bin/python -m pip install -r %s\n" + "and re-run this script with ~/.venvs/umatools/bin/python\n\n" + % (", ".join(needed), os.path.join(os.path.dirname(__file__), "requirements.txt")) + ) + raise SystemExit(2) + + +def has_sqlite3mc() -> bool: + """True when the installed apsw build actually carries the MC ciphers.""" + if apsw is None: + return False + try: + con = apsw.Connection(":memory:") + try: + list(con.execute("PRAGMA cipher")) + return True + finally: + con.close() + except Exception: + return False + + +# -------------------------------------------------------------------------- +# keys -- verbatim from Assets/Scripts/Config.cs +# -------------------------------------------------------------------------- + +DB_BASE_KEY = bytes( + [0xF1, 0x70, 0xCE, 0xA4, 0xDF, 0xCE, 0xA3, 0xE1, + 0xA5, 0xD8, 0xC7, 0x0B, 0xD1, 0x00, 0x00, 0x00] +) + +DB_KEY = bytes( + [0x6D, 0x5B, 0x65, 0x33, 0x63, 0x36, + 0x63, 0x25, 0x54, 0x71, 0x2D, 0x73, + 0x50, 0x53, 0x63, 0x38, 0x6D, 0x34, + 0x37, 0x7B, 0x35, 0x63, 0x70, 0x23, + 0x37, 0x34, 0x53, 0x29, 0x73, 0x43, + 0x36, 0x33] +) + +GLOBAL_DB_KEY = bytes( + [0x56, 0x63, 0x6B, 0x63, 0x42, 0x72, 0x37, 0x76, 0x65, 0x70, 0x41, 0x62] +) + +AB_KEY = bytes([0x53, 0x2B, 0x46, 0x31, 0xE4, 0xA7, 0xB9, 0x47, 0x3E, 0x7C, 0xFB]) + +HEADER_SIZE = 256 +FKEY_LEN = len(AB_KEY) * 8 # 88 + +#: sqlite3mc cipher index used by UmaDatabaseController.ReadMetaFromEncryptedDb +META_CIPHER = "chacha20" # == sqlite3mc_config(db, "cipher", 3) + +DEFAULT_GAME_PATHS = ( + "/mnt/d/Umamusume/umamusume_Data/Persistent", + os.path.expandvars(r"%USERPROFILE%\AppData\LocalLow\Cygames\umamusume"), +) + + +def meta_key(region: str = "jp") -> bytes: + """Final meta-DB passphrase: ``DBKey[i] XOR DBBaseKey[i % 13]``. + + Mirrors ``UmaDatabaseController.GenFinalKey``. + """ + key = GLOBAL_DB_KEY if region.lower() in ("global", "en") else DB_KEY + return bytes(key[i] ^ DB_BASE_KEY[i % 13] for i in range(len(key))) + + +# -------------------------------------------------------------------------- +# asset bundle cipher +# -------------------------------------------------------------------------- + +def build_fkey(key: int) -> bytes: + """Expand a per-bundle 64-bit key into the 88-byte XOR keystream. + + Mirrors ``UmaDatabaseEntry.FKey``. + """ + key_bytes = struct.pack(" bytes: + """Decrypt a whole asset bundle image held in memory.""" + if not key or len(data) <= HEADER_SIZE: + return bytes(data) + fk = build_fkey(key) + body = data[HEADER_SIZE:] + n = len(body) + # byte at file offset p uses fk[p % 88]; body[0] is at offset HEADER_SIZE + off = HEADER_SIZE % FKEY_LEN + rotated = fk[off:] + fk[:off] + stream = (rotated * (n // FKEY_LEN + 1))[:n] + plain = (int.from_bytes(body, "big") ^ int.from_bytes(stream, "big")).to_bytes(n, "big") + return bytes(data[:HEADER_SIZE]) + plain + + +def decrypt_bundle_file(path: str, key: int) -> bytes: + with open(path, "rb") as fh: + return decrypt_bundle_bytes(fh.read(), key) + + +# -------------------------------------------------------------------------- +# meta database +# -------------------------------------------------------------------------- + +@dataclass +class MetaEntry: + """One row of the meta ``a`` table (see UmaDatabaseEntry.cs).""" + + type: str # column m, e.g. "_3d_cutt" + name: str # column n, the asset path, e.g. cutt/cutt_son1157/type_01 + url: str # column h, the hashed filename under dat/ + checksum: str # column c + prerequisites: str # column d, ';' separated dependency asset paths + key: int # column e, per-bundle cipher key (0 = plaintext bundle) + + @property + def is_encrypted(self) -> bool: + return self.key != 0 + + def path(self, game_path: str) -> str: + return os.path.join(game_path, "dat", self.url[:2], self.url) + + def deps(self) -> list[str]: + return [d for d in (self.prerequisites or "").split(";") if d] + + +def find_game_path(explicit: Optional[str] = None) -> str: + candidates = [] + if explicit: + candidates.append(explicit) + env = os.environ.get("UMA_GAME_PATH") + if env: + candidates.append(env) + candidates.extend(DEFAULT_GAME_PATHS) + for c in candidates: + if c and os.path.isfile(os.path.join(c, "meta")): + return c + raise SystemExit( + "Could not locate the game's Persistent folder (needs a 'meta' file).\n" + "Tried: %s\nPass --game-path or set $UMA_GAME_PATH." % ", ".join(filter(None, candidates)) + ) + + +class MetaDb: + """Read-only access to the encrypted ``meta`` SQLite database. + + The game keeps ``meta`` open, so we work on a temporary copy by default. + """ + + def __init__(self, game_path: Optional[str] = None, region: str = "jp", copy: bool = True, + readonly: bool = True): + require("apsw-sqlite3mc") + self.game_path = find_game_path(game_path) + self.meta_path = os.path.join(self.game_path, "meta") + self._tmp = None + src = self.meta_path + if copy: + fd, tmp = tempfile.mkstemp(prefix="uma_meta_", suffix=".db") + os.close(fd) + shutil.copyfile(self.meta_path, tmp) + self._tmp = tmp + src = tmp + elif not readonly: + raise ValueError("refusing to open the game's meta file writable; use copy=True") + flags = apsw.SQLITE_OPEN_READONLY if readonly else ( + apsw.SQLITE_OPEN_READWRITE | apsw.SQLITE_OPEN_CREATE + ) + self.con = apsw.Connection(src, flags=flags) + key = meta_key(region) + try: + list(self.con.execute("PRAGMA cipher='%s'" % META_CIPHER)) + list(self.con.execute("PRAGMA hexkey='%s'" % key.hex())) + list(self.con.execute("SELECT name FROM sqlite_master LIMIT 1")) + except Exception as exc: + self.close() + raise SystemExit( + "Failed to decrypt the meta database (%s).\n" + "Either the keys in Assets/Scripts/Config.cs are out of date, or the\n" + "installed apsw build has no SQLite3MultipleCiphers support " + "(need the 'apsw-sqlite3mc' wheel, not plain 'apsw')." % exc + ) + + # -- lifecycle --------------------------------------------------------- + def close(self) -> None: + try: + self.con.close() + except Exception: + pass + if self._tmp and os.path.exists(self._tmp): + try: + os.remove(self._tmp) + except OSError: + pass + self._tmp = None + + def __enter__(self) -> "MetaDb": + return self + + def __exit__(self, *exc) -> None: + self.close() + + # -- queries ----------------------------------------------------------- + _COLS = "m,n,h,c,d,e" + + def _rows(self, where: str, args: Iterable = ()) -> Iterator[MetaEntry]: + sql = "SELECT %s FROM a %s" % (self._COLS, where) + for m, n, h, c, d, e in self.con.execute(sql, tuple(args)): + yield MetaEntry(m or "", n or "", h or "", c or "", d or "", e or 0) + + def get(self, name: str) -> Optional[MetaEntry]: + for e in self._rows("WHERE n = ?", (name,)): + return e + return None + + def like(self, pattern: str) -> list[MetaEntry]: + return list(self._rows("WHERE n LIKE ? ORDER BY n", (pattern,))) + + def all(self) -> list[MetaEntry]: + return list(self._rows("ORDER BY n")) + + def song_ids(self) -> list[str]: + """All live song ids that have a ``cutt/cutt_son/`` bundle group.""" + import re + + ids = set() + for (n,) in self.con.execute("SELECT n FROM a WHERE n LIKE 'cutt/cutt_son%'"): + m = re.match(r"cutt/cutt_son(\d+)/", n or "") + if m: + ids.add(m.group(1)) + return sorted(ids, key=int) + + # -- bundle loading ---------------------------------------------------- + def bundle_bytes(self, name_or_entry) -> bytes: + entry = name_or_entry if isinstance(name_or_entry, MetaEntry) else self.get(name_or_entry) + if entry is None: + raise KeyError("no meta entry named %r" % name_or_entry) + p = entry.path(self.game_path) + if not os.path.isfile(p): + raise FileNotFoundError( + "%s -> %s (not downloaded; open it once in the game or in UmaViewer)" % (entry.name, p) + ) + return decrypt_bundle_file(p, entry.key) + + def load(self, name_or_entry, with_deps: bool = False): + """Decrypt and open a bundle with UnityPy; returns an Environment.""" + require("UnityPy") + import io + + entry = name_or_entry if isinstance(name_or_entry, MetaEntry) else self.get(name_or_entry) + if entry is None: + raise KeyError("no meta entry named %r" % name_or_entry) + env = UnityPy.Environment() + if with_deps: + for dep in entry.deps(): + try: + env.load_file(io.BytesIO(self.bundle_bytes(dep)), name=dep, is_dependency=True) + except Exception as exc: + sys.stderr.write(" warning: dependency %s not loaded (%s)\n" % (dep, exc)) + env.load_file(io.BytesIO(self.bundle_bytes(entry)), name=entry.name) + return env + + +# -------------------------------------------------------------------------- +# helpers shared by the cutt tools +# -------------------------------------------------------------------------- + +#: A MonoBehaviour is a LiveTimelineWorkSheet when its TypeTree has this field. +WORKSHEET_MARKER = "cameraPosKeys" + + +def is_worksheet(obj) -> bool: + node = getattr(getattr(obj, "serialized_type", None), "node", None) + if node is None: + return False + return any(ch.m_Name == WORKSHEET_MARKER for ch in node.m_Children) + + +def worksheets(env): + """Yield ``(name, obj, typetree_root_node)`` for each WorkSheet in *env*.""" + for obj in env.objects: + if obj.type.name != "MonoBehaviour" or not is_worksheet(obj): + continue + node = obj.serialized_type.node + try: + name = obj.read_typetree().get("m_Name") or "?" + except Exception: + name = "?" + yield name, obj, node + + +def count_keyframes(value) -> int: + """Total number of timeline keyframes reachable from *value*. + + Every timeline key list is serialized as ``{_attribute, _playMode, + thisList}``; we sum the ``thisList`` lengths and never descend into the + keyframes themselves. + """ + if isinstance(value, dict): + if "thisList" in value: + tl = value["thisList"] + return len(tl) if isinstance(tl, list) else 0 + return sum(count_keyframes(v) for v in value.values() if isinstance(v, (dict, list))) + if isinstance(value, list): + return sum(count_keyframes(v) for v in value if isinstance(v, (dict, list))) + return 0 + + +def count_groups(value) -> int: + """Size of the top-level container of a track field. + + * key list (``thisList``) -> number of keyframes + * list of named track groups -> number of groups + * struct of several key lists -> number of sub key lists + """ + if isinstance(value, list): + return len(value) + if isinstance(value, dict): + if "thisList" in value: + tl = value["thisList"] + return len(tl) if isinstance(tl, list) else 0 + return sum(1 for v in value.values() if isinstance(v, (dict, list))) + return 0 + + +def group_names(value) -> list[str]: + """Names of the track groups inside a list-style field, if any.""" + if isinstance(value, list): + return [v.get("name", "") for v in value if isinstance(v, dict) and "name" in v] + return [] + + +def add_common_args(parser) -> None: + parser.add_argument( + "--game-path", + default=None, + help="game Persistent folder (default: $UMA_GAME_PATH or %s)" % DEFAULT_GAME_PATHS[0], + ) + parser.add_argument( + "--region", default="jp", choices=("jp", "global"), help="which DB key to use (default: jp)" + )