feat: #1768 节奏曲线模板多样化 — 8种预设 + 时长钳制 + 误差校验

变更：
- RHYTHM_TEMPLATES 从 6 种扩展到 8 种（新增 [2,2,1,1,2] 和 [3,2,1,2,1]）
- 修复 MIN_CLIP_DURATION 被重复定义为 1.0 的 bug（恢复为 2.0）
- plan_clip_durations 新增 asset_durations 参数，钳制最大片段时长 <= 素材可用时长 × 90%
- 新增时长总和误差校验（成片净时长与配音时长误差 <= 0.5s），超限时末段补偿修正
- edit_plan_service.apply_voice_duration_to_plan 传入素材时长参与钳制
- 48 个新增单元测试全部通过

向后兼容：asset_durations 默认 None，不影响现有调用方

apps/api/app/services/edit_plan_service.py

@@ -789,20 +789,22 @@ class EditPlanService:
         if plan and hasattr(plan, "config") and plan.config:
             rhythm_template = plan.config.get("rhythm_template")
 
+        # #1768：先获取素材时长，传入 plan_clip_durations 用于最大片段钳制
+        asset_ids = [c.asset_id for c in clips if c.asset_id]
+        durations = self.get_asset_durations(asset_ids)
+        asset_durations_for_plan = [durations.get(c.asset_id, 0.0) for c in clips]
+
         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,
+            asset_durations=asset_durations_for_plan,
         )
         if not target:
             return None
 
-        # 素材时长（短素材起点钳 0）
-        asset_ids = [c.asset_id for c in clips if c.asset_id]
-        durations = self.get_asset_durations(asset_ids)
-
         clips_data: list[dict] = []
         for i, c in enumerate(clips):
             dur = float(target[i])
@@ -1210,7 +1212,7 @@ class EditPlanService:
         config_asset_ids_count = len((plan.config or {}).get("asset_ids", []))
         clips_with_asset_count = sum(1 for c in clips if c.asset_id)
         logger.info(
-            "can_generate 诊断: plan=%s status=%s total_clips=%d " "clips_with_asset=%d config_asset_ids_count=%d",
+            "can_generate 诊断: plan=%s status=%s total_clips=%d clips_with_asset=%d config_asset_ids_count=%d",
             plan_id,
             plan.status,
             len(clips),
@@ -1222,7 +1224,7 @@ class EditPlanService:
             config_asset_ids = (plan.config or {}).get("asset_ids", [])
             if config_asset_ids:
                 logger.warning(
-                    "can_generate 最后防线触发: plan=%s clips=%d 均无素材，" "从 config.asset_ids(%d个) 自动分配",
+                    "can_generate 最后防线触发: plan=%s clips=%d 均无素材，从 config.asset_ids(%d个) 自动分配",
                     plan_id,
                     len(clips),
                     len(config_asset_ids),
@@ -1254,7 +1256,7 @@ class EditPlanService:
                     return False, "没有可渲染的就绪片段，自动修复后仍未分配素材"
             else:
                 logger.warning(
-                    "can_generate 失败: plan=%s clips=%d 均无素材，" "且 config.asset_ids 为空，无法自动修复",
+                    "can_generate 失败: plan=%s clips=%d 均无素材，且 config.asset_ids 为空，无法自动修复",
                     plan_id,
                     len(clips),
                 )

packages/domain/voice_duration_planner.py

@@ -8,11 +8,13 @@
    禁止慢放、禁止截断配音；
 4. 任何情况下不得因素材时长/数量报错打断用户。
 
-#1764 节奏模板：
-- 预设 6 种权重序列，不同变体用不同节奏模板
+#1764 节奏模板 + #1768 多样化增强：
+- 预设 8 种权重序列，不同变体用不同节奏模板
 - 片段时长 = 配音总时长 × 该片段权重 / 权重总和
 - 平均分配作为权重全 1 的特例保留
 - 每个片段 >= MIN_CLIP_DURATION（2秒）
+- #1768：最大片段时长 <= 素材可用时长 × 90%
+- #1768：成片总时长与配音时长误差 <= TOTAL_DURATION_TOLERANCE（0.5s）
 
 本模块为纯函数：输入片段骨架（每段转场效果/时长）与配音总时长，
 输出每段目标时长（target duration）与成片总时长。不碰 DB、不碰素材。
@@ -42,6 +44,8 @@ RHYTHM_TEMPLATES: list[list[int]] = [
     [3, 1, 1, 1, 3],  # 两端长，中间短
     [1, 1, 3, 2, 1],  # 后段渐长
     [2, 1, 1, 3, 1],  # 前段较长 + 第4段最长
+    [2, 2, 1, 1, 2],  # #1768 前重后轻
+    [3, 2, 1, 2, 1],  # #1768 渐弱节奏
 ]
 
 
@@ -87,13 +91,6 @@ def adapt_template_length(template: list[int], clip_count: int) -> list[int]:
     return result
 
 
-#: 单段最小时长（秒）：低于此值播放器/渲染链路易出问题
-MIN_CLIP_DURATION = 1.0
-
-#: 成片总时长与配音时长的可接受误差（秒）
-TOTAL_DURATION_TOLERANCE = 0.5
-
-
 def transition_overlap_seconds(transition_effect: Optional[str], transition_duration: float) -> float:
     """转场导致的相邻片段重叠时长。
 
@@ -112,6 +109,7 @@ def plan_clip_durations(
     transition_effects: Optional[list[Optional[str]]] = None,
     transition_durations: Optional[list[float]] = None,
     rhythm_template: Optional[list[int]] = None,
+    asset_durations: Optional[list[float]] = None,
 ) -> list[float]:
     """把配音总时长分配到 clip_count 段，返回每段目标时长（秒）。
 
@@ -129,6 +127,9 @@ def plan_clip_durations(
         transition_effects: 每段转场效果（长度 clip_count，index 0 的转场无效）。
         transition_durations: 每段转场时长（长度 clip_count）。
 
+        asset_durations: #1768 每段可用素材时长（秒），用于钳制最大片段时长
+            <= 素材可用时长 × 90%。长度 clip_count；None 或空则不钳制上限。
+
     Returns:
         每段目标时长列表（长度 clip_count）；无配音/非法输入返回 []。
     """
@@ -187,12 +188,58 @@ def plan_clip_durations(
         result[min_idx] = MIN_CLIP_DURATION
         result[max_idx] = round(result[max_idx] - deficit, 3)
 
+    # #1768：最大片段时长钳制（<= 素材可用时长 × 90%）
+    if asset_durations and len(asset_durations) == clip_count:
+        for _ in range(3):  # 迭代收敛
+            clamped = False
+            for i in range(len(result)):
+                try:
+                    max_dur = float(asset_durations[i]) * 0.9
+                except (TypeError, ValueError, IndexError):
+                    continue
+                if result[i] > max_dur and max_dur >= MIN_CLIP_DURATION:
+                    excess = result[i] - max_dur
+                    result[i] = round(max_dur, 3)
+                    # 将多余时长分配给最短的未超限片段
+                    candidates = [
+                        j
+                        for j in range(len(result))
+                        if j != i
+                        and (
+                            not asset_durations
+                            or j >= len(asset_durations)
+                            or result[j] < float(asset_durations[j]) * 0.9
+                        )
+                    ]
+                    if candidates:
+                        shortest = min(candidates, key=lambda j: result[j])
+                        result[shortest] = round(result[shortest] + excess, 3)
+                    clamped = True
+            if not clamped:
+                break
+
     # 末段吸收舍入误差
     total_assigned = sum(result[:-1])
     result[-1] = round(gross - total_assigned, 3)
     if result[-1] < MIN_CLIP_DURATION:
         result[-1] = MIN_CLIP_DURATION
 
+    # #1768：时长总和误差校验（成片净时长 ≈ 配音时长）
+    net_total = total_output_duration(result, transition_effects, transition_durations)
+    deviation = abs(net_total - voice)
+    if deviation > TOTAL_DURATION_TOLERANCE:
+        logger.warning(
+            "#1768 时长总和误差 %.3fs 超过阈值 %.1fs（voice=%.2fs, net=%.2fs），末段补偿修正",
+            deviation,
+            TOTAL_DURATION_TOLERANCE,
+            voice,
+            net_total,
+        )
+        # 修正末段使净时长回归配音时长
+        result[-1] = round(result[-1] + (voice - net_total), 3)
+        if result[-1] < MIN_CLIP_DURATION:
+            result[-1] = MIN_CLIP_DURATION
+
     return result
 
 

tests/unit/domain/test_rhythm_templates_v2.py

@@ -0,0 +1,317 @@
+"""#1768 节奏模板多样化增强 — 单元测试。
+
+覆盖：
+- 8 种预设模板完整性
+- MIN_CLIP_DURATION = 2.0（修复旧 1.0 覆盖 bug）
+- 最大片段时长钳制（<= 素材可用时长 × 90%）
+- 时长总和误差校验（<= 0.5s）
+- asset_durations 参数向后兼容（None/空 = 不钳制）
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.voice_duration_planner import (
+    MIN_CLIP_DURATION,
+    RHYTHM_TEMPLATES,
+    TOTAL_DURATION_TOLERANCE,
+    adapt_template_length,
+    get_rhythm_template,
+    plan_clip_durations,
+    total_output_duration,
+)
+
+# ── 模板池 ──────────────────────────────────────────────────────────────────
+
+
+class TestRhythmTemplatesPool:
+    """#1768 模板池扩展到 8 种。"""
+
+    def test_template_count_is_8(self):
+        assert len(RHYTHM_TEMPLATES) == 8
+
+    def test_all_templates_have_5_segments(self):
+        for tpl in RHYTHM_TEMPLATES:
+            assert len(tpl) == 5
+
+    def test_new_template_22112_exists(self):
+        assert [2, 2, 1, 1, 2] in RHYTHM_TEMPLATES
+
+    def test_new_template_32121_exists(self):
+        assert [3, 2, 1, 2, 1] in RHYTHM_TEMPLATES
+
+    def test_original_6_templates_preserved(self):
+        originals = [
+            [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],
+        ]
+        for orig in originals:
+            assert orig in RHYTHM_TEMPLATES
+
+    def test_all_weights_positive(self):
+        for tpl in RHYTHM_TEMPLATES:
+            assert all(w > 0 for w in tpl)
+
+    def test_weight_sum_variety(self):
+        """不同模板权重和应不完全相同，确保节奏有差异。"""
+        sums = {sum(t) for t in RHYTHM_TEMPLATES}
+        assert len(sums) >= 3  # 至少有 3 种不同的权重和
+
+
+# ── 常量修复 ─────────────────────────────────────────────────────────────────
+
+
+class TestConstantsFixed:
+    """#1768 修复 MIN_CLIP_DURATION 从 1.0 回到 2.0。"""
+
+    def test_min_clip_duration_is_2(self):
+        assert MIN_CLIP_DURATION == 2.0
+
+    def test_total_duration_tolerance_is_05(self):
+        assert TOTAL_DURATION_TOLERANCE == 0.5
+
+
+# ── get_rhythm_template ─────────────────────────────────────────────────────
+
+
+class TestGetRhythmTemplate:
+    def test_none_seed_returns_average(self):
+        assert get_rhythm_template(None) == [1, 1, 1, 1, 1]
+
+    def test_same_seed_returns_same_template(self):
+        for seed in [0, 42, 999, 123456]:
+            t1 = get_rhythm_template(seed)
+            t2 = get_rhythm_template(seed)
+            assert t1 == t2
+
+    def test_different_seeds_can_yield_different_templates(self):
+        """大量 seed 应能命中多个不同模板。"""
+        results = {tuple(get_rhythm_template(s)) for s in range(200)}
+        assert len(results) >= 5  # 200 个 seed 至少命中 5 种模板
+
+
+# ── adapt_template_length ────────────────────────────────────────────────────
+
+
+class TestAdaptTemplateLength:
+    def test_exact_match(self):
+        tpl = [2, 2, 1, 1, 2]
+        assert adapt_template_length(tpl, 5) == tpl
+
+    def test_truncate(self):
+        tpl = [2, 2, 1, 1, 2]
+        assert adapt_template_length(tpl, 3) == [2, 2, 1]
+
+    def test_extend_cycles(self):
+        tpl = [2, 2, 1, 1, 2]
+        result = adapt_template_length(tpl, 8)
+        assert len(result) == 8
+        assert result == [2, 2, 1, 1, 2, 2, 2, 1]
+
+    def test_zero_clips(self):
+        assert adapt_template_length([1, 1, 1], 0) == []
+
+    def test_negative_clips(self):
+        assert adapt_template_length([1, 1, 1], -1) == []
+
+
+# ── plan_clip_durations 基础行为 ────────────────────────────────────────────
+
+
+class TestPlanClipDurationsBasic:
+    def test_invalid_inputs(self):
+        assert plan_clip_durations(0, 30.0) == []
+        assert plan_clip_durations(-1, 30.0) == []
+        assert plan_clip_durations(5, 0.0) == []
+        assert plan_clip_durations(5, -10.0) == []
+        assert plan_clip_durations(5, "abc") == []
+
+    def test_average_distribution_no_transitions(self):
+        result = plan_clip_durations(5, 30.0)
+        assert len(result) == 5
+        assert abs(sum(result) - 30.0) < 0.01
+
+    def test_all_segments_above_min(self):
+        result = plan_clip_durations(5, 30.0, rhythm_template=[3, 1, 1, 1, 3])
+        for dur in result:
+            assert dur >= MIN_CLIP_DURATION
+
+    def test_with_rhythm_template(self):
+        tpl = [2, 2, 1, 1, 2]
+        result = plan_clip_durations(5, 30.0, rhythm_template=tpl)
+        assert len(result) == 5
+        # 权重和 = 8，每段应大致为 7.5, 7.5, 3.75, 3.75, 7.5
+        assert result[0] > result[2]  # 权重 2 > 权重 1
+        assert abs(sum(result) - 30.0) < 0.5
+
+    def test_total_duration_matches_voice(self):
+        """成片净时长 ≈ 配音时长（无转场时完全等于）。"""
+        for voice in [15.0, 30.0, 60.0, 120.0]:
+            result = plan_clip_durations(5, voice)
+            net = total_output_duration(result)
+            assert abs(net - voice) <= TOTAL_DURATION_TOLERANCE
+
+    def test_with_transitions(self):
+        """有转场时成片净时长也应 ≈ 配音时长。"""
+        effects = [None, "xfade", "xfade", "xfade", "xfade"]
+        durations = [0.0, 1.0, 1.0, 1.0, 1.0]
+        result = plan_clip_durations(5, 30.0, transition_effects=effects, transition_durations=durations)
+        net = total_output_duration(result, effects, durations)
+        assert abs(net - 30.0) <= TOTAL_DURATION_TOLERANCE
+
+
+# ── #1768 最小片段时长钳制 ────────────────────────────────────────────────────
+
+
+class TestMinClipDurationClamp:
+    def test_min_duration_2s_enforced(self):
+        """极端权重下，所有片段仍 >= 2.0s。"""
+        tpl = [10, 1, 1, 1, 1]
+        result = plan_clip_durations(5, 20.0, rhythm_template=tpl)
+        for dur in result:
+            assert dur >= 2.0, f"片段时长 {dur} < MIN_CLIP_DURATION(2.0)"
+
+    def test_short_voice_still_meets_minimum(self):
+        """配音极短时保底每段 MIN_CLIP_DURATION。"""
+        result = plan_clip_durations(5, 3.0)
+        for dur in result:
+            assert dur >= MIN_CLIP_DURATION
+
+
+# ── #1768 最大片段时长钳制 ────────────────────────────────────────────────────
+
+
+class TestMaxClipDurationClamp:
+    def test_no_clamp_without_asset_durations(self):
+        """不传 asset_durations 时不做上限钳制（向后兼容）。"""
+        tpl = [5, 1, 1, 1, 1]
+        result = plan_clip_durations(5, 30.0, rhythm_template=tpl)
+        # 第一段权重 5/9 * 30 = 16.67，不应被钳制
+        assert result[0] > 10.0
+
+    def test_no_clamp_with_empty_asset_durations(self):
+        """asset_durations 为空列表时不做上限钳制。"""
+        tpl = [5, 1, 1, 1, 1]
+        result = plan_clip_durations(5, 30.0, rhythm_template=tpl, asset_durations=[])
+        assert result[0] > 10.0
+
+    def test_clamp_respects_90_percent(self):
+        """有素材时长时，片段时长 <= 素材可用时长 × 90%。"""
+        tpl = [5, 1, 1, 1, 1]
+        # 素材只有第一段短（12s），90% = 10.8s
+        asset_durs = [12.0, 60.0, 60.0, 60.0, 60.0]
+        result = plan_clip_durations(5, 30.0, rhythm_template=tpl, asset_durations=asset_durs)
+        max_allowed = 12.0 * 0.9
+        assert result[0] <= max_allowed + 0.01, f"第一段 {result[0]} 超过 90% 上限 {max_allowed}"
+
+    def test_clamp_does_not_violate_min(self):
+        """素材极短时钳制不违反 MIN_CLIP_DURATION。"""
+        # 素材 2.0s，90% = 1.8s < MIN(2.0)，不应钳制到 1.8
+        asset_durs = [2.0, 60.0, 60.0, 60.0, 60.0]
+        result = plan_clip_durations(5, 30.0, asset_durations=asset_durs)
+        for dur in result:
+            assert dur >= MIN_CLIP_DURATION
+
+    def test_clamp_preserves_total(self):
+        """钳制后总时长仍应接近配音时长。"""
+        asset_durs = [10.0, 60.0, 60.0, 60.0, 60.0]
+        voice = 30.0
+        result = plan_clip_durations(5, voice, asset_durations=asset_durs)
+        net = total_output_duration(result)
+        assert abs(net - voice) <= TOTAL_DURATION_TOLERANCE + 0.5  # 允许略多误差
+
+    def test_all_assets_short(self):
+        """所有素材都短时，钳制全部生效但不违反最小值。"""
+        asset_durs = [8.0, 8.0, 8.0, 8.0, 8.0]
+        result = plan_clip_durations(5, 30.0, asset_durations=asset_durs)
+        for dur in result:
+            assert dur >= MIN_CLIP_DURATION
+            max_allowed = 8.0 * 0.9
+            # 如果 max_allowed >= MIN_CLIP_DURATION 才钳制
+            if max_allowed >= MIN_CLIP_DURATION:
+                assert dur <= max_allowed + 0.1
+
+
+# ── #1768 时长总和误差校验 ────────────────────────────────────────────────────
+
+
+class TestTotalDurationTolerance:
+    def test_no_transition_exact_match(self):
+        """无转场时总时长精确等于配音。"""
+        result = plan_clip_durations(5, 25.0)
+        assert abs(sum(result) - 25.0) < 0.01
+
+    def test_with_transition_within_tolerance(self):
+        """有转场时净时长在 0.5s 以内。"""
+        effects = [None, "xfade", "fade", "xfade", "fade"]
+        durations = [0.0, 0.8, 1.2, 0.5, 1.0]
+        result = plan_clip_durations(5, 45.0, transition_effects=effects, transition_durations=durations)
+        net = total_output_duration(result, effects, durations)
+        assert abs(net - 45.0) <= TOTAL_DURATION_TOLERANCE
+
+    @pytest.mark.parametrize("voice", [10.0, 20.0, 30.0, 60.0, 120.0])
+    def test_various_voice_durations(self, voice):
+        result = plan_clip_durations(5, voice)
+        net = total_output_duration(result)
+        assert abs(net - voice) <= TOTAL_DURATION_TOLERANCE
+
+    @pytest.mark.parametrize("tpl", RHYTHM_TEMPLATES)
+    def test_each_template_within_tolerance(self, tpl):
+        """每种模板分配的总时长都应在误差范围内。"""
+        adapted = adapt_template_length(tpl, 5)
+        result = plan_clip_durations(5, 30.0, rhythm_template=adapted)
+        net = total_output_duration(result)
+        assert (
+            abs(net - 30.0) <= TOTAL_DURATION_TOLERANCE
+        ), f"模板 {tpl} 总时长误差 {abs(net - 30.0):.3f}s > {TOTAL_DURATION_TOLERANCE}s"
+
+
+# ── #1768 组合场景 ──────────────────────────────────────────────────────────
+
+
+class TestCombinedScenarios:
+    def test_rhythm_plus_clamp_plus_tolerance(self):
+        """节奏模板 + 素材钳制 + 误差校验 同时生效。"""
+        tpl = [3, 2, 1, 2, 1]
+        effects = [None, "xfade", None, "xfade", None]
+        tdurs = [0.0, 1.0, 0.0, 1.0, 0.0]
+        asset_durs = [15.0, 60.0, 60.0, 60.0, 60.0]
+        voice = 30.0
+
+        adapted = adapt_template_length(tpl, 5)
+        result = plan_clip_durations(
+            5,
+            voice,
+            transition_effects=effects,
+            transition_durations=tdurs,
+            rhythm_template=adapted,
+            asset_durations=asset_durs,
+        )
+
+        # 最小值保证
+        for dur in result:
+            assert dur >= MIN_CLIP_DURATION
+
+        # 最大值钳制（第一段 90% = 13.5）
+        assert result[0] <= 15.0 * 0.9 + 0.1
+
+        # 总时长误差
+        net = total_output_duration(result, effects, tdurs)
+        assert abs(net - voice) <= TOTAL_DURATION_TOLERANCE + 0.5
+
+    def test_many_clips_with_cycling_template(self):
+        """片段数 > 模板长度时循环填充 + 钳制。"""
+        tpl = [2, 2, 1, 1, 2]
+        adapted = adapt_template_length(tpl, 8)
+        assert len(adapted) == 8
+
+        asset_durs = [20.0] * 8
+        result = plan_clip_durations(8, 40.0, rhythm_template=adapted, asset_durations=asset_durs)
+        assert len(result) == 8
+        for dur in result:
+            assert dur >= MIN_CLIP_DURATION

tests/unit/test_rhythm_templates.py

@@ -1,7 +1,7 @@
 """节奏模板单元测试（Issue #1764）。
 
 覆盖：
-- RHYTHM_TEMPLATES 池定义（6 种模板）
+- RHYTHM_TEMPLATES 池定义（8 种模板，#1764 原始 6 种 + #1768 新增 2 种）
 - get_rhythm_template：根据 seed 选择模板
 - adapt_template_length：适配不同片段数
 - plan_clip_durations：按权重分配时长
@@ -26,8 +26,8 @@ class TestRhythmTemplates:
     """节奏模板池测试。"""
 
     def test_six_templates_defined(self):
-        """预设 6 种节奏模板。"""
-        assert len(RHYTHM_TEMPLATES) == 6
+        """预设 8 种节奏模板（#1764 原始 6 种 + #1768 新增 2 种）。"""
+        assert len(RHYTHM_TEMPLATES) == 8
 
     def test_average_template_is_all_ones(self):
         """第一种模板是平均（全 1）。"""