90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
"""诊断侧树读取适配：把嵌套 tree.json 展平成诊断消费的扁平 nodes dict。
|
||
|
||
诊断编排（core/evolution/diagnose.py）期望 tree_data 形如
|
||
{"nodes": {node_id: {card, level, time_range}}}，但 TRM5 建树产物
|
||
store/videos/<vid>/tree.json 是嵌套 {"metadata","roots":[...]}。本模块递归展平，
|
||
接通 TRM4→TRM5 迁移时断掉的 ground_truth 加载环。
|
||
|
||
不走 TreeIndex 对象层：仅 L1Node 有 to_dict（app/tree/index.py:260），L2/L3 为其内部闭包，
|
||
且 to_dict 输出无 level、L3 用 timestamp 无 time_range。直接遍历 json 更省且零改建树模块。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
if TYPE_CHECKING:
|
||
from pathlib import Path
|
||
|
||
|
||
def load_tree_nodes(store_dir: Path, video_id: str) -> dict[str, Any]:
|
||
"""加载单视频 tree.json 并展平成扁平 nodes dict。
|
||
|
||
参数:
|
||
store_dir: store 根目录（含 videos/<video_id>/tree.json）。
|
||
video_id: 视频标识。
|
||
|
||
返回:
|
||
{"nodes": {node_id: {"card": dict, "level": int, "time_range": list}}}。
|
||
|
||
异常:
|
||
FileNotFoundError: tree.json 不存在（沿用 factory.py fail-loud 先例）。
|
||
ValueError: roots 非 list 或为空、节点缺 id、或节点既无 time_range 又无 timestamp。
|
||
|
||
关键实现:
|
||
level 由遍历深度赋值（root=1/child=2/孙=3），不解析 node_id——node_id 累积式
|
||
（..._L1_..._L2_..._L3_）用正则首匹配会把 L2/L3 误判成 1。
|
||
L3 无 time_range，用 timestamp 合成 [t, t]。
|
||
"""
|
||
tree_path = store_dir / "videos" / video_id / "tree.json"
|
||
if not tree_path.exists():
|
||
raise FileNotFoundError(f"树索引文件不存在: {tree_path}（诊断需真实树，P5 fail loud）")
|
||
tree = json.loads(tree_path.read_text(encoding="utf-8"))
|
||
roots = tree.get("roots")
|
||
if not isinstance(roots, list) or not roots:
|
||
raise ValueError(f"树无有效 roots: {tree_path}")
|
||
|
||
nodes: dict[str, Any] = {}
|
||
|
||
def _walk(node: dict[str, Any], level: int) -> None:
|
||
node_id = node.get("id")
|
||
if not isinstance(node_id, str) or not node_id:
|
||
raise ValueError(f"节点缺 id: {tree_path}")
|
||
time_range = node.get("time_range")
|
||
if time_range is None:
|
||
ts = node.get("timestamp")
|
||
if ts is None:
|
||
raise ValueError(
|
||
f"节点既无 time_range 又无 timestamp（树损坏）: {node_id} in {tree_path}"
|
||
)
|
||
time_range = [ts, ts]
|
||
nodes[node_id] = {
|
||
"card": node.get("card", {}),
|
||
"level": level,
|
||
"time_range": time_range,
|
||
}
|
||
for child in node.get("children", []) or []:
|
||
_walk(child, level + 1)
|
||
|
||
for root in roots:
|
||
_walk(root, 1)
|
||
|
||
return {"nodes": nodes}
|
||
|
||
|
||
def load_tree_data_for_videos(store_dir: Path, video_ids: list[str]) -> dict[str, Any]:
|
||
"""按一组 video_id 去重加载展平树，供诊断按 video 注入。
|
||
|
||
参数:
|
||
store_dir: store 根目录。
|
||
video_ids: 视频标识列表（可含重复，内部按首次出现顺序去重）。
|
||
|
||
返回:
|
||
{video_id: {"nodes": {...}}}。
|
||
|
||
异常:
|
||
同 load_tree_nodes（任一视频树缺失/无效即 fail-loud）。
|
||
"""
|
||
return {vid: load_tree_nodes(store_dir, vid) for vid in dict.fromkeys(video_ids)}
|