"""树修复检测器:扫描 TreeIndex 识别缺失/低质量节点。""" from __future__ import annotations from dataclasses import dataclass from typing import TYPE_CHECKING from loguru import logger if TYPE_CHECKING: from pathlib import Path from app.tree.index import TreeIndex # 相邻 L2 片段之间允许的最大时间间隙(秒) _MAX_TIME_GAP_S = 1.0 @dataclass(frozen=True) class NodeIssue: """检测到的节点问题。 参数: node_id: 问题节点 ID。 level: 节点层级(1/2/3)。 issue_type: 问题类型。 details: 详细描述。 """ node_id: str level: int issue_type: str # "empty_field" | "missing_frame" | "no_children" | "time_gap" details: str def detect_issues( index: TreeIndex, frames_dir: Path | None = None, ) -> list[NodeIssue]: """扫描树,返回所有问题节点列表。 检查项: - L3: card 必填字段为空(frame_summary / visible_entities / ongoing_actions / spatial_layout) - L3: frame_path 对应文件不存在(需提供 frames_dir) - L2/L1: children 列表为空 - L2: 相邻 clips 时间范围不连续(gap > 1秒) 参数: index: 待检测的 TreeIndex。 frames_dir: 帧文件根目录(可选,提供时检查帧文件存在性)。 返回: 问题列表,按 level 降序(L3 → L2 → L1)排列。 """ issues: list[NodeIssue] = [] for l1 in index.roots: # L1: children 不为空 if not l1.children: issues.append( NodeIssue( node_id=l1.id, level=1, issue_type="no_children", details="L1 节点无 L2 子节点", ) ) continue # L2: 相邻 clips 时间间隙检查 _check_time_gaps(l1.children, issues) for l2 in l1.children: # L2: children 不为空 if not l2.children: issues.append( NodeIssue( node_id=l2.id, level=2, issue_type="no_children", details="L2 节点无 L3 子节点", ) ) continue for l3 in l2.children: # L3: 各必填字段不为空 empty_fields: list[str] = [] if not l3.card.frame_summary: empty_fields.append("frame_summary") if not l3.card.visible_entities: empty_fields.append("visible_entities") if not l3.card.ongoing_actions: empty_fields.append("ongoing_actions") if not l3.card.spatial_layout: empty_fields.append("spatial_layout") if empty_fields: issues.append( NodeIssue( node_id=l3.id, level=3, issue_type="empty_field", details=f"L3 节点字段为空: {', '.join(empty_fields)}", ) ) # L3: frame_path 文件存在性 if ( frames_dir is not None and l3.frame_path is not None and not (frames_dir / l3.frame_path).exists() ): issues.append( NodeIssue( node_id=l3.id, level=3, issue_type="missing_frame", details=f"帧文件不存在: {l3.frame_path}", ) ) # 按 level 降序排列(L3=3 → L2=2 → L1=1) issues.sort(key=lambda i: -i.level) logger.info("树缺陷检测完成,发现 {} 个问题", len(issues)) return issues def _check_time_gaps( l2_nodes: list, issues: list[NodeIssue], ) -> None: """检查同一 L1 下相邻 L2 节点之间的时间间隙。 参数: l2_nodes: 同一 L1 节点下的 L2 子节点列表。 issues: 问题列表(原地追加)。 """ for i in range(len(l2_nodes) - 1): curr = l2_nodes[i] nxt = l2_nodes[i + 1] if curr.time_range is None or nxt.time_range is None: continue gap = nxt.time_range[0] - curr.time_range[1] if gap > _MAX_TIME_GAP_S: issues.append( NodeIssue( node_id=nxt.id, level=2, issue_type="time_gap", details=f"与前一片段间隙 {gap:.1f}s(阈值 {_MAX_TIME_GAP_S}s)", ) )