feat: 节奏模板让批量变体片段时长分布不同 (Issue #1764)

- 新增 RHYTHM_TEMPLATES 节奏模板池（6 种权重序列）
- get_rhythm_template(seed)：根据变体 seed 随机选模板
- adapt_template_length()：将模板适配到实际片段数
- plan_clip_durations() 支持 rhythm_template 参数按权重分配时长
- 保底约束：每段 >= 2s，总时长 = 配音时长
- edit_plan_service: ensure_batch_variant_plans 为每个变体生成不同节奏模板
- 17 个新测试覆盖模板池/适配/分配/验收

apps/api/app/services/edit_plan_service.py

@@ -784,11 +784,17 @@ class EditPlanService:
 
         from packages.domain.voice_duration_planner import plan_clip_durations, total_output_duration
 
+        # #1764：从 plan config 读取节奏模板
+        rhythm_template = None
+        if plan and hasattr(plan, "config") and plan.config:
+            rhythm_template = plan.config.get("rhythm_template")
+
         target = plan_clip_durations(
             len(clips),
             voice,
             transition_effects=[c.transition_effect for c in clips],
             transition_durations=[float(c.transition_duration or 0.0) for c in clips],
+            rhythm_template=rhythm_template,
         )
         if not target:
             return None
@@ -907,6 +913,28 @@ class EditPlanService:
             )
             plan_ids.append(variant.id)
 
+        # #1764：为每个变体生成独立节奏模板（让批量视频片段时长分布不同）
+        from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length
+
+        clip_count = 0
+        if voice_durations and len(voice_durations) > 0:
+            # 从源 plan 获取片段数
+            source_plan = self.get_plan(source_plan_id)
+            if source_plan and hasattr(source_plan, "clips"):
+                clip_count = len(list(source_plan.clips)) if source_plan.clips else 0
+
+        rhythm_templates_for_variants = []
+        if clip_count > 0:
+            for idx in range(len(plan_ids)):
+                # 每个变体用不同的 seed 选择节奏模板
+                variant_seed = rng.randint(0, 999999)
+                template = adapt_template_length(
+                    RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)],
+                    clip_count
+                )
+                rhythm_templates_for_variants.append(template)
+                logger.info("变体 %d 节奏模板: plan=%s template=%s", idx, plan_ids[idx], template)
+
         # 为每个变体生成独立视觉扰动参数（让批量视频画面本身更不同）
         from packages.domain.variant_plan_selector import generate_visual_perturbation
 
@@ -916,7 +944,11 @@ class EditPlanService:
                 # 变体 0 不做 hflip（保持预览 plan 原始画面方向）
                 if idx == 0:
                     perturbation["hflip"] = False
-                self.update_plan_config(pid, {"visual_perturbation": perturbation})
+                config_update = {"visual_perturbation": perturbation}
+                # #1764：写入节奏模板
+                if idx < len(rhythm_templates_for_variants):
+                    config_update["rhythm_template"] = rhythm_templates_for_variants[idx]
+                self.update_plan_config(pid, config_update)
                 logger.info("变体 %d 视觉扰动: plan=%s perturbation=%s", idx, pid, perturbation)
             except Exception:
                 logger.exception("变体 %d 视觉扰动生成失败（不阻断）: plan=%s", idx, pid)

packages/domain/voice_duration_planner.py

@@ -1,4 +1,4 @@
-"""配音时长 → 片段时长分配纯函数（#1749）。
+"""配音时长 → 片段时长分配纯函数（#1749 + #1764 节奏模板）。
 
 定稿规则（工单 #1749）：
 1. 片段数 = 模板片段数，定死，不因素材增减；
@@ -8,6 +8,12 @@
    禁止慢放、禁止截断配音；
 4. 任何情况下不得因素材时长/数量报错打断用户。
 
+#1764 节奏模板：
+- 预设 6 种权重序列，不同变体用不同节奏模板
+- 片段时长 = 配音总时长 × 该片段权重 / 权重总和
+- 平均分配作为权重全 1 的特例保留
+- 每个片段 >= MIN_CLIP_DURATION（2秒）
+
 本模块为纯函数：输入片段骨架（每段转场效果/时长）与配音总时长，
 输出每段目标时长（target duration）与成片总时长。不碰 DB、不碰素材。
 """
@@ -15,10 +21,71 @@
 from __future__ import annotations
 
 import logging
+import random
 from typing import Optional
 
 logger = logging.getLogger(__name__)
 
+#: 单段最小时长（秒）：低于此值播放器/渲染链路易出问题
+MIN_CLIP_DURATION = 2.0
+
+#: 成片总时长与配音时长的可接受误差（秒）
+TOTAL_DURATION_TOLERANCE = 0.5
+
+# ── #1764 节奏模板池 ──────────────────────────────────────────────────────
+# 每种模板是权重序列，权重值代表相对时长比例
+# 变体基于 variant_seed 随机选一个模板，实现不同变体时长结构不同
+RHYTHM_TEMPLATES: list[list[int]] = [
+    [1, 1, 1, 1, 1],       # 平均（基准）
+    [2, 1, 3, 1, 2],       # 中间长，两端短
+    [1, 2, 1, 2, 1],       # 偶数段长
+    [3, 1, 1, 1, 3],       # 两端长，中间短
+    [1, 1, 3, 2, 1],       # 后段渐长
+    [2, 1, 1, 3, 1],       # 前段较长 + 第4段最长
+]
+
+
+def get_rhythm_template(variant_seed: int | None = None) -> list[int]:
+    """根据 variant_seed 选择一个节奏模板。
+
+    Args:
+        variant_seed: 变体随机种子；None 时返回平均模板
+
+    Returns:
+        权重序列（list[int]）
+    """
+    if variant_seed is None:
+        return RHYTHM_TEMPLATES[0]  # 默认平均
+    rng = random.Random(variant_seed)
+    return rng.choice(RHYTHM_TEMPLATES)
+
+
+def adapt_template_length(template: list[int], clip_count: int) -> list[int]:
+    """将节奏模板适配到实际片段数。
+
+    片段数 != 模板长度时：
+    - clip_count < len(template): 截断
+    - clip_count > len(template): 循环填充
+
+    Args:
+        template: 原始权重序列
+        clip_count: 实际片段数
+
+    Returns:
+        适配后的权重序列（长度 == clip_count）
+    """
+    if clip_count <= 0:
+        return []
+    if clip_count == len(template):
+        return template[:]
+    if clip_count < len(template):
+        return template[:clip_count]
+    # clip_count > len(template): 循环填充
+    result = []
+    for i in range(clip_count):
+        result.append(template[i % len(template)])
+    return result
+
 #: 单段最小时长（秒）：低于此值播放器/渲染链路易出问题
 MIN_CLIP_DURATION = 1.0
 
@@ -43,9 +110,12 @@ def plan_clip_durations(
     voice_duration: float,
     transition_effects: Optional[list[Optional[str]]] = None,
     transition_durations: Optional[list[float]] = None,
+    rhythm_template: Optional[list[int]] = None,
 ) -> list[float]:
     """把配音总时长分配到 clip_count 段，返回每段目标时长（秒）。
 
+    #1764：支持节奏模板，按权重比例分配时长；无模板时平均分配（向后兼容）。
+
     分配口径：Σ段长 − Σ转场重叠 = 配音时长（成片净时长 = 配音）。
     转场重叠发生在相邻片段之间，共 clip_count-1 处；第 i 处重叠取
     **后一段（i+1）** 的转场设置（与 xfade 构建口径一致：转场挂在后段）。
@@ -92,13 +162,36 @@ def plan_clip_durations(
             MIN_CLIP_DURATION,
         )
 
-    per_clip = gross / clip_count
-    result = [round(per_clip, 3) for _ in range(clip_count)]
-    # 末段吸收舍入误差：直接用 gross - 前段之和
-    result[-1] = round(gross - sum(result[:-1]), 3)
+    # #1764：按节奏模板权重分配（无模板时全 1 = 平均分配）
+    weights = rhythm_template if rhythm_template and len(rhythm_template) == clip_count else [1] * clip_count
+    
+    # 确保每个片段 >= MIN_CLIP_DURATION
+    # 先按权重分配，再检查最小值
+    total_weight = sum(weights)
+    raw_durations = [(w / total_weight) * gross for w in weights]
+    
+    # 保底检查：如果有片段 < MIN_CLIP_DURATION，提升它并从最长片段扣
+    result = [round(d, 3) for d in raw_durations]
+    for _ in range(3):  # 最多迭代 3 次
+        min_idx = min(range(len(result)), key=lambda i: result[i])
+        if result[min_idx] >= MIN_CLIP_DURATION:
+            break
+        # 从最长片段借时长
+        max_idx = max(range(len(result)), key=lambda i: result[i])
+        if max_idx == min_idx or result[max_idx] <= MIN_CLIP_DURATION:
+            # 无法再调整，强制保底
+            result[min_idx] = MIN_CLIP_DURATION
+            break
+        deficit = MIN_CLIP_DURATION - result[min_idx]
+        result[min_idx] = MIN_CLIP_DURATION
+        result[max_idx] = round(result[max_idx] - deficit, 3)
+    
+    # 末段吸收舍入误差
+    total_assigned = sum(result[:-1])
+    result[-1] = round(gross - total_assigned, 3)
     if result[-1] < MIN_CLIP_DURATION:
-        # 极端情况下末段被舍入压得过小，摊平
         result[-1] = MIN_CLIP_DURATION
+    
     return result
 
 

tests/unit/test_rhythm_templates.py

@@ -0,0 +1,189 @@
+"""节奏模板单元测试（Issue #1764）。
+
+覆盖：
+- RHYTHM_TEMPLATES 池定义（6 种模板）
+- get_rhythm_template：根据 seed 选择模板
+- adapt_template_length：适配不同片段数
+- plan_clip_durations：按权重分配时长
+- 时长约束：总时长 ≈ 配音时长，每段 >= 2s
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.voice_duration_planner import (
+    MIN_CLIP_DURATION,
+    RHYTHM_TEMPLATES,
+    adapt_template_length,
+    get_rhythm_template,
+    plan_clip_durations,
+    total_output_duration,
+)
+
+
+class TestRhythmTemplates:
+    """节奏模板池测试。"""
+
+    def test_six_templates_defined(self):
+        """预设 6 种节奏模板。"""
+        assert len(RHYTHM_TEMPLATES) == 6
+
+    def test_average_template_is_all_ones(self):
+        """第一种模板是平均（全 1）。"""
+        assert RHYTHM_TEMPLATES[0] == [1, 1, 1, 1, 1]
+
+    def test_all_templates_have_5_elements(self):
+        """所有模板长度为 5（会被 adapt 适配）。"""
+        for tpl in RHYTHM_TEMPLATES:
+            assert len(tpl) == 5
+
+
+class TestGetRhythmTemplate:
+    """get_rhythm_template 测试。"""
+
+    def test_none_seed_returns_average(self):
+        """None seed 返回平均模板。"""
+        assert get_rhythm_template(None) == [1, 1, 1, 1, 1]
+
+    def test_same_seed_same_template(self):
+        """相同 seed 返回相同模板。"""
+        tpl1 = get_rhythm_template(42)
+        tpl2 = get_rhythm_template(42)
+        assert tpl1 == tpl2
+
+    def test_different_seeds_may_differ(self):
+        """不同 seed 可能返回不同模板。"""
+        templates_seen = set()
+        for seed in range(100):
+            tpl = tuple(get_rhythm_template(seed))
+            templates_seen.add(tpl)
+        # 100 个 seed 应该至少看到 3 种不同模板
+        assert len(templates_seen) >= 3
+
+
+class TestAdaptTemplateLength:
+    """adapt_template_length 测试。"""
+
+    def test_same_length(self):
+        """片段数 == 模板长度时直接返回。"""
+        tpl = [2, 1, 3, 1, 2]
+        assert adapt_template_length(tpl, 5) == [2, 1, 3, 1, 2]
+
+    def test_shorter_clip_count(self):
+        """片段数 < 模板长度时截断。"""
+        tpl = [2, 1, 3, 1, 2]
+        assert adapt_template_length(tpl, 3) == [2, 1, 3]
+
+    def test_longer_clip_count(self):
+        """片段数 > 模板长度时循环填充。"""
+        tpl = [2, 1, 3]
+        result = adapt_template_length(tpl, 7)
+        assert result == [2, 1, 3, 2, 1, 3, 2]
+
+    def test_zero_clip_count(self):
+        """片段数 0 返回空列表。"""
+        assert adapt_template_length([1, 2, 3], 0) == []
+
+
+class TestPlanClipDurationsWithRhythm:
+    """plan_clip_durations 节奏模板测试。"""
+
+    def test_average_template_equals_old_behavior(self):
+        """全 1 模板 = 原来的平均分配。"""
+        voice = 20.0
+        clips = 4
+        result = plan_clip_durations(clips, voice, rhythm_template=[1, 1, 1, 1])
+        # 每段应该 ≈ 5s
+        assert all(abs(d - 5.0) < 0.1 for d in result)
+        assert abs(sum(result) - voice) < 0.1
+
+    def test_weighted_template_different_durations(self):
+        """权重模板产生不同时长的片段。"""
+        voice = 18.0
+        clips = 5
+        # 权重 [2, 1, 3, 1, 2]：第 3 段最长，第 2/4 段最短
+        template = [2, 1, 3, 1, 2]
+        result = plan_clip_durations(clips, voice, rhythm_template=template)
+        
+        # 总时长 ≈ 配音时长
+        assert abs(sum(result) - voice) < 0.5
+        
+        # 第 3 段应该最长
+        assert result[2] > result[1]
+        assert result[2] > result[3]
+
+    def test_min_clip_duration_enforced(self):
+        """每段 >= MIN_CLIP_DURATION (2s)。"""
+        voice = 15.0
+        clips = 5
+        # 极端权重：某段权重极低
+        template = [10, 1, 1, 1, 1]
+        result = plan_clip_durations(clips, voice, rhythm_template=template)
+        
+        for d in result:
+            assert d >= MIN_CLIP_DURATION
+
+    def test_total_duration_with_transitions(self):
+        """含转场时总时长仍然正确。"""
+        voice = 20.0
+        clips = 4
+        effects = [None, "xfade", "fade", "cut"]
+        durations = [0.0, 0.5, 0.3, 0.0]
+        template = [2, 1, 1, 2]
+        
+        result = plan_clip_durations(
+            clips, voice,
+            transition_effects=effects,
+            transition_durations=durations,
+            rhythm_template=template,
+        )
+        
+        # 成片净时长 = Σ段长 - Σ转场重叠 ≈ 配音时长
+        output = total_output_duration(result, effects, durations)
+        assert abs(output - voice) < 0.5
+
+    def test_no_template_backward_compatible(self):
+        """不传模板时行为与旧版一致（平均分配）。"""
+        voice = 16.0
+        clips = 4
+        result = plan_clip_durations(clips, voice)
+        assert all(abs(d - 4.0) < 0.1 for d in result)
+
+    def test_six_templates_produce_different_structures(self):
+        """6 种模板产生不同的时长结构。"""
+        voice = 25.0
+        clips = 5
+        structures = set()
+        
+        for tpl in RHYTHM_TEMPLATES:
+            result = plan_clip_durations(clips, voice, rhythm_template=tpl)
+            # 用 round 后的元组作为结构指纹
+            structure = tuple(round(d, 1) for d in result)
+            structures.add(structure)
+        
+        # 至少 4 种不同结构
+        assert len(structures) >= 4
+
+
+class TestIssue1764Acceptance:
+    """Issue #1764 验收测试。"""
+
+    def test_batch_3_variants_at_least_2_different(self):
+        """批量 3 个变体，至少 2 组不同片段时长序列。"""
+        voice = 20.0
+        clips = 5
+        
+        # 模拟 3 个变体用不同 seed
+        seeds = [100, 200, 300]
+        structures = []
+        
+        for seed in seeds:
+            template = get_rhythm_template(seed)
+            adapted = adapt_template_length(template, clips)
+            durations = plan_clip_durations(clips, voice, rhythm_template=adapted)
+            structures.append(tuple(round(d, 1) for d in durations))
+        
+        # 至少 2 种不同结构
+        unique = len(set(structures))
+        assert unique >= 2, f"Expected >= 2 unique structures, got {unique}: {structures}"