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 <noreply@anthropic.com>
This commit is contained in:
ritenseki
@@ -0,0 +1,4 @@
|
||||
# generated output of the asset tools (meta_plain.db, scan_*.csv, ...)
|
||||
out/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
@@ -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())
|
||||
@@ -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<id>/``:
|
||||
|
||||
cutt_son<id> prefab bundle (no worksheets)
|
||||
data LiveTimelineData (no worksheets)
|
||||
son<id>_camera WorkSheet "son<id>_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())
|
||||
@@ -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())
|
||||
@@ -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
|
||||
@@ -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<id>/`` 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<id>_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())
|
||||
@@ -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/<first 2 chars of hash>/<hash>``. 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("<q", key) # signed, little endian
|
||||
out = bytearray(FKEY_LEN)
|
||||
for i, b in enumerate(AB_KEY):
|
||||
base = i << 3
|
||||
for j in range(8):
|
||||
out[base + j] = b ^ key_bytes[j]
|
||||
return bytes(out)
|
||||
|
||||
|
||||
def decrypt_bundle_bytes(data: bytes, key: int) -> 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<id>/`` 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)"
|
||||
)
|
||||
Reference in New Issue
Block a user