154 lines
4.8 KiB
Python
154 lines
4.8 KiB
Python
"""树修复检测器：扫描 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）",
|
||
)
|
||
)
|