From 4686adf2666c942ce940b6a6dbc91e6f0fbd3859 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 03:15:42 -0400 Subject: [PATCH] =?UTF-8?q?feat(tree/repair):=20=E6=A3=80=E6=B5=8B?= =?UTF-8?q?=E5=99=A8=E6=89=A9=E5=B1=95=20=E2=80=94=20visible=5Fentities/on?= =?UTF-8?q?going=5Factions/spatial=5Flayout=20=E4=B8=BA=E7=A9=BA=E4=B9=9F?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/tree/repair/detector.py | 153 +++++++++++++++++++++++ tests/unit/test_repair_detector.py | 187 +++++++++++++++++++++++++++++ 2 files changed, 340 insertions(+) create mode 100644 app/tree/repair/detector.py create mode 100644 tests/unit/test_repair_detector.py diff --git a/app/tree/repair/detector.py b/app/tree/repair/detector.py new file mode 100644 index 0000000..ce62d99 --- /dev/null +++ b/app/tree/repair/detector.py @@ -0,0 +1,153 @@ +"""树修复检测器：扫描 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）", + ) + ) diff --git a/tests/unit/test_repair_detector.py b/tests/unit/test_repair_detector.py new file mode 100644 index 0000000..50bb7b0 --- /dev/null +++ b/tests/unit/test_repair_detector.py @@ -0,0 +1,187 @@ +"""修复检测器单元测试。""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from app.tree.index import ( + IndexMeta, + L1Card, + L1Node, + L2Card, + L2Node, + L3Card, + L3Node, + TreeIndex, +) +from app.tree.repair.detector import detect_issues + +if TYPE_CHECKING: + from pathlib import Path + + +def _card_l3(summary: str = "正常描述") -> L3Card: + return L3Card(summary, ["实体"], ["动作"], [], "居中", {}) + + +def _card_l2() -> L2Card: + return L2Card("事件", [], [], [], [], "", None) + + +def _card_l1() -> L1Card: + return L1Card("场景", "", [], [], [], [], "") + + +class TestDetectIssues: + def test_healthy_tree_no_issues(self) -> None: + l3 = L3Node(id="l1_0_l2_0_l3_0", card=_card_l3(), timestamp=1.0) + l2 = L2Node( + id="l1_0_l2_0", + card=_card_l2(), + time_range=(0.0, 10.0), + children=[l3], + ) + l1 = L1Node( + id="l1_0", + card=_card_l1(), + time_range=(0.0, 10.0), + children=[l2], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + assert detect_issues(index) == [] + + def test_empty_frame_summary(self) -> None: + l3 = L3Node(id="l1_0_l2_0_l3_0", card=_card_l3(""), timestamp=1.0) + l2 = L2Node( + id="l1_0_l2_0", + card=_card_l2(), + time_range=(0.0, 10.0), + children=[l3], + ) + l1 = L1Node( + id="l1_0", + card=_card_l1(), + time_range=(0.0, 10.0), + children=[l2], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "empty_field" and i.node_id == "l1_0_l2_0_l3_0" for i in issues) + + def test_missing_frame_file(self, tmp_path: Path) -> None: + l3 = L3Node( + id="l1_0_l2_0_l3_0", + card=_card_l3(), + timestamp=1.0, + frame_path="frames/missing.jpg", + ) + l2 = L2Node( + id="l1_0_l2_0", + card=_card_l2(), + time_range=(0.0, 10.0), + children=[l3], + ) + l1 = L1Node( + id="l1_0", + card=_card_l1(), + time_range=(0.0, 10.0), + children=[l2], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index, frames_dir=tmp_path) + assert any(i.issue_type == "missing_frame" for i in issues) + + def test_l2_no_children(self) -> None: + l2 = L2Node( + id="l1_0_l2_0", + card=_card_l2(), + time_range=(0.0, 10.0), + children=[], + ) + l1 = L1Node( + id="l1_0", + card=_card_l1(), + time_range=(0.0, 10.0), + children=[l2], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "no_children" and i.level == 2 for i in issues) + + def test_l1_no_children(self) -> None: + l1 = L1Node( + id="l1_0", + card=_card_l1(), + time_range=(0.0, 10.0), + children=[], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "no_children" and i.level == 1 for i in issues) + + def test_empty_visible_entities(self) -> None: + """visible_entities 为空也触发 empty_field。""" + card = L3Card("正常描述", [], ["动作"], [], "居中", {}) + l3 = L3Node(id="l1_0_l2_0_l3_0", card=card, timestamp=1.0) + l2 = L2Node(id="l1_0_l2_0", card=_card_l2(), time_range=(0.0, 10.0), children=[l3]) + l1 = L1Node(id="l1_0", card=_card_l1(), time_range=(0.0, 10.0), children=[l2]) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "empty_field" and "visible_entities" in i.details for i in issues) + + def test_empty_ongoing_actions(self) -> None: + """ongoing_actions 为空也触发 empty_field。""" + card = L3Card("正常描述", ["实体"], [], [], "居中", {}) + l3 = L3Node(id="l1_0_l2_0_l3_0", card=card, timestamp=1.0) + l2 = L2Node(id="l1_0_l2_0", card=_card_l2(), time_range=(0.0, 10.0), children=[l3]) + l1 = L1Node(id="l1_0", card=_card_l1(), time_range=(0.0, 10.0), children=[l2]) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "empty_field" and "ongoing_actions" in i.details for i in issues) + + def test_empty_spatial_layout(self) -> None: + """spatial_layout 为空也触发 empty_field。""" + card = L3Card("正常描述", ["实体"], ["动作"], [], "", {}) + l3 = L3Node(id="l1_0_l2_0_l3_0", card=card, timestamp=1.0) + l2 = L2Node(id="l1_0_l2_0", card=_card_l2(), time_range=(0.0, 10.0), children=[l3]) + l1 = L1Node(id="l1_0", card=_card_l1(), time_range=(0.0, 10.0), children=[l2]) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "empty_field" and "spatial_layout" in i.details for i in issues) + + def test_multiple_empty_fields_single_issue(self) -> None: + """多个字段同时为空只产生一个 issue，details 列出所有空字段。""" + card = L3Card("", [], [], [], "", {}) + l3 = L3Node(id="l1_0_l2_0_l3_0", card=card, timestamp=1.0) + l2 = L2Node(id="l1_0_l2_0", card=_card_l2(), time_range=(0.0, 10.0), children=[l3]) + l1 = L1Node(id="l1_0", card=_card_l1(), time_range=(0.0, 10.0), children=[l2]) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = [i for i in detect_issues(index) if i.issue_type == "empty_field"] + assert len(issues) == 1 + assert "frame_summary" in issues[0].details + assert "visible_entities" in issues[0].details + + def test_time_gap(self) -> None: + l3_a = L3Node(id="l1_0_l2_0_l3_0", card=_card_l3(), timestamp=1.0) + l3_b = L3Node(id="l1_0_l2_1_l3_0", card=_card_l3(), timestamp=20.0) + l2_a = L2Node( + id="l1_0_l2_0", + card=_card_l2(), + time_range=(0.0, 5.0), + children=[l3_a], + ) + l2_b = L2Node( + id="l1_0_l2_1", + card=_card_l2(), + time_range=(15.0, 25.0), + children=[l3_b], + ) + l1 = L1Node( + id="l1_0", + card=_card_l1(), + time_range=(0.0, 25.0), + children=[l2_a, l2_b], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + assert any(i.issue_type == "time_gap" for i in issues)