From ee4636e087a7043ece9745dbd050cdbd83cc4f78 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 8 Sep 2026 10:28:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20#1768=20=E8=8A=82=E5=A5=8F=E6=9B=B2?= =?UTF-8?q?=E7=BA=BF=E6=A8=A1=E6=9D=BF=E5=A4=9A=E6=A0=B7=E5=8C=96=20?= =?UTF-8?q?=E2=80=94=208=E7=A7=8D=E9=A2=84=E8=AE=BE=20+=20=E6=97=B6?= =?UTF-8?q?=E9=95=BF=E9=92=B3=E5=88=B6=20+=20=E8=AF=AF=E5=B7=AE=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 变更： - 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 | 16 +- packages/domain/voice_duration_planner.py | 65 +++- tests/unit/domain/test_rhythm_templates_v2.py | 317 ++++++++++++++++++ tests/unit/test_rhythm_templates.py | 6 +- 4 files changed, 385 insertions(+), 19 deletions(-) create mode 100644 tests/unit/domain/test_rhythm_templates_v2.py diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index b66695bfa..1d66f366d 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/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), ) diff --git a/packages/domain/voice_duration_planner.py b/packages/domain/voice_duration_planner.py index 4909c889a..080d3a9ba 100644 --- a/packages/domain/voice_duration_planner.py +++ b/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 diff --git a/tests/unit/domain/test_rhythm_templates_v2.py b/tests/unit/domain/test_rhythm_templates_v2.py new file mode 100644 index 000000000..5715e07c1 --- /dev/null +++ b/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 diff --git a/tests/unit/test_rhythm_templates.py b/tests/unit/test_rhythm_templates.py index efbb6b325..9d5085667 100644 --- a/tests/unit/test_rhythm_templates.py +++ b/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）。"""