From 12f20493c1cce9ab879ce22342a0b08419301dd2 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 01:57:32 -0400 Subject: [PATCH] =?UTF-8?q?feat(tree):=20=E8=B4=A8=E9=87=8F=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C=20=E2=80=94=20=E4=BA=A4=E5=8F=89=E9=AA=8C=E8=AF=81=20?= =?UTF-8?q?entities/visible=5Ftext?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) --- app/tree/verify.py | 291 ++++++++++++++++++++++++++++++++++++++ tests/unit/test_verify.py | 156 ++++++++++++++++++++ 2 files changed, 447 insertions(+) create mode 100644 app/tree/verify.py create mode 100644 tests/unit/test_verify.py diff --git a/app/tree/verify.py b/app/tree/verify.py new file mode 100644 index 0000000..524d9e5 --- /dev/null +++ b/app/tree/verify.py @@ -0,0 +1,291 @@ +"""质量校验模块：交叉验证树节点 Card 字段与子节点证据。 + +验证策略: +- L2 entities: 仅保留在子 L3 文本语料中模糊匹配到的实体。 +- L2 visible_text: 仅保留在子 L3 visible_text 中出现的条目。 +- L1 visible_text: 仅保留在后代 L2/L3 visible_text 中出现的条目。 +- L1 key_entities: 仅保留在后代 L2/L3 文本语料中模糊匹配到的实体。 + +Card 为 frozen dataclass，无法原地修改——移除幻觉字段时 +创建新 Card 实例并赋值给 node.card（Node 非 frozen）。 +""" + +from __future__ import annotations + +import string +from dataclasses import dataclass + +from loguru import logger + +from app.tree.index import ( + L1Card, + L1Node, + L2Card, + L2Node, + TreeIndex, +) + +# --------------------------------------------------------------------------- +# 校验统计 +# --------------------------------------------------------------------------- + + +@dataclass +class VerifyStats: + """校验统计信息。""" + + l2_entities_kept: int = 0 + l2_entities_removed: int = 0 + l2_visible_text_kept: int = 0 + l2_visible_text_removed: int = 0 + l1_visible_text_kept: int = 0 + l1_visible_text_removed: int = 0 + l1_key_entities_kept: int = 0 + l1_key_entities_removed: int = 0 + + +# --------------------------------------------------------------------------- +# 文本归一化 & 模糊匹配 +# --------------------------------------------------------------------------- + + +def _normalize(text: str) -> str: + """归一化文本：小写 + 去除标点。 + + 参数: + text: 原始文本。 + + 返回: + 归一化后的纯小写无标点字符串。 + """ + return text.lower().translate(str.maketrans("", "", string.punctuation)) + + +def fuzzy_match(entity: str | None, corpus: str | None) -> bool: + """模糊子串匹配：归一化后判断 entity 是否为 corpus 的子串。 + + 参数: + entity: 待匹配的实体文本（None 视为不匹配）。 + corpus: 证据语料文本（None 视为空）。 + + 返回: + True 表示匹配成功。 + """ + if not entity or not corpus: + return False + return _normalize(str(entity)) in _normalize(str(corpus)) + + +# --------------------------------------------------------------------------- +# 语料收集 +# --------------------------------------------------------------------------- + + +def _collect_l3_text(l2_node: L2Node) -> str: + """收集 L2 节点所有子 L3 的文本语料。 + + 从每个 L3 子节点的 card 和顶层字段中提取: + frame_summary、visible_text、subtitle。 + + 参数: + l2_node: L2 节点。 + + 返回: + 拼接后的文本语料（用换行分隔）。 + """ + parts: list[str] = [] + for l3 in l2_node.children: + parts.append(l3.card.frame_summary) + parts.extend(l3.card.visible_text) + if l3.subtitle: + parts.append(l3.subtitle) + return "\n".join(parts) + + +def _collect_descendant_visible_text(l1_node: L1Node) -> str: + """收集 L1 节点所有后代（L2/L3）的 visible_text。 + + 参数: + l1_node: L1 节点。 + + 返回: + 所有后代 visible_text 拼接后的文本（用换行分隔）。 + """ + parts: list[str] = [] + for l2 in l1_node.children: + parts.extend(l2.card.visible_text) + for l3 in l2.children: + parts.extend(l3.card.visible_text) + return "\n".join(parts) + + +def _collect_descendant_text_corpus(l1_node: L1Node) -> str: + """收集 L1 节点所有后代（L2/L3）的完整文本语料。 + + 用于 L1 key_entities 的交叉验证，范围包括 + L2/L3 的所有文本字段（frame_summary、visible_text、subtitle 等）。 + + 参数: + l1_node: L1 节点。 + + 返回: + 所有后代文本语料拼接后的文本（用换行分隔）。 + """ + parts: list[str] = [] + for l2 in l1_node.children: + parts.append(l2.card.event_description) + parts.extend(l2.card.entities) + parts.extend(l2.card.visible_text) + for l3 in l2.children: + parts.append(l3.card.frame_summary) + parts.extend(l3.card.visible_text) + if l3.subtitle: + parts.append(l3.subtitle) + return "\n".join(parts) + + +# --------------------------------------------------------------------------- +# 主校验函数 +# --------------------------------------------------------------------------- + + +def verify_tree(index: TreeIndex) -> VerifyStats: + """交叉验证视频树的 Card 字段与子节点证据，原地替换不合格的 Card。 + + Cards 为 frozen dataclass，移除幻觉字段时创建新 Card 实例 + 并赋值给 node.card。 + + 参数: + index: 树索引（会被原地修改）。 + + 返回: + VerifyStats 校验统计。 + """ + stats = VerifyStats() + + for l1 in index.roots: + # Phase 1: L2 字段验证 + for l2 in l1.children: + _verify_l2(l2, stats) + + # Phase 2: L1 字段验证 + _verify_l1(l1, stats) + + logger.info( + "verify_tree: source={} " + "l2_ent_kept={} l2_ent_rm={} " + "l2_vt_kept={} l2_vt_rm={} " + "l1_vt_kept={} l1_vt_rm={} " + "l1_ke_kept={} l1_ke_rm={}", + index.metadata.source_path, + stats.l2_entities_kept, + stats.l2_entities_removed, + stats.l2_visible_text_kept, + stats.l2_visible_text_removed, + stats.l1_visible_text_kept, + stats.l1_visible_text_removed, + stats.l1_key_entities_kept, + stats.l1_key_entities_removed, + ) + + return stats + + +def _verify_l2(l2: L2Node, stats: VerifyStats) -> None: + """校验单个 L2 节点的 entities 和 visible_text。 + + 参数: + l2: L2 节点（card 可能被替换）。 + stats: 统计对象（原地累加）。 + """ + corpus = _collect_l3_text(l2) + old_card = l2.card + + # entities: 模糊匹配过滤 + kept_entities = [e for e in old_card.entities if fuzzy_match(e, corpus)] + stats.l2_entities_kept += len(kept_entities) + stats.l2_entities_removed += len(old_card.entities) - len(kept_entities) + + # visible_text: 子 L3 visible_text 中必须存在 + l3_visible = _collect_l3_visible_text_set(l2) + kept_vt = [vt for vt in old_card.visible_text if _text_in_set(vt, l3_visible)] + stats.l2_visible_text_kept += len(kept_vt) + stats.l2_visible_text_removed += len(old_card.visible_text) - len(kept_vt) + + # 创建新 Card 替换（frozen dataclass） + l2.card = L2Card( + event_description=old_card.event_description, + entities=kept_entities, + actions=old_card.actions, + action_subjects=old_card.action_subjects, + visible_text=kept_vt, + spatial_relations=old_card.spatial_relations, + state_changes=old_card.state_changes, + ) + + +def _verify_l1(l1: L1Node, stats: VerifyStats) -> None: + """校验单个 L1 节点的 visible_text 和 key_entities。 + + 参数: + l1: L1 节点（card 可能被替换）。 + stats: 统计对象（原地累加）。 + """ + old_card = l1.card + + # visible_text: 必须出现在后代 L2/L3 visible_text 中 + descendant_vt = _collect_descendant_visible_text(l1) + kept_vt = [vt for vt in old_card.visible_text if fuzzy_match(vt, descendant_vt)] + stats.l1_visible_text_kept += len(kept_vt) + stats.l1_visible_text_removed += len(old_card.visible_text) - len(kept_vt) + + # key_entities: 交叉验证后代文本语料 + descendant_corpus = _collect_descendant_text_corpus(l1) + kept_ke = [ke for ke in old_card.key_entities if fuzzy_match(ke, descendant_corpus)] + stats.l1_key_entities_kept += len(kept_ke) + stats.l1_key_entities_removed += len(old_card.key_entities) - len(kept_ke) + + # 创建新 Card 替换（frozen dataclass） + l1.card = L1Card( + scene_summary=old_card.scene_summary, + main_setting=old_card.main_setting, + key_entities=kept_ke, + main_actions=old_card.main_actions, + topic_keywords=old_card.topic_keywords, + visible_text=kept_vt, + temporal_flow=old_card.temporal_flow, + ) + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _collect_l3_visible_text_set(l2: L2Node) -> set[str]: + """收集 L2 下所有 L3 子节点的 visible_text 归一化集合。 + + 参数: + l2: L2 节点。 + + 返回: + 归一化后的 visible_text 集合。 + """ + result: set[str] = set() + for l3 in l2.children: + for vt in l3.card.visible_text: + result.add(_normalize(vt)) + return result + + +def _text_in_set(text: str, normalized_set: set[str]) -> bool: + """检查文本归一化后是否存在于集合中。 + + 参数: + text: 待检查文本。 + normalized_set: 归一化后的文本集合。 + + 返回: + True 表示匹配成功。 + """ + return _normalize(text) in normalized_set diff --git a/tests/unit/test_verify.py b/tests/unit/test_verify.py new file mode 100644 index 0000000..ff85ae8 --- /dev/null +++ b/tests/unit/test_verify.py @@ -0,0 +1,156 @@ +"""质量校验模块单元测试。""" + +from __future__ import annotations + +from app.tree.index import ( + IndexMeta, + L1Card, + L1Node, + L2Card, + L2Node, + L3Card, + L3Node, + TreeIndex, +) +from app.tree.verify import VerifyStats, _normalize, fuzzy_match, verify_tree + + +class TestNormalize: + def test_lowercase(self): + assert _normalize("Hello World") == "hello world" + + def test_strip_punctuation(self): + assert _normalize("Hello, World!") == "hello world" + + def test_empty(self): + assert _normalize("") == "" + + +class TestFuzzyMatch: + def test_exact_match(self): + assert fuzzy_match("hello", "hello world") + + def test_case_insensitive(self): + assert fuzzy_match("Hello", "say hello world") + + def test_no_match(self): + assert not fuzzy_match("xyz", "hello world") + + def test_none_entity(self): + assert not fuzzy_match(None, "hello") + + def test_none_corpus(self): + assert not fuzzy_match("hello", None) + + +class TestVerifyTree: + def _make_tree(self): + """构造一棵树，L2 有混合实体（有出处/无出处）。""" + l3_0 = L3Node( + id="l1_0_l2_0_l3_0", + card=L3Card( + frame_summary="一个运动员在跑步", + visible_entities=["运动员", "跑道"], + ongoing_actions=["跑步"], + visible_text=["Nike", "2024"], + spatial_layout="居中", + visual_attributes={}, + ), + timestamp=1.0, + subtitle="the athlete is running fast", + ) + l3_1 = L3Node( + id="l1_0_l2_0_l3_1", + card=L3Card( + frame_summary="观众在欢呼", + visible_entities=["观众"], + ongoing_actions=["欢呼"], + visible_text=["Stadium"], + spatial_layout="广角", + visual_attributes={}, + ), + timestamp=3.0, + ) + l2 = L2Node( + id="l1_0_l2_0", + card=L2Card( + event_description="比赛片段", + entities=["运动员", "裁判", "幻觉实体"], # "裁判"和"幻觉实体"无 L3 出处 + actions=["跑步"], + action_subjects=["运动员"], + visible_text=["Nike", "不存在的文字"], # "不存在的文字"无 L3 出处 + spatial_relations="", + state_changes=None, + ), + time_range=(0.0, 10.0), + children=[l3_0, l3_1], + ) + l1 = L1Node( + id="l1_0", + card=L1Card( + scene_summary="体育比赛", + main_setting="体育场", + key_entities=["运动员", "不存在的人"], # "不存在的人"无出处 + main_actions=["比赛"], + topic_keywords=["体育"], + visible_text=["Nike", "Ghost"], # "Ghost"无出处 + temporal_flow="从左到右", + ), + time_range=(0.0, 10.0), + children=[l2], + ) + return TreeIndex(metadata=IndexMeta("/test.mp4", "video"), roots=[l1]) + + def test_removes_ungrounded_l2_entities(self): + index = self._make_tree() + stats = verify_tree(index) + l2 = index.roots[0].children[0] + assert "运动员" in l2.card.entities + assert "幻觉实体" not in l2.card.entities + assert stats.l2_entities_removed >= 1 + + def test_removes_ungrounded_l2_visible_text(self): + index = self._make_tree() + stats = verify_tree(index) + l2 = index.roots[0].children[0] + assert "Nike" in l2.card.visible_text + assert "不存在的文字" not in l2.card.visible_text + assert stats.l2_visible_text_removed >= 1 + + def test_removes_ungrounded_l1_visible_text(self): + index = self._make_tree() + stats = verify_tree(index) + l1 = index.roots[0] + assert "Nike" in l1.card.visible_text + assert "Ghost" not in l1.card.visible_text + assert stats.l1_visible_text_removed >= 1 + + def test_removes_ungrounded_l1_key_entities(self): + index = self._make_tree() + stats = verify_tree(index) + l1 = index.roots[0] + assert "运动员" in l1.card.key_entities + assert "不存在的人" not in l1.card.key_entities + assert stats.l1_key_entities_removed >= 1 + + def test_preserves_grounded_entities(self): + index = self._make_tree() + verify_tree(index) + l2 = index.roots[0].children[0] + assert "运动员" in l2.card.entities + + def test_returns_verify_stats(self): + index = self._make_tree() + stats = verify_tree(index) + assert isinstance(stats, VerifyStats) + total_kept = stats.l2_entities_kept + stats.l1_key_entities_kept + assert total_kept > 0 + + def test_frozen_card_replaced(self): + """验证 Card 被替换为新实例（frozen dataclass 不能原地修改）。""" + index = self._make_tree() + old_l2_card = index.roots[0].children[0].card + verify_tree(index) + new_l2_card = index.roots[0].children[0].card + # Card should be a different object if anything was removed + assert old_l2_card is not new_l2_card