From 714ad8bff2afb9e2e5825d9c6d2793dbac8e8fed Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 30 Jul 2026 06:16:53 -0500 Subject: [PATCH] feat: add Unity asset export via optional agentdecompile[unity] extra New `unity-export` CLI subcommand (agentdecompile-recover unity-export --install-root --out-dir ) and src/agentdecompile_recovery/ unity_assets.py, adapted from a bounded-memory UnityPy-based export script proven out on a large real-world title where AssetRipper's whole-graph loading OOM'd (heavy Mesh/BlendShapeData). Exports textures, sprites, audio clips, text assets, and fonts one object at a time (freeing each immediately), so peak memory stays roughly constant regardless of asset file size on disk -- unlike full project-reconstruction tools that load the entire asset graph before writing anything out. Deliberately does not attempt meshes/materials/ scene-prefab structure; that needs the fuller project-graph tooling this module is not trying to replace. UnityPy is optional (`pip install agentdecompile[unity]`, or via the `[all]` extra) and imported lazily -- importing unity_assets never requires it to be installed; a clear ImportError names the extra to install if the CLI command is invoked without it. Also fixed, discovered while running the test suite on this branch: - tests/test_rewrite_queue.py's two multiprocessing concurrency tests failed under Python 3.14 even on Linux (PicklingError on a locally- nested function target) -- Python 3.14 changed the default multiprocessing start method away from fork on Linux too (previously only macOS defaulted away from fork). Forced multiprocessing.get_context ("fork") explicitly, same fix already shipped for the macOS-only case. - source_parity_synthesize.py: removed an unused `.state.now` import (ruff F401), pre-existing and unrelated, blocking the lint CI step. 610 unit tests pass; ruff clean. --- pyproject.toml | 5 +- src/agentdecompile_recovery/cli.py | 21 ++ .../source_parity_synthesize.py | 1 - src/agentdecompile_recovery/unity_assets.py | 232 ++++++++++++++++++ tests/test_rewrite_queue.py | 31 ++- tests/test_unity_assets.py | 100 ++++++++ 6 files changed, 380 insertions(+), 10 deletions(-) create mode 100644 src/agentdecompile_recovery/unity_assets.py create mode 100644 tests/test_unity_assets.py diff --git a/pyproject.toml b/pyproject.toml index 3d54059c..60990a2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,6 +86,9 @@ agentdecompile-proxy = "agentdecompile_cli.server:proxy_main" semantic = [ "chromadb", ] +unity = [ + "UnityPy>=1.25.2", +] test = [ "pytest>=9.0.3", "pytest-asyncio>=1.3.0", @@ -98,7 +101,7 @@ test = [ # PyPI package, and cannot be expressed as a pip dependency. Install it # separately with `dotnet tool install -g ilspycmd`; see README.md. all = [ - "agentdecompile[semantic]", + "agentdecompile[semantic,unity]", ] [build-system] diff --git a/src/agentdecompile_recovery/cli.py b/src/agentdecompile_recovery/cli.py index 1ea3096d..c8debd6a 100644 --- a/src/agentdecompile_recovery/cli.py +++ b/src/agentdecompile_recovery/cli.py @@ -329,6 +329,13 @@ def build_parser() -> argparse.ArgumentParser: ) cleanup_package.add_argument("--package", type=Path, required=True, help="Recovered-source package directory.") cleanup_package.add_argument("--out-dir", type=Path, required=True, help="Output directory for the cleaned package.") + + unity_export = sub.add_parser( + "unity-export", + help="Export textures/sprites/audio/text/fonts from a Unity game install (requires agentdecompile[unity]).", + ) + unity_export.add_argument("--install-root", type=Path, required=True, help="Unity game install directory (containing a *_Data folder).") + unity_export.add_argument("--out-dir", type=Path, required=True, help="Output directory for exported assets.") return parser @@ -816,6 +823,18 @@ def run_source_cleanup_package(args: argparse.Namespace) -> int: return 0 +def run_unity_export(args: argparse.Namespace) -> int: + from .unity_assets import export_primary_content + + try: + receipt = export_primary_content(install_root=args.install_root, output_dir=args.out_dir) + except ImportError as exc: + print(json.dumps({"schema": "agentdecompile.unity-primary-content-export.v1", "status": "error", "reason": str(exc)}, indent=2, sort_keys=True)) + return 1 + print(json.dumps(receipt, indent=2, sort_keys=True)) + return 0 if receipt.get("status") == "complete" else 1 + + def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv if argv is not None else sys.argv[1:]) @@ -851,6 +870,8 @@ def main(argv: list[str] | None = None) -> int: return run_source_cleanup(args) if args.command == "source-cleanup-package": return run_source_cleanup_package(args) + if args.command == "unity-export": + return run_unity_export(args) parser.print_help() return 2 diff --git a/src/agentdecompile_recovery/source_parity_synthesize.py b/src/agentdecompile_recovery/source_parity_synthesize.py index 482ce25e..aa782977 100755 --- a/src/agentdecompile_recovery/source_parity_synthesize.py +++ b/src/agentdecompile_recovery/source_parity_synthesize.py @@ -24,7 +24,6 @@ from typing import Any, Iterable from .package_verify import build_shim, compile_with_msvc -from .state import now ROOT = Path.cwd() DEFAULT_VC_ROOT: Path | None = None diff --git a/src/agentdecompile_recovery/unity_assets.py b/src/agentdecompile_recovery/unity_assets.py new file mode 100644 index 00000000..78ddd887 --- /dev/null +++ b/src/agentdecompile_recovery/unity_assets.py @@ -0,0 +1,232 @@ +"""Unity asset extraction via UnityPy (optional `agentdecompile[unity]` extra). + +Unity games ship two separable recovery surfaces: managed C# assemblies +(decompile with ILSpy/`ilspycmd` -- see README's "Optional: .NET/IL +decompilation support" section) and serialized asset data (textures, audio, +text, fonts) inside the built player's `*.assets`/`*.resS` files. This module +handles the second surface. + +Tools like AssetRipper reconstruct a full re-openable Unity project (scenes, +prefabs, meshes, MonoBehaviour data) but load the entire asset graph into +memory before writing anything out, which can exceed available RAM on large +titles with heavy mesh/blend-shape data. This module trades full project +reconstruction for a bounded-memory alternative: it processes one top-level +data file at a time and reads/exports one object at a time, freeing each +before moving to the next, so peak memory stays roughly constant regardless +of how large the asset file is on disk. Mesh/BlendShapeData export is +intentionally out of scope here for the same reason -- that is exactly the +class of data that makes whole-graph tools OOM, and reconstructing it +correctly (skinned mesh renderers, bone weights, blend shape frames) needs +the fuller project-graph tooling this module deliberately does not attempt. + +`UnityPy` is an optional dependency (`agentdecompile[unity]`) and is imported +lazily inside the functions that need it, so importing this module never +requires it to be installed. +""" + +from __future__ import annotations + +import gc +from pathlib import Path +from typing import Any + +EXPORTABLE_TYPES = {"Texture2D", "Sprite", "AudioClip", "TextAsset", "Font"} + +# Companion/metadata files alongside the top-level serialized data files. +# UnityPy follows references to *.resS/*.resource automatically when loading +# the owning data file, so these are never scanned directly. +_SKIP_SUFFIXES = (".resS", ".resource", ".json", ".config", ".info") +_SKIP_NAMES = {"ScriptingAssemblies.json", "RuntimeInitializeOnLoads.json"} + + +def _require_unitypy() -> Any: + try: + import UnityPy + except ImportError as exc: + raise ImportError( + "UnityPy is required for Unity asset export. Install it with " + "`pip install agentdecompile[unity]` (or `pipx install agentdecompile[unity]`)." + ) from exc + return UnityPy + + +def sanitize_asset_name(name: str) -> str: + """Replace filesystem-unsafe characters; never return an empty string.""" + + bad = '\\/:*?"<>|' + out = "".join("_" if ch in bad else ch for ch in name) + return out.strip() or "unnamed" + + +def discover_data_files(data_dir: Path) -> list[Path]: + """List top-level serialized Unity data files under a `*_Data` directory.""" + + out = [] + for path in sorted(data_dir.iterdir()): + if not path.is_file(): + continue + if path.name in _SKIP_NAMES: + continue + if path.suffix in _SKIP_SUFFIXES: + continue + out.append(path) + return out + + +def find_unity_data_dir(install_root: Path) -> Path | None: + """Return the first `*_Data` directory directly under a Unity install root.""" + + candidates = [p for p in install_root.iterdir() if p.is_dir() and p.name.endswith("_Data")] + return candidates[0] if candidates else None + + +def _export_texture2d(data: Any, dest_dir: Path, path_id: int) -> None: + name = sanitize_asset_name(data.m_Name) if data.m_Name else f"texture_{path_id}" + dest = dest_dir / "Texture2D" / f"{name}.png" + dest.parent.mkdir(parents=True, exist_ok=True) + if dest.exists(): + return + image = data.image + if image is None: + return + image.save(dest) + + +def _export_sprite(data: Any, dest_dir: Path, path_id: int) -> None: + name = sanitize_asset_name(data.m_Name) if data.m_Name else f"sprite_{path_id}" + dest = dest_dir / "Sprite" / f"{name}.png" + dest.parent.mkdir(parents=True, exist_ok=True) + if dest.exists(): + return + image = data.image + if image is None: + return + image.save(dest) + + +def _export_audioclip(data: Any, dest_dir: Path, path_id: int) -> None: + base = sanitize_asset_name(data.m_Name) if data.m_Name else f"audio_{path_id}" + out_dir = dest_dir / "AudioClip" + out_dir.mkdir(parents=True, exist_ok=True) + samples = data.samples + if not samples: + return + if len(samples) == 1: + (out_dir / f"{base}.wav").write_bytes(next(iter(samples.values()))) + return + for sub_name, clip_bytes in samples.items(): + (out_dir / f"{sanitize_asset_name(sub_name)}.wav").write_bytes(clip_bytes) + + +def _export_textasset(data: Any, dest_dir: Path, path_id: int) -> None: + name = sanitize_asset_name(data.m_Name) if data.m_Name else f"text_{path_id}" + out_dir = dest_dir / "TextAsset" + out_dir.mkdir(parents=True, exist_ok=True) + script = data.m_Script + payload = script.encode("utf-8", "surrogateescape") if isinstance(script, str) else bytes(script) + (out_dir / f"{name}.txt").write_bytes(payload) + + +def _export_font(data: Any, dest_dir: Path, path_id: int) -> None: + name = sanitize_asset_name(data.m_Name) if data.m_Name else f"font_{path_id}" + out_dir = dest_dir / "Font" + out_dir.mkdir(parents=True, exist_ok=True) + font_data = getattr(data, "m_FontData", None) + if not font_data: + return + (out_dir / f"{name}.ttf").write_bytes(bytes(font_data)) + + +_EXPORTERS = { + "Texture2D": _export_texture2d, + "Sprite": _export_sprite, + "AudioClip": _export_audioclip, + "TextAsset": _export_textasset, + "Font": _export_font, +} + + +def export_data_file(source: Path, dest_dir: Path) -> tuple[int, int, list[str]]: + """Export supported asset types from one Unity data file. + + Returns ``(exported_count, failed_count, failure_messages)``. Reads and + frees one object at a time so peak memory does not scale with how many + objects the file contains. + """ + + unitypy = _require_unitypy() + exported = 0 + failed = 0 + failures: list[str] = [] + env = unitypy.load(str(source)) + for obj in env.objects: + type_name = obj.type.name + if type_name not in EXPORTABLE_TYPES: + continue + try: + data = obj.read() + _EXPORTERS[type_name](data, dest_dir, obj.path_id) + exported += 1 + except Exception as exc: # noqa: BLE001 - keep scanning past one bad object + failed += 1 + failures.append(f"{type_name} path_id={obj.path_id}: {exc}") + finally: + data = None + gc.collect() + env = None + gc.collect() + return exported, failed, failures + + +def export_primary_content(install_root: Path, output_dir: Path) -> dict[str, Any]: + """Export textures, sprites, audio, text assets, and fonts from a Unity + install directory into ``output_dir``, one type-named subfolder each. + + Does not export meshes, materials, or scene/prefab structure -- use a + full project-reconstruction tool (e.g. AssetRipper) when that's needed + and the asset graph fits in available memory. + """ + + install_root = install_root.resolve() + output_dir = output_dir.resolve() + output_dir.mkdir(parents=True, exist_ok=True) + + data_dir = find_unity_data_dir(install_root) + if data_dir is None: + return { + "schema": "agentdecompile.unity-primary-content-export.v1", + "status": "error", + "reason": "no *_Data directory found under install root", + "installRoot": str(install_root), + } + + files = discover_data_files(data_dir) + per_file: list[dict[str, Any]] = [] + total_exported = 0 + total_failed = 0 + for source in files: + exported, failed, failures = export_data_file(source, output_dir) + total_exported += exported + total_failed += failed + per_file.append( + { + "file": source.name, + "sizeBytes": source.stat().st_size, + "exported": exported, + "failed": failed, + "failures": failures[:20], + } + ) + + return { + "schema": "agentdecompile.unity-primary-content-export.v1", + "status": "complete", + "installRoot": str(install_root), + "dataDir": str(data_dir), + "outputDir": str(output_dir), + "filesScanned": len(files), + "totalExported": total_exported, + "totalFailed": total_failed, + "perFile": per_file, + "claimBoundary": "textures/sprites/audio/text/fonts only -- no meshes, materials, or scene/prefab structure", + } diff --git a/tests/test_rewrite_queue.py b/tests/test_rewrite_queue.py index bc7d530d..8aa91dce 100644 --- a/tests/test_rewrite_queue.py +++ b/tests/test_rewrite_queue.py @@ -23,6 +23,26 @@ def _claim_worker(work_dir: str, request_id: str, claimant: str, result_path: st Path(result_path).write_text("1" if ok else "0", encoding="utf-8") +def _write_request_worker(work_dir: str, name: str) -> None: + rewrite_queue.write_rewrite_request( + Path(work_dir), function_name=name, entry="0x1", candidate_source=f"src-{name}", mismatch_class=None, mismatch_histogram=None + ) + + +# macOS has always defaulted multiprocessing to the "spawn" start method +# (vs "fork" on Linux); Python 3.14 additionally changed the default away +# from fork on Linux too (deprecated as unsafe in multi-threaded processes -- +# see PEP 734 / the multiprocessing docs). Spawn re-imports the target +# function in a fresh interpreter rather than fork()ing the already-loaded +# parent, which fails for a locally-nested function with a PicklingError +# ("Can't pickle local object"), and can also fail to re-import the test +# module at all depending on how the fresh interpreter's sys.path is set up. +# Force fork explicitly -- available on both Linux and macOS (the only two +# CI platforms) -- since these tests only need process-level isolation, not +# spawn's clean-slate import behavior. +_FORK_CONTEXT = multiprocessing.get_context("fork") + + def test_write_rewrite_request_creates_pending_entry(tmp_path: Path) -> None: request_id = rewrite_queue.write_rewrite_request( tmp_path, @@ -241,8 +261,8 @@ def test_concurrent_claims_from_separate_processes_only_one_wins(tmp_path: Path) ) result_a = tmp_path / "result_a.txt" result_b = tmp_path / "result_b.txt" - proc_a = multiprocessing.Process(target=_claim_worker, args=(str(tmp_path), request_id, "proc-a", str(result_a))) - proc_b = multiprocessing.Process(target=_claim_worker, args=(str(tmp_path), request_id, "proc-b", str(result_b))) + proc_a = _FORK_CONTEXT.Process(target=_claim_worker, args=(str(tmp_path), request_id, "proc-a", str(result_a))) + proc_b = _FORK_CONTEXT.Process(target=_claim_worker, args=(str(tmp_path), request_id, "proc-b", str(result_b))) proc_a.start() proc_b.start() proc_a.join(timeout=10) @@ -264,13 +284,8 @@ def test_write_rewrite_request_survives_concurrent_writes_to_different_entries(t not lose each other's entries (the lock serializes the whole file, not just same-entry races).""" - def _writer(work_dir: str, name: str) -> None: - rewrite_queue.write_rewrite_request( - Path(work_dir), function_name=name, entry="0x1", candidate_source=f"src-{name}", mismatch_class=None, mismatch_histogram=None - ) - procs = [ - multiprocessing.Process(target=_writer, args=(str(tmp_path), f"sub_{i}")) + _FORK_CONTEXT.Process(target=_write_request_worker, args=(str(tmp_path), f"sub_{i}")) for i in range(6) ] for proc in procs: diff --git a/tests/test_unity_assets.py b/tests/test_unity_assets.py new file mode 100644 index 00000000..01f6eb2d --- /dev/null +++ b/tests/test_unity_assets.py @@ -0,0 +1,100 @@ +"""Unit tests for the optional Unity asset-export module (agentdecompile[unity]). + +UnityPy itself is not exercised here (no real Unity asset fixtures are +available in this repo, and UnityPy is an optional dependency) -- these tests +cover the pure filesystem/naming logic that doesn't need it, plus the +lazy-import error path. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from agentdecompile_recovery import unity_assets + +pytestmark = pytest.mark.unit + + +def test_sanitize_asset_name_replaces_unsafe_characters() -> None: + assert unity_assets.sanitize_asset_name('a/b\\c:d*e?f"gi|j') == "a_b_c_d_e_f_g_h_i_j" + + +def test_sanitize_asset_name_strips_whitespace() -> None: + assert unity_assets.sanitize_asset_name(" spaced ") == "spaced" + + +def test_sanitize_asset_name_never_returns_empty() -> None: + assert unity_assets.sanitize_asset_name("") == "unnamed" + assert unity_assets.sanitize_asset_name(" ") == "unnamed" + + +def test_discover_data_files_skips_companion_and_metadata_files(tmp_path: Path) -> None: + data_dir = tmp_path / "Game_Data" + data_dir.mkdir() + (data_dir / "resources.assets").write_bytes(b"x") + (data_dir / "resources.assets.resS").write_bytes(b"x") + (data_dir / "resources.resource").write_bytes(b"x") + (data_dir / "ScriptingAssemblies.json").write_bytes(b"x") + (data_dir / "RuntimeInitializeOnLoads.json").write_bytes(b"x") + (data_dir / "boot.config").write_bytes(b"x") + (data_dir / "app.info").write_bytes(b"x") + (data_dir / "level0").write_bytes(b"x") + (data_dir / "sub").mkdir() + + files = unity_assets.discover_data_files(data_dir) + + assert [f.name for f in files] == ["level0", "resources.assets"] + + +def test_discover_data_files_empty_dir_returns_empty_list(tmp_path: Path) -> None: + data_dir = tmp_path / "Empty_Data" + data_dir.mkdir() + assert unity_assets.discover_data_files(data_dir) == [] + + +def test_find_unity_data_dir_locates_the_data_folder(tmp_path: Path) -> None: + install_root = tmp_path / "MyGame" + install_root.mkdir() + (install_root / "MyGame_Data").mkdir() + (install_root / "MyGame.exe").write_bytes(b"x") + + found = unity_assets.find_unity_data_dir(install_root) + + assert found == install_root / "MyGame_Data" + + +def test_find_unity_data_dir_returns_none_when_absent(tmp_path: Path) -> None: + install_root = tmp_path / "NotUnity" + install_root.mkdir() + (install_root / "readme.txt").write_bytes(b"x") + + assert unity_assets.find_unity_data_dir(install_root) is None + + +def test_export_primary_content_reports_error_when_no_data_dir(tmp_path: Path) -> None: + install_root = tmp_path / "NotUnity" + install_root.mkdir() + output_dir = tmp_path / "out" + + receipt = unity_assets.export_primary_content(install_root, output_dir) + + assert receipt["status"] == "error" + assert "no *_Data directory" in receipt["reason"] + + +def test_require_unitypy_raises_actionable_error_when_missing(monkeypatch: pytest.MonkeyPatch) -> None: + import builtins + + real_import = builtins.__import__ + + def _fake_import(name: str, *args: object, **kwargs: object) -> object: + if name == "UnityPy": + raise ImportError("No module named 'UnityPy'") + return real_import(name, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(builtins, "__import__", _fake_import) + + with pytest.raises(ImportError, match=r"agentdecompile\[unity\]"): + unity_assets._require_unitypy()