292 lines
8.8 KiB
Python
292 lines
8.8 KiB
Python
"""质量校验模块：交叉验证树节点 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
|