From 7a2436a6126a53438de698c542e3c235bbdd7732 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Wed, 10 Jun 2026 00:50:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(asset-catalog):=20=E7=AE=80=E5=8C=96?= =?UTF-8?q?=E5=88=86=E6=9E=90=E5=B7=A5=E5=85=B7+=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E6=B1=87=E6=80=BB=E5=92=8CAI=E6=8E=A8=E7=90=86=E8=84=9A?= =?UTF-8?q?=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - analyze_meshes: 移除失败的 LineTrace 和六面分析，只保留基础属性 (包围盒、锚点、复杂度、Socket、材质槽位、自动Tag) - build_database: 新增离线汇总脚本，将 meshes/*.json 扁平化为 统一的 asset_database.json - ai_inference: 新增 GLM API 批量推理脚本，按类别分批推断语义信息 (朝向、表面高度、功能描述、交互属性、降级方案) Co-Authored-By: Claude Opus 4.8 --- tools/asset-catalog/ai_inference.py | 366 ++++++++++++++++++++++++++ tools/asset-catalog/analyze_meshes.py | 227 +--------------- tools/asset-catalog/build_database.py | 147 +++++++++++ 3 files changed, 518 insertions(+), 222 deletions(-) create mode 100644 tools/asset-catalog/ai_inference.py create mode 100644 tools/asset-catalog/build_database.py diff --git a/tools/asset-catalog/ai_inference.py b/tools/asset-catalog/ai_inference.py new file mode 100644 index 0000000..72da73c --- /dev/null +++ b/tools/asset-catalog/ai_inference.py @@ -0,0 +1,366 @@ +"""GLM API 批量推理脚本 — 资产语义增强。 + +读取 asset_database.json，按类别分批调用 GLM API（OpenAI 兼容接口）， +推断每个资产的朝向、功能、交互属性等语义信息。 + +此脚本不依赖 UE5，可离线运行。 + +环境变量: + GLM_API_KEY — API Key（必需） + GLM_API_BASE — API Base URL（默认 https://open.bigmodel.cn/api/paas/v4） + GLM_MODEL — 模型名（默认 glm-4-flash） + +用法: + python ai_inference.py [--input FILE] [--output FILE] [--batch-size N] [--dry-run] +""" + +import sys +import os +import json +import argparse +import time +from pathlib import Path + +_SCRIPT_DIR = Path(__file__).resolve().parent +if str(_SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPT_DIR)) + +from _common import read_json, write_json, iso_now + +TAG = "[AIInference]" + +# ---- 配置 ---- +DEFAULT_API_BASE = "https://open.bigmodel.cn/api/paas/v4" +DEFAULT_MODEL = "glm-4-flash" +DEFAULT_BATCH_SIZE = 20 + +# ---- Prompt 模板 ---- +SYSTEM_PROMPT = """你是一个 UE5 游戏开发助手，专注于为 3D 模型资产提取语义信息。 + +你会收到一批 StaticMesh 资产的几何数据（名称、尺寸、锚点、材质等）， +需要推断每个资产的以下语义字段： + +1. facing_direction: 主朝向方向 + - 格式: "+X", "-X", "+Y", "-Y", null + - 判断依据: 根据资产类型和包围盒尺寸推断"正面"方向 + - 对称物体（桌子、球、圆柱等）设为 null + - 电视/显示器: 屏幕面方向; 门: 开门方向; 椅子: 坐面方向 + +2. surface_height_cm: 可用水平表面高度 (cm) + - 如桌面高度、台面高度、床面高度 + - 对于无水平表面的物体（墙体、栏杆）设为 null + - 使用锚点类型判断: CENTER_BOTTOM 时直接取 max.z + +3. function_description: 一句话功能描述 (中文, ≤50字) + +4. placement_notes: 放置注意事项 (中文, ≤80字) + - 锚点位置提醒、旋转建议、缩放注意等 + +5. interaction_hints: 交互提示标签数组 + - 可选值: can_sit, can_place_items, can_open, can_interact_screen, + climbable, electronics, lighting, container, decoration, + pass_through, driveable, weapon, vegetation + - 可多选 + +6. plausible_substitutes: BSP 基础体降级方案 (中文, 或 null) + - 当原始资产不可用时，用什么 Cube/Cylinder + 缩放 + 材质替代 + +你必须返回严格的 JSON 数组，每个元素对应输入中的一个资产，顺序一致。 +不要输出任何 JSON 之外的内容。""" + +BATCH_PROMPT_TEMPLATE = """以下是 {category} 类别的 {count} 个 StaticMesh 资产数据： + +{assets_json} + +请为每个资产推断语义信息，返回 JSON 数组。每个元素包含: +facing_direction, surface_height_cm, function_description, +placement_notes, interaction_hints, plausible_substitutes""" + + +def _get_api_config(): + """从环境变量读取 API 配置。""" + api_key = os.environ.get("GLM_API_KEY", "") + api_base = os.environ.get("GLM_API_BASE", DEFAULT_API_BASE) + model = os.environ.get("GLM_MODEL", DEFAULT_MODEL) + + if not api_key: + print(f"{TAG} 错误: 未设置 GLM_API_KEY 环境变量") + print(f"{TAG} 设置方法: set GLM_API_KEY=your_key_here") + return None + + return { + "api_key": api_key, + "api_base": api_base, + "model": model, + } + + +def _classify_batch_category(tags: dict) -> str: + """根据 tags 推断批次类别名。""" + style = tags.get("style", []) + if not style: + return "未分类" + # 取第一个非空 style tag 作为类别 + return style[0] if style[0] != "unknown" else "未分类" + + +def _prepare_batches(assets: dict, batch_size: int) -> list: + """将资产按类别分批。 + + Returns: + list of (category_name, batch_index, [asset_keys]) + """ + # 按类别分组 + by_category = {} + for name, data in assets.items(): + cat = _classify_batch_category(data.get("tags", {})) + by_category.setdefault(cat, []).append(name) + + # 每个类别内按 batch_size 切分 + batches = [] + for cat, keys in sorted(by_category.items()): + for i in range(0, len(keys), batch_size): + batch_keys = keys[i:i + batch_size] + batches.append((cat, i // batch_size + 1, batch_keys)) + + return batches + + +def _format_asset_for_prompt(name: str, data: dict) -> dict: + """提取 prompt 中需要的精简字段。""" + bb = data.get("bounding_box", {}) + pivot = data.get("pivot", {}) + tags = data.get("tags", {}) + + return { + "name": name, + "size_cm": bb.get("size"), + "pivot_type": pivot.get("type"), + "pivot_offset_xy": pivot.get("offset_xy"), + "material_count": len(data.get("material_slots", [])), + "socket_count": len(data.get("sockets", [])), + "tags_style": tags.get("style", []), + } + + +def _call_glm_api(config: dict, category: str, assets_json: str) -> list: + """调用 GLM API 进行推理。 + + Returns: + 推理结果列表，失败时返回空列表 + """ + try: + from openai import OpenAI + except ImportError: + print(f"{TAG} 错误: 需要安装 openai 包") + print(f"{TAG} pip install openai") + return [] + + client = OpenAI( + api_key=config["api_key"], + base_url=config["api_base"], + ) + + user_prompt = BATCH_PROMPT_TEMPLATE.format( + category=category, + count=assets_json.count('"name"'), + assets_json=assets_json, + ) + + try: + response = client.chat.completions.create( + model=config["model"], + messages=[ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": user_prompt}, + ], + temperature=0.1, + response_format={"type": "json_object"}, + ) + except Exception as e: + print(f"{TAG} API 调用失败: {e}") + return [] + + content = response.choices[0].message.content.strip() + if not content: + print(f"{TAG} API 返回空内容") + return [] + + # 解析 JSON + try: + result = json.loads(content) + # 如果返回的是 {"results": [...]} 形式，提取数组 + if isinstance(result, dict): + for key in ("results", "data", "assets", "items"): + if key in result and isinstance(result[key], list): + return result[key] + # 如果找不到数组，可能是单条结果被包在 dict 里 + print(f"{TAG} 警告: 返回 JSON 不是数组，尝试整体作为单条: {list(result.keys())}") + return [result] + if isinstance(result, list): + return result + except json.JSONDecodeError as e: + print(f"{TAG} JSON 解析失败: {e}") + print(f"{TAG} 原始响应前 200 字: {content[:200]}") + return [] + + return [] + + +def run_inference( + input_file: Path, + output_file: Path, + batch_size: int = DEFAULT_BATCH_SIZE, + dry_run: bool = False, +) -> dict: + """执行批量 AI 推理。 + + Args: + input_file: asset_database.json 路径 + output_file: 输出路径（通常与 input 相同，就地更新） + batch_size: 每批资产数量 + dry_run: 只打印批次信息，不调用 API + + Returns: + 统计信息 dict + """ + # 读取数据库 + if not input_file.exists(): + print(f"{TAG} 错误: {input_file} 不存在，请先运行 build_database") + return {"status": "error", "reason": "no input file"} + + db = read_json(input_file) + assets = db.get("assets", {}) + if not assets: + print(f"{TAG} 错误: 数据库中无资产") + return {"status": "error", "reason": "empty assets"} + + # 过滤: 跳过已推理的 + pending = {} + for name, data in assets.items(): + if data.get("ai_inference") is not None: + continue + pending[name] = data + + print(f"{TAG} 数据库共 {len(assets)} 个资产，待推理 {len(pending)} 个") + + if not pending: + print(f"{TAG} 全部已推理，无需处理") + return {"status": "ok", "total": len(assets), "inferred": 0, "skipped": len(assets)} + + # 分批 + batches = _prepare_batches(pending, batch_size) + print(f"{TAG} 分为 {len(batches)} 批") + + if dry_run: + for cat, idx, keys in batches: + print(f"{TAG} 批次 [{cat}] #{idx}: {len(keys)} 个资产") + for k in keys[:3]: + print(f"{TAG} - {k}") + if len(keys) > 3: + print(f"{TAG} ... 还有 {len(keys) - 3} 个") + return {"status": "ok", "dry_run": True, "batches": len(batches)} + + # 获取 API 配置 + config = _get_api_config() + if not config: + return {"status": "error", "reason": "no api key"} + + print(f"{TAG} API: {config['api_base']} 模型: {config['model']}") + + # 逐批推理 + total_inferred = 0 + total_failed = 0 + + for batch_idx, (cat, batch_num, keys) in enumerate(batches): + print(f"{TAG} 批次 [{batch_idx + 1}/{len(batches)}] [{cat}] #{batch_num}: {len(keys)} 个资产...") + + # 构造 prompt 数据 + prompt_assets = [] + for k in keys: + prompt_assets.append(_format_asset_for_prompt(k, pending[k])) + + assets_json = json.dumps(prompt_assets, ensure_ascii=False, indent=2) + + # 调用 API + results = _call_glm_api(config, cat, assets_json) + + if not results: + print(f"{TAG} ⚠️ 推理失败，跳过此批") + total_failed += len(keys) + continue + + # 合并结果 + matched = 0 + for i, k in enumerate(keys): + if i < len(results): + assets[k]["ai_inference"] = results[i] + matched += 1 + total_inferred += 1 + else: + print(f"{TAG} ⚠️ {k}: API 返回结果不足") + + print(f"{TAG} ✅ 匹配 {matched}/{len(keys)}") + + # 每批都保存一次（断点续跑） + db["assets"] = assets + # 更新 metadata + if "ai_inference_metadata" not in db["metadata"]: + db["metadata"]["ai_inference_metadata"] = { + "model": config["model"], + "api_base": config["api_base"], + "started_at": iso_now(), + } + db["metadata"]["ai_inference_metadata"]["last_batch"] = batch_idx + 1 + db["metadata"]["ai_inference_metadata"]["total_batches"] = len(batches) + db["metadata"]["ai_inference_metadata"]["inferred"] = total_inferred + db["metadata"]["ai_inference_metadata"]["failed"] = total_failed + write_json(db, output_file) + + # 限速: 避免触发 API 限流 + if batch_idx < len(batches) - 1: + time.sleep(1) + + db["metadata"]["ai_inference_metadata"]["completed_at"] = iso_now() + write_json(db, output_file) + + print(f"{TAG} 推理完成: {total_inferred} 成功, {total_failed} 失败") + return { + "status": "ok", + "total": len(assets), + "inferred": total_inferred, + "failed": total_failed, + } + + +def main(): + parser = argparse.ArgumentParser(description="GLM API 批量推理 — 资产语义增强") + parser.add_argument( + "--input", type=str, + default=str(_SCRIPT_DIR.parent.parent / "output" / "asset-catalog" / "asset_database.json"), + help="输入 asset_database.json 路径", + ) + parser.add_argument( + "--output", type=str, default=None, + help="输出路径 (默认与 input 相同，就地更新)", + ) + parser.add_argument( + "--batch-size", type=int, default=DEFAULT_BATCH_SIZE, + help=f"每批资产数量 (默认 {DEFAULT_BATCH_SIZE})", + ) + parser.add_argument( + "--dry-run", action="store_true", + help="只打印批次信息，不调用 API", + ) + args = parser.parse_args() + + input_file = Path(args.input) + output_file = Path(args.output) if args.output else input_file + + result = run_inference(input_file, output_file, args.batch_size, args.dry_run) + if result.get("status") != "ok" and not result.get("dry_run"): + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/tools/asset-catalog/analyze_meshes.py b/tools/asset-catalog/analyze_meshes.py index d8e5ce1..037e80a 100644 --- a/tools/asset-catalog/analyze_meshes.py +++ b/tools/asset-catalog/analyze_meshes.py @@ -6,8 +6,6 @@ - 网格复杂度（顶点数、三角形数） - Socket 列表 - 材质槽位 -- 包围盒 6 面候选 -- 射线检测细节包围盒（spawn + Line Trace） - AI 占位字段（key_surfaces, style_tags, scene_tags 等） 输出: output/meshes/.json，每个子目录一个文件 @@ -18,7 +16,6 @@ """ import sys -import math from pathlib import Path _SCRIPT_DIR = Path(__file__).resolve().parent @@ -28,7 +25,7 @@ if str(_SCRIPT_DIR) not in sys.path: from _common import ( HAS_UNREAL, get_scan_root, get_output_dir, get_steps, write_json, read_json, wait_for_editor, progress_log, - iso_now, safe_subfolder_name, classify_pivot, compute_bbox_faces, + iso_now, safe_subfolder_name, classify_pivot, ) TAG = "[MeshAnalysis]" @@ -263,12 +260,9 @@ def _analyze_single_mesh(asset_info: dict, subfolder: str = "") -> dict: "mesh_complexity": {}, "sockets": [], "material_slots": [], - "detail_boxes": [], - "bbox_faces": [], "analysis_notes": [], # AI Pass 2 占位 "key_surfaces": [], - "detail_boxes_annotated": [], "ground_sink_estimate": None, "style_tags": [], "scene_tags": [], @@ -301,39 +295,20 @@ def _analyze_single_mesh(asset_info: dict, subfolder: str = "") -> dict: # ---- 2. 锚点 ---- result["pivot"] = classify_pivot(bmin, bmax) - # ---- 3. 包围盒面 ---- - result["bbox_faces"] = compute_bbox_faces(bmin, bmax) - except Exception as e: result["analysis_notes"].append(f"包围盒提取失败: {e}") return result - # ---- 4. 网格复杂度 ---- + # ---- 3. 网格复杂度 ---- result["mesh_complexity"] = _try_get_complexity(mesh) - # ---- 5. Sockets ---- + # ---- 4. Sockets ---- result["sockets"] = _try_get_sockets(mesh) - # ---- 6. 材质槽位 ---- + # ---- 5. 材质槽位 ---- result["material_slots"] = _try_get_material_slots(mesh) - # ---- 7. 细节包围盒 ---- - detail_boxes, trace_debug = _try_trace_detail_boxes(mesh, bmin, bmax) - if trace_debug: - result["analysis_notes"].extend(trace_debug) - if detail_boxes: - result["detail_boxes"] = detail_boxes - else: - # 降级: 仅整体包围盒 - result["detail_boxes"] = [{ - "label": "overall", - "min": _round3(list(bmin)), - "max": _round3(list(bmax)), - "size": _round3(list(size)), - "source": "overall", - }] - - # ---- 8. 自动 Tags ---- + # ---- 6. 自动 Tags ---- style_tags, scene_tags = _generate_mesh_tags(name, subfolder) result["style_tags"] = style_tags result["scene_tags"] = scene_tags @@ -341,198 +316,6 @@ def _analyze_single_mesh(asset_info: dict, subfolder: str = "") -> dict: return result -# ============================================================ -# Line Trace 细节包围盒 -# ============================================================ - -# UE5.7 Python correct enum names (discovered via probe) -_TRACE_CHANNEL = None -_DRAW_DEBUG = None - -def _get_trace_enums(): - """Lazy-init: get the correct UE5.7 trace enums.""" - global _TRACE_CHANNEL, _DRAW_DEBUG - if _TRACE_CHANNEL is not None: - return _TRACE_CHANNEL is not None - import unreal - try: - _TRACE_CHANNEL = unreal.TraceTypeQuery.TRACE_TYPE_QUERY1 - _DRAW_DEBUG = unreal.DrawDebugTrace.NONE - return True - except Exception: - return False - - -def _try_trace_detail_boxes(mesh, bmin: tuple, bmax: tuple) -> tuple: - """通过 Line Trace 获取细节包围盒。 - - Returns: - (detail_boxes, debug_notes) - """ - import unreal - - detail_boxes = [] - debug = [] - - size = (bmax[0] - bmin[0], bmax[1] - bmin[1], bmax[2] - bmin[2]) - if max(size) < 10: - return [], ["skip_trace: model too small"] - - # Check trace API availability - if not _get_trace_enums(): - return [], ["trace_unavailable: enums not found"] - - # ---- Spawn temporary Actor ---- - try: - editor_subsystem = unreal.get_editor_subsystem(unreal.EditorActorSubsystem) - if not editor_subsystem: - return [], ["trace_fail: no subsystem"] - except Exception as e: - return [], [f"trace_fail: {e}"] - - try: - actor = editor_subsystem.spawn_actor_from_object(mesh, unreal.Vector(0, 0, 0)) - except Exception as e: - return [], [f"trace_fail: spawn {e}"] - - if not actor: - return [], ["trace_fail: spawn None"] - - # Set collision to BlockAll so traces can hit this mesh - smc = actor.get_component_by_class(unreal.StaticMeshComponent) - if smc: - try: - smc.set_collision_enabled(unreal.CollisionEnabled.QUERY_AND_PHYSICS) - smc.set_collision_profile_name("BlockAll") - except Exception as e: - debug.append(f"collision_set_warn: {e}") - - try: - hit_points = _perform_line_traces(actor, bmin, bmax) - debug.append(f"trace_hits: {len(hit_points)}") - - if len(hit_points) >= 4: - detail_boxes = _cluster_hit_points(hit_points, bmin, bmax) - if detail_boxes: - debug.append(f"trace_boxes: {len(detail_boxes)}") - finally: - try: - unreal.EditorLevelLibrary.destroy_actor(actor) - except Exception: - pass - - return detail_boxes, debug - - -def _perform_line_traces(actor, bmin: tuple, bmax: tuple) -> list: - """多方向 Line Trace 收集命中点。""" - import unreal - - hit_points = [] - mn_x, mn_y, mn_z = bmin - mx_x, mx_y, mx_z = bmax - sx, sy, sz = mx_x - mn_x, mx_y - mn_y, mx_z - mn_z - - step = min(50, max(10, min(sx, sy) / 5)) - margin = max(20, max(sx, sy, sz) * 0.1) - z_step = min(50, max(15, sz / 4)) - - trace_ch = _TRACE_CHANNEL - draw_dbg = _DRAW_DEBUG - - def _trace(start_v, end_v): - try: - hit = unreal.SystemLibrary.line_trace_single( - actor, start_v, end_v, - trace_ch, False, [], draw_dbg, - ) - if hit and hasattr(hit, 'impact_point'): - pt = hit.impact_point - hit_points.append((pt.x, pt.y, pt.z)) - except Exception: - pass - - try: - # Top-down (Z axis) - z = mn_z - while z <= mx_z + 0.1: - x = mn_x + step / 2 - while x <= mx_x + 0.1: - y = mn_y + step / 2 - while y <= mx_y + 0.1: - _trace(unreal.Vector(x, y, mx_z + margin), - unreal.Vector(x, y, mn_z - margin)) - y += step - x += step - z += z_step - - # Front (Y+) - x = mn_x + step / 2 - while x <= mx_x + 0.1: - z = mn_z + step / 2 - while z <= mx_z + 0.1: - _trace(unreal.Vector(x, mx_y + margin, z), - unreal.Vector(x, mn_y - margin, z)) - z += step - x += step - - # Right (X+) - y = mn_y + step / 2 - while y <= mx_y + 0.1: - z = mn_z + step / 2 - while z <= mx_z + 0.1: - _trace(unreal.Vector(mx_x + margin, y, z), - unreal.Vector(mn_x - margin, y, z)) - z += step - y += step - except Exception: - pass - - return hit_points - - -def _cluster_hit_points(hit_points: list, bmin: tuple, bmax: tuple) -> list: - """将命中点按 Z 轴聚类为细节包围盒。""" - if not hit_points: - return [] - - sorted_pts = sorted(hit_points, key=lambda p: p[2]) - z_slices = [] - current_slice = [sorted_pts[0]] - z_gap = 10.0 - - for i in range(1, len(sorted_pts)): - if sorted_pts[i][2] - current_slice[-1][2] > z_gap: - z_slices.append(current_slice) - current_slice = [sorted_pts[i]] - else: - current_slice.append(sorted_pts[i]) - z_slices.append(current_slice) - - boxes = [] - for idx, pts in enumerate(z_slices): - if len(pts) < 3: - continue - xs = [p[0] for p in pts] - ys = [p[1] for p in pts] - zs = [p[2] for p in pts] - bmn = [min(xs), min(ys), min(zs)] - bmx = [max(xs), max(ys), max(zs)] - bsz = [bmx[0]-bmn[0], bmx[1]-bmn[1], bmx[2]-bmn[2]] - if bsz[0] < 2 and bsz[1] < 2: - continue - boxes.append({ - "label": f"part_{idx}", - "min": _round3(bmn), "max": _round3(bmx), "size": _round3(bsz), - "source": "trace", "hit_count": len(pts), - }) - - if len(boxes) > 8: - boxes.sort(key=lambda b: b["size"][0]*b["size"][1]*b["size"][2], reverse=True) - boxes = boxes[:8] - return boxes - - # ============================================================ # 主流程 # ============================================================ diff --git a/tools/asset-catalog/build_database.py b/tools/asset-catalog/build_database.py new file mode 100644 index 0000000..919f13b --- /dev/null +++ b/tools/asset-catalog/build_database.py @@ -0,0 +1,147 @@ +"""资产数据库汇总脚本。 + +读取 asset-catalog 的扫描结果（meshes/*.json），汇总为统一的单文件 +asset_database.json，以资产名为 key 扁平化存储。 + +此脚本不依赖 UE5，可离线运行。 + +用法: + python build_database.py [--input DIR] [--output FILE] + + --input 扫描输出目录，默认 output/asset-catalog + --output 输出文件路径，默认 /asset_database.json +""" + +import sys +import argparse +from pathlib import Path + +_SCRIPT_DIR = Path(__file__).resolve().parent +if str(_SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPT_DIR)) + +from _common import read_json, write_json, iso_now + +TAG = "[BuildDB]" + + +def build_database(input_dir: Path, output_file: Path) -> dict: + """汇总 meshes/*.json 为统一数据库。 + + Args: + input_dir: 包含 meshes/ 子目录的扫描输出根目录 + output_file: 输出的 asset_database.json 路径 + + Returns: + 统计信息 dict + """ + meshes_dir = input_dir / "meshes" + if not meshes_dir.exists(): + print(f"{TAG} 错误: {meshes_dir} 不存在，请先运行 analyze_meshes") + return {"status": "error", "reason": "no meshes dir"} + + # 收集所有子目录的扫描文件 + mesh_files = sorted(meshes_dir.glob("*.json")) + if not mesh_files: + print(f"{TAG} 错误: {meshes_dir} 下无 JSON 文件") + return {"status": "error", "reason": "no mesh files"} + + print(f"{TAG} 发现 {len(mesh_files)} 个扫描文件") + + # 扁平化合并 + assets = {} + total = 0 + skipped = 0 + + for mf in mesh_files: + try: + data = read_json(mf) + except Exception as e: + print(f"{TAG} 警告: 读取 {mf.name} 失败: {e}") + continue + + subfolder = data.get("subfolder", mf.stem) + meshes = data.get("meshes", {}) + if not meshes: + continue + + for name, mesh_data in meshes.items(): + total += 1 + + # 跳过无包围盒的（加载失败的） + if mesh_data.get("bounding_box") is None: + skipped += 1 + continue + + # 构建统一条目 + entry = { + "ue_path": mesh_data.get("ue_path", ""), + "subfolder": subfolder, + "bounding_box": mesh_data.get("bounding_box"), + "pivot": mesh_data.get("pivot"), + "mesh_complexity": mesh_data.get("mesh_complexity", {}), + "material_slots": mesh_data.get("material_slots", []), + "sockets": mesh_data.get("sockets", []), + "tags": { + "style": mesh_data.get("style_tags", []), + "scene": mesh_data.get("scene_tags", []), + }, + "ai_inference": None, + } + + # 如果资产名重复（不同子目录下同名），加后缀 + key = name + if key in assets: + safe_sub = subfolder.replace("/", "_").replace("\\", "_") + key = f"{name}_{safe_sub}" + print(f"{TAG} 注意: 资产名重复 {name}，使用 {key}") + + assets[key] = entry + + # 写入数据库 + db = { + "metadata": { + "generated_at": iso_now(), + "source": "asset_catalog_v1", + "total_assets": len(assets), + "skipped_no_bbox": skipped, + "input_files": [mf.name for mf in mesh_files], + }, + "assets": assets, + } + + write_json(db, output_file) + print(f"{TAG} ✅ 汇总完成: {len(assets)} 个资产 → {output_file}") + print(f"{TAG} 跳过 {skipped} 个无包围盒资产 / 总计 {total}") + + return { + "status": "ok", + "total": total, + "assets": len(assets), + "skipped": skipped, + } + + +def main(): + parser = argparse.ArgumentParser(description="汇总资产扫描数据为统一 JSON 数据库") + parser.add_argument( + "--input", type=str, + default=str(_SCRIPT_DIR.parent.parent / "output" / "asset-catalog"), + help="扫描输出目录 (默认: output/asset-catalog)", + ) + parser.add_argument( + "--output", type=str, default=None, + help="输出文件路径 (默认: /asset_database.json)", + ) + args = parser.parse_args() + + input_dir = Path(args.input) + output_file = Path(args.output) if args.output else input_dir / "asset_database.json" + + result = build_database(input_dir, output_file) + if result.get("status") != "ok": + sys.exit(1) + + +if __name__ == "__main__": + main()