diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py old mode 100644 new mode 100755 index a01c42c7e..0235e65a8 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -339,8 +339,8 @@ class UnifiedRenderService: ) filters.append(f"pad={self.output_width}:{self.output_height}" ":(ow-iw)/2:(oh-ih)/2:black") - filters.append(f"fps={self.output_fps}") filters.append("setpts=PTS-STARTPTS") + filters.append(f"fps={self.output_fps}") filter_str = f"[{i}:v]{','.join(filters)}[{label}]" filter_parts.append(filter_str) diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py old mode 100644 new mode 100755 index 4670c132b..8cae2ad27 --- a/tests/unit/test_unified_render_service.py +++ b/tests/unit/test_unified_render_service.py @@ -356,6 +356,89 @@ class TestBuildFilterComplex: assert "overlay=" in fc assert "[final_video]" in fc + def test_setpts_before_fps_in_xfade_inputs(self): + """多视频 xfade 模式:setpts=PTS-STARTPTS 必须在 fps 之前,确保 xfade 时各片段 PTS 一致。 + + 构造两个不同时长的视频片段,验证生成的 filter_complex 中每个片段的 + 预处理滤镜链里 setpts 都在 fps 前面。 + """ + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=5.0), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + # 确保 xfade 存在 + assert "xfade=" in fc + + # 提取每个 clip 的预处理滤镜链([i:v]...[vi] 部分) + # 验证:每个 clip 滤镜链中,setpts=PTS-STARTPTS 的最后一次出现 + # 必须在 fps= 的前面(PTS 归一化后再统一帧率) + import re + + clip_pattern = re.compile(r"\[(\d+):v\](.+?)\[v\d+\]") + matches = clip_pattern.findall(fc) + assert len(matches) == 2, f"Expected 2 clip preprocessing chains, got {len(matches)}" + + for idx, chain_str in matches: + # 找到所有 setpts 和 fps 的位置 + setpts_positions = [m.start() for m in re.finditer(r"setpts=PTS-STARTPTS", chain_str)] + fps_positions = [m.start() for m in re.finditer(r"fps=\d+", chain_str)] + + assert setpts_positions, f"clip {idx}: 未找到 setpts=PTS-STARTPTS" + assert fps_positions, f"clip {idx}: 未找到 fps=" + + # 最后一个 setpts 必须在第一个 fps 之前 + last_setpts = max(setpts_positions) + first_fps = min(fps_positions) + assert last_setpts < first_fps, ( + f"clip {idx}: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" + f"滤镜链: {chain_str}" + ) + + def test_setpts_before_fps_single_clip(self): + """单视频模式(一镜到底):setpts 也必须在 fps 之前。 + + 单视频虽然没有 xfade,但滤镜链顺序应保持一致,确保 PTS 处理逻辑统一。 + """ + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + import re + + clip_pattern = re.compile(r"\[(\d+):v\](.+?)\[v\d+\]") + matches = clip_pattern.findall(fc) + assert len(matches) == 1 + + chain_str = matches[0][1] + setpts_positions = [m.start() for m in re.finditer(r"setpts=PTS-STARTPTS", chain_str)] + fps_positions = [m.start() for m in re.finditer(r"fps=\d+", chain_str)] + + assert setpts_positions, "单视频: 未找到 setpts=PTS-STARTPTS" + assert fps_positions, "单视频: 未找到 fps=" + + last_setpts = max(setpts_positions) + first_fps = min(fps_positions) + assert last_setpts < first_fps, ( + f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" + f"滤镜链: {chain_str}" + ) + def test_empty_layers_raises(self): """空图层列表抛出 ValueError。""" svc = _make_service()