iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(question_gen): reduce cyclomatic complexity in sampler_v2

Browse Source 
Extract shared _resolve_subtree helper to eliminate repeated tri-level
node resolution. Break _validate_sampling_constraints into focused
single-purpose helpers:
- _count_l3_descendants
- _has_frames
- _count_subtitles
- _resolve_subtree / _find_l3_parent

Extract _subtitles_from_l2_list and _frames_from_l2_list to simplify
collection functions.

Complexity improvements:
- _validate_sampling_constraints: D(23) -> B(8)
- _collect_subtitle_sentences: C(16) -> A(3)
- _collect_frame_paths: C(13) -> A(2)

All functions now grade B or better per radon cc.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-11 23:33:48 -04:00
parent f74711cd11
commit 9f739e831d
1 changed files with 152 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/question_gen/sampler_v2.py
+152 -108
View File
@@ -95,6 +95,19 @@ class MaterialContext:
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class _ResolvedSubtree:
"""节点解析结果 — 将任意层级 node_id 统一解析为 L2 节点列表 + 父 L1。
属性:
l2_nodes: 与目标节点关联的 L2 节点列表。
parent_l1: 父 L1 节点（用于跨 L2 判断）。
"""
l2_nodes: list[L2Node]
parent_l1: L1Node | None
def _find_l2_node(tree: TreeIndex, l2_id: str) -> tuple[L2Node, L1Node] | None:
"""按 ID 定位 L2 节点及其父 L1。
@@ -128,6 +141,100 @@ def _find_l1_node(tree: TreeIndex, l1_id: str) -> L1Node | None:
return None
def _find_l3_parent(tree: TreeIndex, l3_id: str) -> tuple[L2Node, L1Node] | None:
"""按 L3 ID 定位其所属的 L2 和 L1 节点。
参数:
tree: 三层树索引。
l3_id: L3 节点 ID。
返回:
(L2Node, L1Node) 元组；未找到返回 None。
"""
for l1 in tree.roots:
for l2 in l1.children:
for l3 in l2.children:
if l3.id == l3_id:
return (l2, l1)
return None
def _resolve_subtree(tree: TreeIndex, node_id: str) -> _ResolvedSubtree | None:
"""将任意层级节点 ID 解析为关联的 L2 节点列表和父 L1。
参数:
tree: 三层树索引。
node_id: 任意层级的节点 ID。
返回:
_ResolvedSubtree 实例；节点不存在时返回 None。
"""
# 尝试作为 L2
result = _find_l2_node(tree, node_id)
if result is not None:
l2_node, parent_l1 = result
return _ResolvedSubtree(l2_nodes=[l2_node], parent_l1=parent_l1)
# 尝试作为 L1
l1_node = _find_l1_node(tree, node_id)
if l1_node is not None:
return _ResolvedSubtree(l2_nodes=list(l1_node.children), parent_l1=l1_node)
# 尝试作为 L3
l3_result = _find_l3_parent(tree, node_id)
if l3_result is not None:
l2_node, parent_l1 = l3_result
return _ResolvedSubtree(l2_nodes=[l2_node], parent_l1=parent_l1)
return None
# ---------------------------------------------------------------------------
# 约束检查辅助（单一职责）
# ---------------------------------------------------------------------------
def _count_l3_descendants(l2_nodes: list[L2Node]) -> int:
"""统计 L2 节点列表下的 L3 总数。
参数:
l2_nodes: L2 节点列表。
返回:
L3 节点总数。
"""
return sum(len(l2.children) for l2 in l2_nodes)
def _has_frames(l2_nodes: list[L2Node]) -> bool:
"""检查 L2 节点列表的子树中是否有可用帧。
参数:
l2_nodes: L2 节点列表。
返回:
True 表示至少有一个 L3 有 frame_path。
"""
return any(l3.frame_path for l2 in l2_nodes for l3 in l2.children)
def _count_subtitles(l2_nodes: list[L2Node]) -> int:
"""统计 L2 节点列表中全部字幕数（L2 + L3）。
参数:
l2_nodes: L2 节点列表。
返回:
非空字幕总数。
"""
count = 0
for l2 in l2_nodes:
if l2.card.subtitle:
count += 1
count += sum(1 for l3 in l2.children if l3.card.subtitle)
return count
# ---------------------------------------------------------------------------
# 公开辅助函数
# ---------------------------------------------------------------------------
@@ -151,62 +258,39 @@ def _validate_sampling_constraints(
返回:
True 表示满足所有约束，False 表示至少一项不满足。
"""
# Phase 1: 确定目标 L2 节点列表
target_l2_nodes: list[L2Node] = []
parent_l1: L1Node | None = None
# 先尝试作为 L2
result = _find_l2_node(tree, node_id)
if result is not None:
l2_node, parent_l1 = result
target_l2_nodes = [l2_node]
else:
# 尝试作为 L1
l1_node = _find_l1_node(tree, node_id)
if l1_node is not None:
target_l2_nodes = list(l1_node.children)
parent_l1 = l1_node
else:
# 尝试作为 L3 — 找到其所属 L2
for l1 in tree.roots:
for l2 in l1.children:
for l3 in l2.children:
if l3.id == node_id:
target_l2_nodes = [l2]
parent_l1 = l1
break
if target_l2_nodes:
break
if target_l2_nodes:
break
if not target_l2_nodes:
resolved = _resolve_subtree(tree, node_id)
if resolved is None:
return False
# Phase 2: 统计 L3 节点数
total_l3 = sum(len(l2.children) for l2 in target_l2_nodes)
if total_l3 < constraint.min_l3_nodes:
if _count_l3_descendants(resolved.l2_nodes) < constraint.min_l3_nodes:
return False
if constraint.require_frames and not _has_frames(resolved.l2_nodes):
return False
if _count_subtitles(resolved.l2_nodes) < constraint.min_subtitles:
return False
return not (
constraint.cross_l2_span
and (resolved.parent_l1 is None or len(resolved.parent_l1.children) < 2)
)
# Phase 3: 检查帧路径可用性
if constraint.require_frames:
has_frame = any(l3.frame_path for l2 in target_l2_nodes for l3 in l2.children)
if not has_frame:
return False
# Phase 4: 统计字幕数
subtitle_count = 0
for l2 in target_l2_nodes:
def _subtitles_from_l2_list(l2_nodes: list[L2Node]) -> list[str]:
"""从 L2 节点列表收集全部非空字幕。
参数:
l2_nodes: L2 节点列表。
返回:
非空字幕字符串列表。
"""
sentences: list[str] = []
for l2 in l2_nodes:
if l2.card.subtitle:
subtitle_count += 1
sentences.append(l2.card.subtitle)
for l3 in l2.children:
if l3.card.subtitle:
subtitle_count += 1
if subtitle_count < constraint.min_subtitles:
return False
# Phase 5: 检查跨 L2 可用性
return not (constraint.cross_l2_span and (parent_l1 is None or len(parent_l1.children) < 2))
sentences.append(l3.card.subtitle)
return sentences
def _collect_subtitle_sentences(tree: TreeIndex, node_ids: tuple[str, ...]) -> list[str]:
@@ -220,40 +304,13 @@ def _collect_subtitle_sentences(tree: TreeIndex, node_ids: tuple[str, ...]) -> l
node_ids: 待收集字幕的节点 ID 元组。
返回:
非空字幕句列表（去除空白后非空的字幕）。
非空字幕句列表。
"""
sentences: list[str] = []
for nid in node_ids:
# 尝试作为 L2
result = _find_l2_node(tree, nid)
if result is not None:
l2_node, _ = result
if l2_node.card.subtitle:
sentences.append(l2_node.card.subtitle)
for l3 in l2_node.children:
if l3.card.subtitle:
sentences.append(l3.card.subtitle)
continue
# 尝试作为 L1
l1_node = _find_l1_node(tree, nid)
if l1_node is not None:
for l2 in l1_node.children:
if l2.card.subtitle:
sentences.append(l2.card.subtitle)
for l3 in l2.children:
if l3.card.subtitle:
sentences.append(l3.card.subtitle)
continue
# 尝试作为 L3
for l1 in tree.roots:
for l2 in l1.children:
for l3 in l2.children:
if l3.id == nid and l3.card.subtitle:
sentences.append(l3.card.subtitle)
resolved = _resolve_subtree(tree, nid)
if resolved is not None:
sentences.extend(_subtitles_from_l2_list(resolved.l2_nodes))
return sentences
@@ -369,6 +426,18 @@ def _sample_l1_node(
return rng.choice(candidates)
def _frames_from_l2_list(l2_nodes: list[L2Node]) -> list[str]:
"""从 L2 节点列表收集全部可用帧路径。
参数:
l2_nodes: L2 节点列表。
返回:
帧路径字符串列表。
"""
return [l3.frame_path for l2 in l2_nodes for l3 in l2.children if l3.frame_path]
def _collect_frame_paths(tree: TreeIndex, node_id: str) -> list[str]:
"""收集节点子树下的所有可用帧路径。
@@ -379,35 +448,10 @@ def _collect_frame_paths(tree: TreeIndex, node_id: str) -> list[str]:
返回:
帧路径列表。
"""
paths: list[str] = []
# L2 节点
result = _find_l2_node(tree, node_id)
if result is not None:
l2_node, _ = result
for l3 in l2_node.children:
if l3.frame_path:
paths.append(l3.frame_path)
return paths
# L1 节点
l1_node = _find_l1_node(tree, node_id)
if l1_node is not None:
for l2 in l1_node.children:
for l3 in l2.children:
if l3.frame_path:
paths.append(l3.frame_path)
return paths
# L3 节点
for l1 in tree.roots:
for l2 in l1.children:
for l3 in l2.children:
if l3.id == node_id and l3.frame_path:
paths.append(l3.frame_path)
return paths
return paths
resolved = _resolve_subtree(tree, node_id)
if resolved is None:
return []
return _frames_from_l2_list(resolved.l2_nodes)
def _collect_source_nodes(tree: TreeIndex, node_id: str) -> tuple[str, ...]: