refactor: extract _assemble_pair to lower build_units complexity to B

将 pair 分组的数量/角色校验与组装下沉到 _assemble_pair 私有 helper,
build_units 主体只做分组+派发,radon 复杂度从 C(11) 降到 A(4)。
fail-fast 语义、公共函数签名、稳定顺序均不变。
This commit is contained in:
2026-07-15 05:54:08 -04:00
parent bef46636fe
commit 5ef5f2b8b7
+33 -17
View File
@@ -9,6 +9,35 @@ from collections import defaultdict
from core.types import GeneratedQuestion, QuestionUnit
def _assemble_pair(pair_id: str, group: list[GeneratedQuestion]) -> QuestionUnit:
"""校验单个 pair 分组的数量/角色并组装为 pair 单元(fail-fast)。
参数:
pair_id: 该分组共享的孪生对标识。
group: 归属同一 pair_id 的题目列表。
返回:
kind="pair" 的 QuestionUnit。
关键实现:
- 分组必须恰好 2 条,否则视为孤儿/超员,raise ValueError。
- 显式检查 pair_original / pair_mirror 角色齐备且唯一,缺失或重复
直接 raise ValueError(防 next(...) 静默 StopIteration)。
- 合法孪生对交由 QuestionUnit.from_pair 做 video_id/task_type/flip_axis
一致性断言。
"""
if len(group) != 2:
raise ValueError(f"pair {pair_id} 数量={len(group)}≠2(孤儿或超员)")
originals = [q for q in group if q.question_role == "pair_original"]
mirrors = [q for q in group if q.question_role == "pair_mirror"]
if len(originals) != 1 or len(mirrors) != 1:
raise ValueError(
f"pair {pair_id} 角色非法:original={len(originals)} mirror={len(mirrors)}"
"需各恰好 1 条"
)
return QuestionUnit.from_pair(originals[0], mirrors[0])
def build_units(questions: list[GeneratedQuestion]) -> list[QuestionUnit]:
"""将扁平题目列表聚合为单元列表:single 单封、pair 按 pair_id 成对聚合。
@@ -20,11 +49,9 @@ def build_units(questions: list[GeneratedQuestion]) -> list[QuestionUnit]:
pair 按首次出现的 pair_id 顺序)。
关键实现:
- pair_id 为空 → single 单元;非空 → 归入对应 pair 桶。
- 每个 pair 桶必须恰好 2 条,否则视为孤儿/超员,raise ValueError。
- 显式检查 pair_original / pair_mirror 角色齐备且唯一,缺失或重复
直接 raise ValueError(防 next(...) 静默 StopIteration),保持
fail-fast。合法孪生对交由 QuestionUnit.from_pair 做一致性断言。
pair_id 为空 → single 单元;非空 → 归入对应 pair 桶。各 pair 桶的
数量/角色校验与组装下沉到 _assemble_pairfail-fast),本体只做分组
与派发。
"""
by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
singles: list[QuestionUnit] = []
@@ -34,18 +61,7 @@ def build_units(questions: list[GeneratedQuestion]) -> list[QuestionUnit]:
else:
singles.append(QuestionUnit.from_single(q))
pairs: list[QuestionUnit] = []
for pid, qs in by_pair.items():
if len(qs) != 2:
raise ValueError(f"pair {pid} 数量={len(qs)}≠2(孤儿或超员)")
originals = [q for q in qs if q.question_role == "pair_original"]
mirrors = [q for q in qs if q.question_role == "pair_mirror"]
if len(originals) != 1 or len(mirrors) != 1:
raise ValueError(
f"pair {pid} 角色非法:original={len(originals)} mirror={len(mirrors)}"
"需各恰好 1 条"
)
pairs.append(QuestionUnit.from_pair(originals[0], mirrors[0]))
pairs = [_assemble_pair(pid, qs) for pid, qs in by_pair.items()]
return singles + pairs