diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index eb5d32790..ac38928fe 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -59,13 +59,18 @@ def build_xfade_filter_chain( transitions: list[str], *, transition_duration: float = DEFAULT_TRANSITION_DURATION, + transition_durations: list[float] | None = None, + jitters: list[float] | None = None, output_label: str = "outv", ) -> tuple[str, float]: + """构建 xfade 转场滤镜链（re-export，#1766 增加逐转场时长与 jitter 支持）.""" return _build_xfade_filter_chain_base( clip_durations, clip_video_labels, transitions, transition_duration=transition_duration, + transition_durations=transition_durations, + jitters=jitters, output_label=output_label, ) diff --git a/apps/worker/video_processing/transition_engine.py b/apps/worker/video_processing/transition_engine.py index 0f757714a..e953f89d4 100755 --- a/apps/worker/video_processing/transition_engine.py +++ b/apps/worker/video_processing/transition_engine.py @@ -105,17 +105,24 @@ class TransitionEngine: transitions: list[str], *, transition_duration: float | None = None, + transition_durations: list[float] | None = None, + jitters: list[float] | None = None, output_label: str = "outv", ) -> tuple[str, float]: """构建 xfade 转场滤镜链. 对每步转场应用验证和降级，然后调用底层 ffmpeg_utils 构建。 + #1766 增强：支持逐转场独立时长（transition_durations）和位置微调（jitters）。 + Args: clip_durations: 每个片段的时长 clip_video_labels: 每个片段的视频流标签 transitions: 每个片段对应的转场效果 - transition_duration: 统一转场时长，None 则使用引擎默认值 + transition_duration: 全局默认转场时长，None 则使用引擎默认值 + transition_durations: #1766 逐转场时长列表，与 transitions 等长； + None 时使用各 resolved config 的 duration + jitters: #1766 逐转场位置偏移列表（秒） output_label: 最终输出标签 Returns: @@ -127,6 +134,8 @@ class TransitionEngine: clip_video_labels=clip_video_labels, transitions=transitions, transition_duration=transition_duration or self._default_duration, + transition_durations=transition_durations, + jitters=jitters, output_label=output_label, ) @@ -134,17 +143,20 @@ class TransitionEngine: resolved = self.resolve_clip_transitions(transitions, clip_durations) resolved_effects = [c.effect for c in resolved] - # 使用统一的时长（取各转场中最大的时长作为基准，底层会做每步钳制） - dur = transition_duration or self._default_duration - if not dur: - dur = max(c.duration for c in resolved) if resolved else DEFAULT_TRANSITION_DURATION + # #1766: 逐转场时长（优先使用传入的 transition_durations，否则用 resolved config） + if transition_durations is not None: + resolved_durations = list(transition_durations) + else: + resolved_durations = [c.duration for c in resolved] # 调用底层构建 return build_xfade_filter_chain( clip_durations=clip_durations, clip_video_labels=clip_video_labels, transitions=resolved_effects, - transition_duration=dur, + transition_duration=transition_duration or self._default_duration, + transition_durations=resolved_durations, + jitters=jitters, output_label=output_label, ) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 1f4360b2d..86ea5027c 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -1870,16 +1870,27 @@ class UnifiedRenderService: ) else: # 有转场效果：用 TransitionEngine 构建 xfade 链 - layer_dur = 0.0 - for d in layer_transition_durations: - if d > 0: - layer_dur = d - break + # #1766: 提取逐转场时长（每个 clip 的 transition_duration，跳过第一个） + # layer_transition_durations[i] 对应 clip i 的转场，第 0 个忽略 + per_transition_durations = [d for idx, d in enumerate(layer_transition_durations) if idx > 0] + # #1766: 提取每个 clip 的 jitter（存在 config 中），跳过第一个 + layer_jitters = [ + ( + all_clips[layer_clip_indices[idx]].config.get("transition_jitter", 0.0) + if isinstance(all_clips[layer_clip_indices[idx]].config, dict) + else 0.0 + ) + for idx in range(len(layer_clip_indices)) + ] + per_transition_jitters = [j for idx, j in enumerate(layer_jitters) if idx > 0] xfade_filter, xfade_estimated_dur = self._transition_engine.build_xfade_chain( clip_durations=layer_durations, clip_video_labels=layer_labels, transitions=layer_transitions, - transition_duration=layer_dur if layer_dur > 0 else None, + transition_durations=( + per_transition_durations if any(d > 0 for d in per_transition_durations) else None + ), + jitters=per_transition_jitters if any(j != 0.0 for j in per_transition_jitters) else None, output_label=out_label, ) if xfade_filter: diff --git a/packages/domain/transition_randomizer.py b/packages/domain/transition_randomizer.py new file mode 100644 index 000000000..2494fb336 --- /dev/null +++ b/packages/domain/transition_randomizer.py @@ -0,0 +1,177 @@ +"""转场位置与类型随机化（Issue #1766）. + +为不同变体生成不同的转场序列（类型 + 时长 + 位置微调）， +打破"所有变体转场节奏完全一致"的结构相似性， +降低平台查重识别为结构相似视频的风险。 + +设计要点： +1. 转场类型池：5 种效果（dissolve / zoom / slideleft / wipeleft / fade） +2. 硬切概率：保证 30%-50% 的转场是硬切（保持节奏感） +3. 转场时长随机：0.3s ~ 0.8s +4. 转场位置微调：±0.5s 偏移（通过 xfade jitter 实现） +5. 与 #1764 节奏模板协同：长片段之间的转场倾向于更长时长 +""" + +from __future__ import annotations + +import logging +import random + +logger = logging.getLogger(__name__) + +# ── 转场类型池 ────────────────────────────────────────────────────────────── + +#: 非硬切转场类型池（5 种效果，均为 FFmpeg xfade 支持的 transition 名称） +#: 选择标准：视觉效果差异大、FFmpeg 渲染稳定、肉眼可区分 +TRANSITION_POOL: list[str] = [ + "dissolve", # 溶解 + "zoomin", # 缩放（放大进入） + "slideleft", # 左滑入 + "wipeleft", # 左擦除 + "fade", # 淡入淡出 +] + +#: 硬切（无转场效果），由 build_xfade_filter_chain 特殊处理（concat filter） +HARD_CUT = "cut" + +# ── 时长约束 ──────────────────────────────────────────────────────────────── + +#: 随机转场时长下限（秒） +TRANSITION_DURATION_MIN = 0.3 + +#: 随机转场时长上限（秒） +TRANSITION_DURATION_MAX = 0.8 + +# ── 硬切比例 ──────────────────────────────────────────────────────────────── + +#: 硬切概率下限（至少 30% 硬切，保持节奏感） +CUT_RATIO_MIN = 0.3 + +#: 硬切概率上限（最多 50% 硬切，保证足够视觉变化） +CUT_RATIO_MAX = 0.5 + +# ── 位置微调 ──────────────────────────────────────────────────────────────── + +#: 转场位置最大偏移（秒），实际偏移在 [-MAX, +MAX] 均匀分布 +#: 正 = 转场推迟（多留一点前一片段），负 = 转场提前 +TIMING_JITTER_MAX = 0.5 + + +# ── 内部辅助 ──────────────────────────────────────────────────────────────── + + +def _cut_probability_for_pair( + prev_duration: float, + next_duration: float, +) -> float: + """根据相邻片段时长计算硬切概率。 + + 与 #1764 节奏模板协同： + - 两片段都较短（<3s，快节奏）→ 硬切概率更高（节奏更紧凑） + - 两片段都较长（>6s，慢节奏）→ 硬切概率稍低（留出过渡空间） + - 混合场景 → 基准概率（CUT_RATIO_MIN + CUT_RATIO_MAX）/ 2 + + 返回概率始终在 [CUT_RATIO_MIN, CUT_RATIO_MAX] 范围内。 + """ + base = (CUT_RATIO_MIN + CUT_RATIO_MAX) / 2 # 0.4 + avg_dur = (prev_duration + next_duration) / 2 + + if avg_dur < 3.0: + # 快节奏：硬切概率偏高 + return min(CUT_RATIO_MAX, base + 0.1) + elif avg_dur > 6.0: + # 慢节奏：硬切概率偏低（更多视觉过渡） + return max(CUT_RATIO_MIN, base - 0.1) + return base + + +def _apply_jitter( + base_duration: float, + jitter: float, + clip_duration: float, +) -> float: + """给转场时长应用微调偏移，钳制到安全范围。 + + Args: + base_duration: 基础转场时长 + jitter: 偏移量（可正可负） + clip_duration: 较短的相邻片段时长（转场不能超过此值） + + Returns: + 钳制后的实际转场时长 + """ + effective = base_duration + jitter + # 上界：不超过相邻片段时长的 40%（留足内容时间），也不超过 MAX + upper = min(TRANSITION_DURATION_MAX, clip_duration * 0.4) + lower = TRANSITION_DURATION_MIN if jitter < 0 else max(TRANSITION_DURATION_MIN, base_duration) + return max(lower, min(upper, effective)) + + +# ── 核心函数 ──────────────────────────────────────────────────────────────── + + +def generate_transition_plan( + num_transitions: int, + *, + clip_durations: list[float] | None = None, + rng: random.Random | None = None, +) -> list[dict]: + """为变体生成一组随机化的转场计划。 + + 每个转场点独立随机选择类型和时长，硬切比例保持在 30%-50%。 + + Args: + num_transitions: 转场点数量（= 主片段数 - 1） + clip_durations: 各片段时长（用于协同节奏：长片段间转场更长）， + 长度应 >= num_transitions + 1；不足时用默认值 + rng: 可选随机数生成器（测试可注入固定种子） + + Returns: + 转场计划列表，每项： + - effect: str — "cut" 或 TRANSITION_POOL 中某一效果 + - duration: float — 转场时长（cut 为 0.0） + - jitter: float — 位置偏移量（秒，-0.5 ~ +0.5） + """ + if num_transitions <= 0: + return [] + if rng is None: + rng = random.Random() + + if clip_durations is None: + clip_durations = [5.0] * (num_transitions + 1) + + plan: list[dict] = [] + for i in range(num_transitions): + prev_dur = clip_durations[i] if i < len(clip_durations) else 5.0 + next_dur = clip_durations[i + 1] if (i + 1) < len(clip_durations) else 5.0 + + # 计算硬切概率（协同节奏） + cut_prob = _cut_probability_for_pair(prev_dur, next_dur) + + # 随机决定是否硬切 + if rng.random() < cut_prob: + effect = HARD_CUT + duration = 0.0 + else: + effect = rng.choice(TRANSITION_POOL) + base_dur = rng.uniform(TRANSITION_DURATION_MIN, TRANSITION_DURATION_MAX) + # 协同节奏：长片段间转场基础时长更长 + avg_dur = (prev_dur + next_dur) / 2 + if avg_dur > 6.0: + base_dur = min(TRANSITION_DURATION_MAX, base_dur * 1.15) + # 应用位置微调 + jitter = rng.uniform(-TIMING_JITTER_MAX, TIMING_JITTER_MAX) + shorter_clip = min(prev_dur, next_dur) + duration = _apply_jitter(base_dur, jitter, shorter_clip) + + jitter_val = rng.uniform(-TIMING_JITTER_MAX, TIMING_JITTER_MAX) if effect != HARD_CUT else 0.0 + + plan.append( + { + "effect": effect, + "duration": round(duration, 3), + "jitter": round(jitter_val, 3), + } + ) + + return plan diff --git a/packages/domain/variant_plan_selector.py b/packages/domain/variant_plan_selector.py index d40fd6267..8191545eb 100644 --- a/packages/domain/variant_plan_selector.py +++ b/packages/domain/variant_plan_selector.py @@ -29,6 +29,7 @@ import logging import random from packages.domain.plan_generator_utils import _resolve_start_time +from packages.domain.transition_randomizer import generate_transition_plan logger = logging.getLogger(__name__) @@ -225,6 +226,9 @@ def reselect_clips_for_variant( batch_segments.setdefault(aid, []).append(interval) result[idx] = _base_clip_data(src, asset_id=aid, start=start, duration=target_dur) + # ── 4. #1766 转场随机化：为相邻 main 片段对生成随机转场序列 ──────── + _apply_transition_randomization(result, rng) + return [c for c in result if c is not None] @@ -378,6 +382,52 @@ def generate_pixel_perturbation(rng: random.Random | int | None = None) -> dict: return result +def _apply_transition_randomization( + result: list[dict | None], + rng: random.Random, +) -> None: + """#1766 对 result 中相邻 main 片段应用转场随机化（就地修改）。 + + 为每对相邻 main 片段独立选择： + - 转场类型（TRANSITION_POOL 中随机，或硬切） + - 转场时长（0.3s ~ 0.8s，协同片段时长） + - 位置微调 jitter（±0.5s，存入 config["transition_jitter"]） + + 硬切比例保证在 30%-50%。intro/outro 等非 main 片段的转场保持源值不变。 + """ + # 收集 main 片段的索引（按 order 排序） + main_indices = [i for i, c in enumerate(result) if c is not None and c.get("clip_type", "main") == "main"] + + if len(main_indices) < 2: + # 不足 2 个 main 片段，无转场点可随机化 + return + + num_transitions = len(main_indices) - 1 + # 用 main 片段的 duration 作为协同节奏的输入 + clip_durations = [result[i]["duration"] for i in main_indices] + + plan = generate_transition_plan( + num_transitions, + clip_durations=clip_durations, + rng=rng, + ) + + # 将转场计划应用到每对相邻 main 片段 + # plan[k] 是 main_indices[k] → main_indices[k+1] 之间的转场 + # 转场信息存储在"目标 clip"（即每对的第二个）的 transition_effect/duration + for k, transition_info in enumerate(plan): + target_idx = main_indices[k + 1] + if result[target_idx] is None: + continue + clip = result[target_idx] + clip["transition_effect"] = transition_info["effect"] + clip["transition_duration"] = transition_info["duration"] + # jitter 存入 config，供渲染侧 xfade_builder 读取 + cfg = clip.get("config") or {} + cfg["transition_jitter"] = transition_info["jitter"] + clip["config"] = cfg + + def _base_clip_data(src: dict, *, asset_id: str, start: float, duration: float | None = None) -> dict: """从源片段构造落库 dict（保留骨架/转场/文案/速度，替换素材与起点）。""" return { diff --git a/packages/domain/xfade_builder.py b/packages/domain/xfade_builder.py index a4345e7c3..ad454c26e 100755 --- a/packages/domain/xfade_builder.py +++ b/packages/domain/xfade_builder.py @@ -109,6 +109,8 @@ def build_xfade_filter_chain( transitions: list[str], *, transition_duration: float = DEFAULT_TRANSITION_DURATION, + transition_durations: list[float] | None = None, + jitters: list[float] | None = None, output_label: str = "outv", ) -> tuple[str, float]: """构建 xfade 转场滤镜链. @@ -116,11 +118,19 @@ def build_xfade_filter_chain( 对每步 xfade 自动钳制 transition duration，确保 ``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。 + #1766 增强：支持逐转场独立时长（transition_durations）和位置微调（jitters）。 + 传入 transition_durations 时，每个转场点使用各自的时长，而非全局统一值。 + jitters 用于在 offset 上做 ±N 秒微调，实现转场位置随机化。 + Args: clip_durations: 每个片段的时长（必须与 trim 后的实际时长一致） clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"） transitions: 每个片段对应的转场效果（第一个片段的转场被忽略） - transition_duration: 转场时长（秒） + transition_duration: 全局默认转场时长（秒），transition_durations 缺失时 fallback + transition_durations: #1766 逐转场时长列表（与 transitions 等长）， + 第 i 项对应 transitions[i] 的时长；None 时使用 transition_duration + jitters: #1766 逐转场位置偏移列表（秒），与 transitions 等长， + 正值推迟转场、负值提前转场；None 时不做微调 output_label: 最终输出标签 Returns: @@ -150,15 +160,28 @@ def build_xfade_filter_chain( else: first_input_dur = cumulative - total_transition - # 正确的 offset 计算：offset 应相对于累积输出时长 - # offset = 累积输出中，转场开始的时间点 - # = first_input_dur - transition_duration - # 这样每个转场之间的"纯内容"时长等于原始 clip 时长 - offset = max(0.0, first_input_dur - transition_duration) + # #1766: 逐转场时长（优先）或全局默认 + step_duration = ( + transition_durations[i - 1] + if transition_durations and (i - 1) < len(transition_durations) + else transition_duration + ) - # 安全钳制：offset + td 不能超过第一个输入的时长 + # #1766: 位置微调 jitter + jitter = jitters[i - 1] if jitters and (i - 1) < len(jitters) else 0.0 + + # offset = 转场开始点（相对于累积输出起点） + # 基础 offset = first_input_dur - step_duration + # jitter > 0 推迟转场（offset 增大）；jitter < 0 提前转场（offset 减小） + offset = max(0.0, first_input_dur - step_duration) + jitter + + # 安全钳制：offset 不能超出可用范围 + max_offset = max(0.0, first_input_dur - 0.001) + offset = max(0.0, min(offset, max_offset)) + + # 安全钳制 td：offset + td 不能超过第一个输入的时长 available = max(0.0, first_input_dur - offset) - safe_td = min(transition_duration, available) + safe_td = min(step_duration, available) # 同时不能超过剩余总时长 remaining = max(0.0, sum(clip_durations) - cumulative) diff --git a/tests/unit/test_transition_randomizer_1766.py b/tests/unit/test_transition_randomizer_1766.py new file mode 100644 index 000000000..4936432d9 --- /dev/null +++ b/tests/unit/test_transition_randomizer_1766.py @@ -0,0 +1,319 @@ +"""#1766 转场位置与类型随机化测试 (packages/domain/transition_randomizer.py). + +覆盖： +- TRANSITION_POOL 定义（5 种效果） +- generate_transition_plan：硬切比例 30%-50% +- generate_transition_plan：转场时长 0.3s ~ 0.8s +- generate_transition_plan：jitter 在 ±0.5s 范围内 +- 不同 seed 产生不同转场序列 +- 与片段时长协同：长片段间转场更长 +- 边界情况：num_transitions=0、1 个片段 +""" + +from __future__ import annotations + +import random +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +for sub in ("packages", ""): + p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT) + if p not in sys.path: + sys.path.insert(0, p) + +from packages.domain import transition_randomizer as tr # noqa: E402 +from packages.domain.transition_randomizer import ( # noqa: E402 + CUT_RATIO_MAX, + CUT_RATIO_MIN, + HARD_CUT, + TIMING_JITTER_MAX, + TRANSITION_DURATION_MAX, + TRANSITION_DURATION_MIN, + TRANSITION_POOL, + _cut_probability_for_pair, + generate_transition_plan, +) + + +class TestTransitionPool: + """转场类型池定义测试。""" + + def test_pool_has_5_effects(self): + assert len(TRANSITION_POOL) == 5 + + def test_pool_contains_expected_effects(self): + assert "dissolve" in TRANSITION_POOL + assert "zoomin" in TRANSITION_POOL + assert "slideleft" in TRANSITION_POOL + assert "wipeleft" in TRANSITION_POOL + assert "fade" in TRANSITION_POOL + + def test_pool_does_not_contain_cut(self): + assert HARD_CUT not in TRANSITION_POOL + + +class TestConstants: + """常量约束测试。""" + + def test_duration_range(self): + assert TRANSITION_DURATION_MIN == 0.3 + assert TRANSITION_DURATION_MAX == 0.8 + + def test_cut_ratio_range(self): + assert CUT_RATIO_MIN == 0.3 + assert CUT_RATIO_MAX == 0.5 + + def test_jitter_max(self): + assert TIMING_JITTER_MAX == 0.5 + + +class TestCutProbability: + """硬切概率计算测试。""" + + def test_short_clips_higher_cut_prob(self): + """短片段（<3s）→ 硬切概率偏高。""" + prob = _cut_probability_for_pair(2.0, 2.5) + assert prob > 0.4 # 高于基准 + + def test_long_clips_lower_cut_prob(self): + """长片段（>6s）→ 硬切概率偏低。""" + prob = _cut_probability_for_pair(8.0, 7.0) + assert prob < 0.4 # 低于基准 + + def test_medium_clips_base_prob(self): + """中等片段（3-6s）→ 基准概率。""" + prob = _cut_probability_for_pair(4.0, 5.0) + assert abs(prob - 0.4) < 1e-6 + + def test_probability_in_range(self): + """概率始终在 [MIN, MAX] 范围内。""" + for prev in [1.0, 3.0, 5.0, 8.0, 15.0]: + for next_ in [1.0, 3.0, 5.0, 8.0, 15.0]: + prob = _cut_probability_for_pair(prev, next_) + assert CUT_RATIO_MIN <= prob <= CUT_RATIO_MAX + + +class TestGenerateTransitionPlan: + """generate_transition_plan 核心测试。""" + + def test_zero_transitions_returns_empty(self): + assert generate_transition_plan(0) == [] + + def test_negative_transitions_returns_empty(self): + assert generate_transition_plan(-1) == [] + + def test_returns_correct_count(self): + plan = generate_transition_plan(5, rng=random.Random(42)) + assert len(plan) == 5 + + def test_each_item_has_required_keys(self): + plan = generate_transition_plan(3, rng=random.Random(42)) + for item in plan: + assert "effect" in item + assert "duration" in item + assert "jitter" in item + + def test_effects_are_valid(self): + """所有 effect 要么是 cut 要么是 TRANSITION_POOL 中的。""" + plan = generate_transition_plan(20, rng=random.Random(42)) + valid_effects = set(TRANSITION_POOL) | {HARD_CUT} + for item in plan: + assert item["effect"] in valid_effects + + def test_hard_cut_ratio_in_range_many_samples(self): + """100 个转场点，硬切比例在 30%-50%（统计保证）。""" + plan = generate_transition_plan( + 100, + clip_durations=[5.0] * 101, + rng=random.Random(42), + ) + num_cuts = sum(1 for item in plan if item["effect"] == HARD_CUT) + ratio = num_cuts / len(plan) + # 统计波动允许 ±10% 的宽松范围 + assert 0.20 <= ratio <= 0.60, f"硬切比例 {ratio:.2%} 超出宽松范围" + # 更严格的范围检查（±5%） + assert CUT_RATIO_MIN - 0.05 <= ratio <= CUT_RATIO_MAX + 0.05, f"硬切比例 {ratio:.2%} 超出 [25%, 55%] 范围" + + def test_transition_duration_in_range(self): + """非硬切转场的时长在 [0.3, 0.8] 范围内。""" + plan = generate_transition_plan(30, rng=random.Random(42)) + for item in plan: + if item["effect"] != HARD_CUT: + assert ( + TRANSITION_DURATION_MIN <= item["duration"] <= TRANSITION_DURATION_MAX + ), f"转场时长 {item['duration']} 超出 [{TRANSITION_DURATION_MIN}, {TRANSITION_DURATION_MAX}]" + + def test_cut_duration_is_zero(self): + """硬切转场的时长必须为 0。""" + plan = generate_transition_plan(20, rng=random.Random(42)) + for item in plan: + if item["effect"] == HARD_CUT: + assert item["duration"] == 0.0 + + def test_jitter_in_range(self): + """jitter 在 [-0.5, +0.5] 范围内。""" + plan = generate_transition_plan(30, rng=random.Random(42)) + for item in plan: + assert -TIMING_JITTER_MAX <= item["jitter"] <= TIMING_JITTER_MAX, f"jitter {item['jitter']} 超出范围" + + def test_cut_jitter_is_zero(self): + """硬切转场的 jitter 必须为 0。""" + plan = generate_transition_plan(20, rng=random.Random(42)) + for item in plan: + if item["effect"] == HARD_CUT: + assert item["jitter"] == 0.0 + + def test_different_seeds_produce_different_plans(self): + """不同 seed 产生不同的转场序列（至少 2 组不同）。""" + plans_seen = set() + for seed in range(20): + plan = generate_transition_plan(5, rng=random.Random(seed)) + plan_sig = tuple((item["effect"], item["duration"]) for item in plan) + plans_seen.add(plan_sig) + assert len(plans_seen) >= 2, "20 个 seed 只产生 1 种转场序列" + + def test_same_seed_same_plan(self): + """相同 seed 产生相同的转场序列（确定性）。""" + plan1 = generate_transition_plan(5, rng=random.Random(42)) + plan2 = generate_transition_plan(5, rng=random.Random(42)) + assert plan1 == plan2 + + def test_long_clips_longer_transitions(self): + """长片段（>6s）之间的转场倾向于比短片段更长。""" + # 长片段 + long_plan = generate_transition_plan( + 20, + clip_durations=[10.0] * 21, + rng=random.Random(42), + ) + # 短片段 + short_plan = generate_transition_plan( + 20, + clip_durations=[2.0] * 21, + rng=random.Random(42), + ) + # 长片段的非硬切转场平均时长 + long_durs = [item["duration"] for item in long_plan if item["effect"] != HARD_CUT] + short_durs = [item["duration"] for item in short_plan if item["effect"] != HARD_CUT] + + if long_durs and short_durs: + avg_long = sum(long_durs) / len(long_durs) + avg_short = sum(short_durs) / len(short_durs) + # 长片段平均转场时长 >= 短片段（协同节奏） + assert avg_long >= avg_short * 0.95, f"长片段转场 {avg_long:.3f}s 不应显著短于短片段 {avg_short:.3f}s" + + def test_clip_durations_none_uses_default(self): + """clip_durations=None 时使用默认值 5.0。""" + plan = generate_transition_plan(3, rng=random.Random(42)) + assert len(plan) == 3 + + def test_fewer_clip_durations_than_needed(self): + """clip_durations 长度不足时用默认值补齐。""" + plan = generate_transition_plan( + 5, + clip_durations=[4.0, 5.0], # 只需前 2 个 + rng=random.Random(42), + ) + assert len(plan) == 5 + + +class TestTransitionRandomizationIntegration: + """转场随机化与变体生成集成测试。""" + + def test_reselect_produces_different_transitions(self): + """多次 reselect_clips_for_variant 产生不同的转场序列。""" + from packages.domain.variant_plan_selector import reselect_clips_for_variant + + source_clips = [ + { + "order": i, + "asset_id": f"asset_{i}", + "start_time": 0.0, + "duration": 5.0, + "clip_type": "main", + "playback_speed": 1.0, + "transition_effect": "cut", + "transition_duration": 0.0, + "text_content": "", + "config": {}, + } + for i in range(4) + ] + asset_durations = {f"asset_{i}": 30.0 for i in range(4)} + + transition_seqs = set() + for seed in range(5): + rng = random.Random(seed) + result = reselect_clips_for_variant( + source_clips, + list(asset_durations.keys()), + asset_durations=asset_durations, + rng=rng, + ) + seq = tuple( + (c.get("transition_effect"), round(c.get("transition_duration", 0), 2)) + for c in result + if c.get("clip_type") == "main" + ) + transition_seqs.add(seq) + + assert len(transition_seqs) >= 2, f"5 个 seed 只产生 {len(transition_seqs)} 种转场序列" + + def test_reselect_preserves_non_main_transitions(self): + """非 main 片段（intro/outro）的转场不被随机化。""" + from packages.domain.variant_plan_selector import reselect_clips_for_variant + + source_clips = [ + { + "order": 0, + "asset_id": "intro_asset", + "start_time": 0.0, + "duration": 3.0, + "clip_type": "intro", + "playback_speed": 1.0, + "transition_effect": "fade", + "transition_duration": 0.5, + "text_content": "", + "config": {}, + }, + { + "order": 1, + "asset_id": "a1", + "start_time": 0.0, + "duration": 5.0, + "clip_type": "main", + "playback_speed": 1.0, + "transition_effect": "cut", + "transition_duration": 0.0, + "text_content": "", + "config": {}, + }, + { + "order": 2, + "asset_id": "a2", + "start_time": 0.0, + "duration": 5.0, + "clip_type": "main", + "playback_speed": 1.0, + "transition_effect": "cut", + "transition_duration": 0.0, + "text_content": "", + "config": {}, + }, + ] + asset_durations = {"intro_asset": 10.0, "a1": 30.0, "a2": 30.0} + + result = reselect_clips_for_variant( + source_clips, + list(asset_durations.keys()), + asset_durations=asset_durations, + rng=random.Random(42), + ) + + # intro 片段的转场保持不变 + intro_clip = next(c for c in result if c["clip_type"] == "intro") + assert intro_clip["transition_effect"] == "fade" + assert intro_clip["transition_duration"] == 0.5 diff --git a/tests/unit/test_xfade_per_transition_1766.py b/tests/unit/test_xfade_per_transition_1766.py new file mode 100644 index 000000000..a820962ca --- /dev/null +++ b/tests/unit/test_xfade_per_transition_1766.py @@ -0,0 +1,261 @@ +"""#1766 xfade_builder 逐转场时长与位置微调测试. + +覆盖： +- build_xfade_filter_chain：transition_durations 参数（逐转场独立时长） +- build_xfade_filter_chain：jitters 参数（位置微调偏移） +- 向后兼容：不传新参数时行为不变 +- TransitionEngine.build_xfade_chain：透传新参数 +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +for sub in ("packages", ""): + p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT) + if p not in sys.path: + sys.path.insert(0, p) + +from packages.domain.xfade_builder import ( # noqa: E402 + DEFAULT_TRANSITION_DURATION, + build_xfade_filter_chain, +) + + +class TestBackwardCompatibility: + """向后兼容测试：不传新参数时行为不变。""" + + def test_single_transition_duration(self): + """全局 transition_duration 仍有效。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + ) + assert "xfade" in result + assert "duration=0.500" in result + assert dur > 0 + + def test_default_transition_duration(self): + """不传 transition_duration 时使用默认值 0.5。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + ) + assert "duration=0.500" in result + + +class TestPerTransitionDurations: + """逐转场独立时长测试。""" + + def test_different_durations_per_transition(self): + """每个转场使用不同的时长。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0, 5.0], + clip_video_labels=["v0", "v1", "v2"], + transitions=["cut", "fade", "dissolve"], + transition_durations=[0.3, 0.8], # 第 1 个转场 0.3s，第 2 个 0.8s + ) + assert "duration=0.300" in result + assert "duration=0.800" in result + + def test_partial_durations_fallback_to_global(self): + """transition_durations 长度不足时 fallback 到 transition_duration。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0, 5.0], + clip_video_labels=["v0", "v1", "v2"], + transitions=["cut", "fade", "dissolve"], + transition_duration=0.5, + transition_durations=[0.3], # 只有第一个，第二个 fallback 到 0.5 + ) + assert "duration=0.300" in result + assert "duration=0.500" in result + + def test_empty_durations_uses_global(self): + """transition_durations=[] 时使用全局 transition_duration。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.6, + transition_durations=[], + ) + assert "duration=0.600" in result + + def test_none_durations_uses_global(self): + """transition_durations=None 时使用全局 transition_duration。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.6, + transition_durations=None, + ) + assert "duration=0.600" in result + + def test_duration_clamped_by_clip_length(self): + """转场时长不能超过相邻片段时长。""" + result, dur = build_xfade_filter_chain( + clip_durations=[2.0, 2.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_durations=[1.5], # 1.5s > 2.0s * 0.4，会被钳制 + ) + # 钳制到可用范围内 + assert "duration=" in result + + def test_total_duration_reflects_per_transition(self): + """总时长反映逐转场的重叠量。""" + # 使用 0.3s 转场 + _, dur_short = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_durations=[0.3], + ) + # 使用 0.8s 转场 + _, dur_long = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_durations=[0.8], + ) + # 更长转场 → 更多重叠 → 总时长更短 + assert dur_long < dur_short + + +class TestJitters: + """位置微调 jitter 测试。""" + + def test_positive_jitter_delays_transition(self): + """正 jitter 推迟转场（offset 增大）。""" + result_no_jitter, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + ) + result_with_jitter, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + jitters=[0.3], # 正 jitter：推迟转场 + ) + # 提取 offset 值 + import re + + offset_no = float(re.search(r"offset=([\d.]+)", result_no_jitter).group(1)) + offset_yes = float(re.search(r"offset=([\d.]+)", result_with_jitter).group(1)) + assert offset_yes > offset_no + + def test_negative_jitter_advances_transition(self): + """负 jitter 提前转场（offset 减小）。""" + result_no_jitter, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + ) + result_with_jitter, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + jitters=[-0.3], # 负 jitter：提前转场 + ) + import re + + offset_no = float(re.search(r"offset=([\d.]+)", result_no_jitter).group(1)) + offset_yes = float(re.search(r"offset=([\d.]+)", result_with_jitter).group(1)) + assert offset_yes < offset_no + + def test_jitter_clamped_to_valid_range(self): + """jitter 不会使 offset 超出有效范围（>=0）。""" + result, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + jitters=[-100.0], # 极大负 jitter + ) + import re + + offset = float(re.search(r"offset=([\d.]+)", result).group(1)) + assert offset >= 0.0 + + def test_per_transition_jitters(self): + """每个转场可以有独立的 jitter。""" + result, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0, 5.0], + clip_video_labels=["v0", "v1", "v2"], + transitions=["cut", "fade", "dissolve"], + transition_duration=0.5, + jitters=[0.2, -0.1], + ) + import re + + offsets = [float(m.group(1)) for m in re.finditer(r"offset=([\d.]+)", result)] + assert len(offsets) == 2 + + def test_empty_jitters_no_effect(self): + """jitters=[] 等同于无 jitter。""" + result_no, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + ) + result_empty, _ = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=0.5, + jitters=[], + ) + assert result_no == result_empty + + +class TestEdgeCases: + """边界情况测试。""" + + def test_single_clip_with_durations(self): + """单片段传入 transition_durations 不报错。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0], + clip_video_labels=["v0"], + transitions=["cut"], + transition_durations=[0.5], + ) + assert "copy" in result + assert dur == 5.0 + + def test_empty_clips(self): + """空片段列表返回空字符串。""" + result, dur = build_xfade_filter_chain( + clip_durations=[], + clip_video_labels=[], + transitions=[], + transition_durations=[], + jitters=[], + ) + assert result == "" + assert dur == 0.0 + + def test_all_cut_transitions(self): + """全硬切场景（不进入 xfade，由调用方处理 concat）。""" + result, dur = build_xfade_filter_chain( + clip_durations=[5.0, 5.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "cut"], + transition_durations=[0.0, 0.0], + ) + # cut 转场仍然会生成 xfade 滤镜（因为底层不区分 cut） + # 调用方（unified_render_service）负责检测全硬切并走 concat + assert dur > 0