diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index b5d055add..df2c5d248 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/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,25 @@ 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 +941,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) diff --git a/packages/domain/voice_duration_planner.py b/packages/domain/voice_duration_planner.py index 69054aa7d..4909c889a 100644 --- a/packages/domain/voice_duration_planner.py +++ b/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,72 @@ 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 +111,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 +163,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 diff --git a/tests/unit/test_rhythm_templates.py b/tests/unit/test_rhythm_templates.py new file mode 100644 index 000000000..efbb6b325 --- /dev/null +++ b/tests/unit/test_rhythm_templates.py @@ -0,0 +1,190 @@ +"""节奏模板单元测试(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}"