ChildHood/tools/asset-catalog/run_all.py

"""UE5 资产探索编排器。

在单次 UE5 编辑器会话内顺序执行所有分析步骤。

步骤:
  1. discover  — 资产发现
  2. mesh      — 模型几何提取
  3. material  — 材质参数提取
  4. blueprint — 蓝图结构提取

通过 AC_STEPS 环境变量控制执行哪些步骤（逗号分隔）。
默认全部执行。

环境变量（由 run_all.bat 设置）:
  AC_SCAN_ROOT   — 扫描路径
  AC_OUTPUT_DIR  — 输出目录
  AC_NO_QUIT     — 是否不退出
  AC_STEPS       — 步骤列表
  AC_PROJECT_ROOT — 项目根目录（用于 sys.path）
"""

import json
import os
import sys
import time
from pathlib import Path

# ---- sys.path 设置 ----
_SCRIPT_DIR = Path(__file__).resolve().parent
if str(_SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(_SCRIPT_DIR))

# 项目根目录
_project_root = os.environ.get("AC_PROJECT_ROOT", "")
if _project_root:
    PROJECT_ROOT = Path(_project_root)
else:
    PROJECT_ROOT = _SCRIPT_DIR.parent.parent

# ---- Unreal import ----
try:
    import unreal
    HAS_UNREAL = True
except ImportError:
    HAS_UNREAL = False
    print("[AssetCatalog] 错误: unreal 模块不可用，必须在 UE5 环境中运行")
    sys.exit(1)

from _common import (
    get_scan_root, get_output_dir, get_steps, should_no_quit,
    write_json, wait_for_editor, quit_editor, iso_now,
)

TAG = "[AssetCatalog]"


def run():
    """主流程：顺序执行所有启用的步骤。"""
    print(f"{TAG} {'=' * 50}")
    print(f"{TAG}   UE5 资产探索工具")
    print(f"{TAG} {'=' * 50}")

    scan_root = get_scan_root()
    output_dir = get_output_dir()
    steps = get_steps()

    print(f"{TAG} 扫描路径: {scan_root}")
    print(f"{TAG} 输出目录: {output_dir}")
    print(f"{TAG} 执行步骤: {', '.join(steps)}")
    print(f"{TAG}")

    # 确保输出目录存在
    output_dir.mkdir(parents=True, exist_ok=True)

    # 等待编辑器就绪
    if not wait_for_editor():
        print(f"{TAG} 错误: 编辑器初始化超时")
        return

    log = {
        "started_at": iso_now(),
        "scan_root": scan_root,
        "output_dir": str(output_dir),
        "steps": [],
    }

    index_data = None

    # ---- Step 0: Discover ----
    if "discover" in steps:
        print(f"\n{TAG} {'-' * 40}")
        print(f"{TAG} Step 0: Discover")
        print(f"{TAG} {'-' * 40}")

        t0 = time.time()
        from discover_assets import run as run_discover
        result = run_discover()
        elapsed = time.time() - t0

        result["elapsed_sec"] = round(elapsed, 1)
        log["steps"].append({"name": "discover", **result})

        if result.get("status") == "ok":
            # load index.json for subsequent steps
            try:
                index_data = json.loads(
                    (output_dir / "index.json").read_text(encoding="utf-8")
                )
            except Exception as e:
                print(f"{TAG} WARN: load index.json failed: {e}")
        else:
            print(f"{TAG} Discover failed, skipping remaining steps")
            log["completed_at"] = iso_now()
            log["status"] = "discover_failed"
            _save_log(log, output_dir)
            _maybe_quit()
            return

    # ---- Step 1: Meshes ----
    if "mesh" in steps:
        print(f"\n{TAG} {'-' * 40}")
        print(f"{TAG} Step 1: Mesh Analysis")
        print(f"{TAG} {'-' * 40}")

        t0 = time.time()
        try:
            from analyze_meshes import run as run_mesh
            result = run_mesh(index_data)
        except Exception as e:
            print(f"{TAG} ERROR in mesh analysis: {e}")
            import traceback
            traceback.print_exc()
            result = {"status": "error", "reason": str(e)}
        elapsed = time.time() - t0

        result["elapsed_sec"] = round(elapsed, 1)
        log["steps"].append({"name": "mesh", **result})

    # ---- Step 2: Materials ----
    if "material" in steps:
        print(f"\n{TAG} {'-' * 40}")
        print(f"{TAG} Step 2: Material Analysis")
        print(f"{TAG} {'-' * 40}")

        t0 = time.time()
        try:
            from analyze_materials import run as run_material
            result = run_material(index_data)
        except Exception as e:
            print(f"{TAG} ERROR in material analysis: {e}")
            import traceback
            traceback.print_exc()
            result = {"status": "error", "reason": str(e)}
        elapsed = time.time() - t0

        result["elapsed_sec"] = round(elapsed, 1)
        log["steps"].append({"name": "material", **result})

    # ---- Step 3: Blueprints ----
    if "blueprint" in steps:
        print(f"\n{TAG} {'-' * 40}")
        print(f"{TAG} Step 3: Blueprint Analysis")
        print(f"{TAG} {'-' * 40}")

        t0 = time.time()
        try:
            from analyze_blueprints import run as run_blueprint
            result = run_blueprint(index_data)
        except Exception as e:
            print(f"{TAG} ERROR in blueprint analysis: {e}")
            import traceback
            traceback.print_exc()
            result = {"status": "error", "reason": str(e)}
        elapsed = time.time() - t0

        result["elapsed_sec"] = round(elapsed, 1)
        log["steps"].append({"name": "blueprint", **result})

    # ---- 汇总 ----
    log["completed_at"] = iso_now()
    log["status"] = "ok"
    _save_log(log, output_dir)

    print(f"\n{TAG} {'=' * 50}")
    print(f"{TAG}   全部完成")
    print(f"{TAG} {'=' * 50}")
    for step in log["steps"]:
        print(f"{TAG}   {step['name']}: {step.get('status', '?')} "
              f"({step.get('analyzed', step.get('total', '?'))} 成功, "
              f"{step.get('elapsed_sec', '?')}s)")

    _maybe_quit()


def _save_log(log: dict, output_dir: Path):
    """保存运行日志。"""
    log_path = output_dir / "catalog_log.json"
    write_json(log, log_path)
    print(f"{TAG} 日志已保存: {log_path}")


def _maybe_quit():
    """根据 AC_NO_QUIT 决定是否退出编辑器。"""
    if should_no_quit():
        print(f"{TAG} AC_NO_QUIT=1，不退出编辑器")
    else:
        print(f"{TAG} 退出编辑器...")
        quit_editor()


# ---- 入口 ----
if __name__ == "__main__":
    run()