feat: 转场位置与类型随机化 (#1766) #1786
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user