From cfa5495300f8318902a3e187c48e63d889e2c687 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sat, 18 Jul 2026 14:46:54 +0800 Subject: [PATCH] =?UTF-8?q?fix(P0):=20unified=E6=B8=B2=E6=9F=93=E5=BC=95?= =?UTF-8?q?=E6=93=8E4=E9=A1=B9normalize=E8=A1=A5=E5=85=A8=EF=BC=8C?= =?UTF-8?q?=E4=BF=AE=E5=A4=8Dconcat=20exit=3D234?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 视频： - main/broll层从increase+crop改为decrease+pad（留黑边不裁剪） - background层保持increase+crop（背景铺满） 音频： - concat主音频前每个clip加aformat归一化(48000Hz/stereo/fltp) - 独立音轨amix前加aformat归一化 - 单clip直通模式输出也加采样率/声道归一化 测试： - 新增20+单元测试覆盖4项normalize - 更新旧的crop测试为pad模式 - _make_service支持output_fps/output_width/output_height参数 --- apps/worker/video_processing/render_audio.py | 28 +- .../unified_render_service.py | 18 +- tests/unit/test_unified_render_service.py | 525 +++++++++++++++++- 3 files changed, 545 insertions(+), 26 deletions(-) diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index c8df40640..45d36d683 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -268,6 +268,10 @@ def concat_main_audio( "aac", "-b:a", "128k", + "-ar", + "48000", + "-ac", + "2", ] if trim_start > 0: command.extend(["-ss", f"{trim_start:.3f}"]) @@ -299,6 +303,9 @@ def concat_main_audio( if reverse_filter: audio_filters.append(reverse_filter) + # aformat 归一化：统一输出格式为 48000Hz + stereo + fltp + audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp") + filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"] if video_duration > 0 and final_duration < adjusted_duration: filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]") @@ -362,6 +369,10 @@ def concat_main_audio( if reverse_filter: audio_filters.append(reverse_filter) + # aformat 归一化：统一采样率48000Hz + 双声道stereo + fltp采样格式 + # concat filter 要求所有输入音频参数完全一致，否则 exit=234 失败 + audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp") + filter_parts.append(f"[{i}:a]{','.join(audio_filters)}[a{i}]") audio_labels = "".join(f"[a{i}]" for i in range(len(clips))) @@ -415,19 +426,24 @@ def mix_with_independent_audio( input_idx = 0 + # aformat 归一化参数（所有音频在 concat/amix 前必须统一） + AFORMAT = "aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp" + # 1. 主图层音频 concat if main_clips: for clip in main_clips: input_args.extend(["-i", str(clip.local_path)]) effective_duration = clip_effective_duration(clip) trim_start = getattr(clip, "start_time", 0) or 0 + clip_filters = [] if effective_duration > 0: - filter_parts.append( - f"[{input_idx}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," - f"asetpts=PTS-STARTPTS[ma{input_idx}]" - ) + clip_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}") + clip_filters.append("asetpts=PTS-STARTPTS") else: - filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]") + clip_filters.append("asetpts=PTS-STARTPTS") + # aformat 归一化：concat/amix 前统一音频参数，否则不同采样率/声道会失败 + clip_filters.append(AFORMAT) + filter_parts.append(f"[{input_idx}:a]{','.join(clip_filters)}[ma{input_idx}]") input_idx += 1 if len(main_clips) == 1: @@ -450,6 +466,8 @@ def mix_with_independent_audio( filters.append("asetpts=PTS-STARTPTS") if volume != 1.0: filters.append(f"volume={volume}") + # aformat 归一化：amix 前统一音频参数 + filters.append(AFORMAT) filter_parts.append(f"[{input_idx}:a]{','.join(filters)}[{label}]") mix_labels.append(label) input_idx += 1 diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index ad4b83601..3e97eeaa3 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -996,15 +996,19 @@ class UnifiedRenderService: if reverse_filter: filters.append(reverse_filter) - # scale + crop（铺满裁剪） + # scale + pad（等比缩放+留黑边） if role in ("overlay", "corner_voice"): pip_w = int(self.output_width * _PIP_SCALE) pip_h = int(self.output_height * _PIP_SCALE) filters.append(f"scale={pip_w}:{pip_h}") - else: - # main / broll / background: 铺满裁剪 + elif role == "background": + # background: 铺满裁剪（作为底图，覆盖全屏） filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase") filters.append(f"crop={self.output_width}:{self.output_height}") + else: + # main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容） + filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease") + filters.append(f"pad={self.output_width}:{self.output_height}:(ow-iw)/2:(oh-ih)/2:black") # 调色滤镜 color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade")) @@ -1333,12 +1337,12 @@ class UnifiedRenderService: ) filters.append(f"crop={self.output_width}:{self.output_height}") else: - # main / broll: 铺满裁剪（scale to cover + center crop） - # 对齐链路A编辑器合成行为，与主流短视频平台一致 + # main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容） + # concat 要求所有输入分辨率完全一致，pad 模式确保不同宽高比的素材都能正常拼接 filters.append( - f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase" + f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease" ) - filters.append(f"crop={self.output_width}:{self.output_height}") + filters.append(f"pad={self.output_width}:{self.output_height}:(ow-iw)/2:(oh-ih)/2:black") # 调色滤镜（每个 clip 独立的 color grade 配置） color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade")) diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py index 3c85a35a0..a9f025f11 100755 --- a/tests/unit/test_unified_render_service.py +++ b/tests/unit/test_unified_render_service.py @@ -85,6 +85,9 @@ def _make_service( clips: list[FakeClip] | None = None, asset_paths: dict[str, Path] | None = None, work_dir: Path | None = None, + output_fps: int = 25, + output_width: int = 1280, + output_height: int = 720, ) -> UnifiedRenderService: """创建测试用的 UnifiedRenderService 实例。 @@ -104,6 +107,9 @@ def _make_service( clips=clips, asset_path_map=asset_paths, work_dir=work_dir, + output_width=output_width, + output_height=output_height, + output_fps=output_fps, ) @@ -486,10 +492,10 @@ class TestBuildFilterComplex: f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}" ) - def test_main_clip_uses_fill_crop_strategy(self): - """main/broll clip 使用铺满裁剪策略（scale increase + crop），不是等比+黑边。 + def test_main_clip_uses_pad_strategy(self): + """main clip 使用 scale+pad 保持比例留黑边（不裁剪内容）。 - 对齐链路A编辑器合成行为，与主流短视频平台一致。 + 渲染引擎不能挑素材，任何素材进来都统一规格后出片；pad留黑边保证内容完整。 """ clips = [_make_clip("c1", "main", order=0, duration=5.0)] asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} @@ -500,15 +506,16 @@ class TestBuildFilterComplex: layers = svc._group_clips_into_layers(resolved) fc, _ = svc._build_filter_complex(layers) - # 验证：scale 使用 force_original_aspect_ratio=increase（铺满） - assert "force_original_aspect_ratio=increase" in fc - # 验证：有 crop（居中裁剪） - assert "crop=1280:720" in fc - # 验证：没有 pad（不是黑边模式） - assert "pad=" not in fc + # 验证：scale 使用 force_original_aspect_ratio=decrease（等比缩小） + assert "force_original_aspect_ratio=decrease" in fc + # 验证：有 pad（留黑边） + assert "pad=" in fc + assert "black" in fc + # 验证：没有 crop（不是裁剪模式） + assert "crop=" not in fc - def test_broll_clip_uses_fill_crop_strategy(self): - """broll clip 同样使用铺满裁剪策略。""" + def test_broll_clip_uses_pad_strategy(self): + """broll clip 同样使用 scale+pad 留黑边策略。""" clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)] asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} svc = _make_service(clips, asset_paths) @@ -518,9 +525,9 @@ class TestBuildFilterComplex: layers = svc._group_clips_into_layers(resolved) fc, _ = svc._build_filter_complex(layers) - assert "force_original_aspect_ratio=increase" in fc - assert "crop=1280:720" in fc - assert "pad=" not in fc + assert "force_original_aspect_ratio=decrease" in fc + assert "pad=" in fc + assert "crop=" not in fc def test_background_uses_fill_crop_strategy(self): """background 层也使用铺满裁剪（已有的行为，保持一致）。""" @@ -1765,3 +1772,493 @@ class TestExtractAudioUsesRunFfmpeg: mock_run.side_effect = subprocess.CalledProcessError(1, "ffmpeg", stderr="error") with pytest.raises(RuntimeError, match="音频提取失败"): svc._extract_audio(video_path, output_path) + + +# ══════════════════════════════════════════════════════════════════════════════ +# concat 前 4 项归一化测试（修复 exit=234） +# ══════════════════════════════════════════════════════════════════════════════ + + +class TestConcatNormalizeVideoResolution: + """视频分辨率归一化：scale + pad 保持比例留黑边，确保 concat 前分辨率一致。 + + 对应 4 项 normalize 之：视频分辨率（scale + pad 到 canvas_width × canvas_height） + 修复前：main/broll 用 scale+crop（铺满裁剪），横屏素材内容被裁掉 + 修复后：main/broll 用 scale+pad（留黑边），保持原始比例不裁剪 + """ + + def test_main_layer_uses_pad_not_crop_in_filter_complex(self): + """_build_filter_complex 中 main 层用 scale+pad（letterbox），不用 crop。""" + clips = [_make_clip("c1", "main", order=0, duration=3.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) + + # 必须用 pad（留黑边），不能用 crop（裁剪） + assert "force_original_aspect_ratio=decrease" in fc, "main 层应使用 decrease 模式（等比缩放到画布内）" + assert "pad=" in fc, "main 层应有 pad 滤镜（留黑边到目标分辨率）" + assert "(ow-iw)/2:(oh-ih)/2:black" in fc, "pad 应该居中+黑边" + # 注意：background 层也用 crop，但 main/broll 应该用 pad + # 检查第一个 [0:v] 处理链（第一个 clip 是 main） + first_v_chain = fc.split("[v0]")[0] + assert "crop=" not in first_v_chain, "main 层不应该有 crop 滤镜" + + def test_broll_layer_uses_pad_not_crop_in_filter_complex(self): + """_build_filter_complex 中 b_roll 层用 scale+pad，不用 crop。""" + clips = [ + _make_clip("c1", "b_roll", order=0, duration=3.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) + + assert "force_original_aspect_ratio=decrease" in fc + assert "pad=" in fc + first_v_chain = fc.split("[v0]")[0] + assert "crop=" not in first_v_chain, "broll 层不应该有 crop 滤镜" + + def test_background_layer_still_uses_crop_in_filter_complex(self): + """_build_filter_complex 中 background 层保持 scale+crop（作为底图铺满）。""" + clips = [ + _make_clip("bg1", "background", order=0, duration=3.0), + _make_clip("c1", "main", order=0, duration=3.0), + ] + asset_paths = { + "asset_bg1.mp4": Path("/tmp/asset_bg1.mp4"), + "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) + + # background 应该用 increase + crop（cover 模式） + # 统计 crop 出现次数：background 1次 + 其他位置（如果有） + # 关键是 background 走 increase 模式 + assert "force_original_aspect_ratio=increase" in fc, "background 层应使用 increase 模式（铺满裁剪）" + + def test_multi_clip_concat_all_same_resolution(self): + """多 clip concat 时，所有 clip 预处理后分辨率一致（pad 到同一尺寸）。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.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) + + # 两个 clip 都应该有 pad + pad_count = fc.count("pad=") + assert pad_count >= 2, f"至少应有 2 个 pad（每个 clip 一个），实际 {pad_count}" + # 都用 decrease 模式 + decrease_count = fc.count("force_original_aspect_ratio=decrease") + assert decrease_count >= 2, f"至少应有 2 个 decrease，实际 {decrease_count}" + # 有 concat + assert "concat=n=2:v=1:a=0" in fc + + def test_pass_through_main_uses_pad_not_crop(self): + """_render_pass_through 中 main 层用 scale+pad，不用 crop。""" + clips = [_make_clip("c1", "main", order=0, duration=3.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), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + + # 直通模式 main 层应该用 pad + assert "force_original_aspect_ratio=decrease" in cmd_str, "直通模式 main 层应使用 decrease 模式" + assert "pad=" in cmd_str, "直通模式应有 pad 滤镜" + # 检查 vf 中没有 crop + vf_idx = cmd.index("-vf") + vf_value = cmd[vf_idx + 1] + assert "crop=" not in vf_value, "直通模式 main 层 vf 中不应该有 crop" + + def test_pass_through_background_uses_crop(self): + """_render_pass_through 中 background 层保持 scale+crop。""" + clips = [_make_clip("bg1", "background", order=0, duration=3.0)] + asset_paths = {"asset_bg1.mp4": Path("/tmp/asset_bg1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + + # background 层应该用 increase + crop + assert "force_original_aspect_ratio=increase" in cmd_str + vf_idx = cmd.index("-vf") + vf_value = cmd[vf_idx + 1] + assert "crop=" in vf_value, "background 层应有 crop 滤镜" + + +class TestConcatNormalizeVideoFps: + """视频帧率归一化：fps 滤镜统一到目标 fps。 + + 对应 4 项 normalize 之：视频帧率（fps 滤镜统一到目标 fps） + """ + + def test_filter_complex_has_fps_filter(self): + """_build_filter_complex 中每个 clip 都有 fps 滤镜。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.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, output_fps=30) + + 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) + + fps_count = fc.count("fps=30") + assert fps_count >= 2, f"至少应有 2 个 fps=30（每个 clip 一个），实际 {fps_count}" + + def test_fps_after_pad_before_final_setpts(self): + """fps 滤镜在 pad 之后、setpts 之前（确保分辨率和帧率都统一后再归一化 PTS）。""" + import re + + clips = [_make_clip("c1", "main", order=0, duration=3.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths, output_fps=30) + + 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) + + # 提取第一个 clip 的视频处理链 + chain_str = fc.split("[v0]")[0] + + pad_pos = chain_str.find("pad=") + fps_pos = chain_str.find("fps=") + assert pad_pos >= 0, "应找到 pad 滤镜" + assert fps_pos >= 0, "应找到 fps 滤镜" + assert fps_pos > pad_pos, "fps 应在 pad 之后" + + +class TestConcatNormalizeAudioFormat: + """音频格式归一化：aformat 统一为 stereo + 48000Hz + fltp。 + + 对应 4 项 normalize 之：音频声道（aformat 统一为 stereo + 48000Hz + fltp） + 修复前：concat 前音频无归一化，不同采样率/声道导致 exit=234 + 修复后：concat/amix 前所有音频统一为 48000Hz + stereo + fltp + """ + + def test_multi_clip_concat_has_aformat(self): + """多 clip 音频 concat 前，每个 clip 都有 aformat 归一化。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.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), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 4.5) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + + # 必须有 aformat 归一化 + assert "aformat=" in cmd_str, "concat 前应有 aformat 滤镜" + assert "sample_rates=48000" in cmd_str, "采样率应统一为 48000Hz" + assert "channel_layouts=stereo" in cmd_str, "声道应统一为 stereo" + assert "sample_fmts=fltp" in cmd_str, "采样格式应统一为 fltp" + + # 两个 clip 都应该有 aformat + aformat_count = cmd_str.count("aformat=") + assert aformat_count >= 2, f"每个 clip 都应有 aformat，实际 {aformat_count} 个" + + # 有 concat + assert "concat=n=2:v=0:a=1" in cmd_str + + def test_aformat_before_concat(self): + """aformat 应在 concat 之前（每个 clip 处理链中 aformat 在 concat 之前）。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.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), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 4.5) + + cmd = mock_run.call_args[0][0] + fc_idx = cmd.index("-filter_complex") + fc_str = cmd[fc_idx + 1] + + # 找到 concat 的位置 + concat_pos = fc_str.find("concat=") + assert concat_pos > 0, "应找到 concat filter" + + # 在 concat 之前的部分应该有 aformat + before_concat = fc_str[:concat_pos] + aformat_before_count = before_concat.count("aformat=") + assert aformat_before_count >= 2, f"concat 之前每个 clip 都应有 aformat，实际 {aformat_before_count} 个" + + def test_single_clip_audio_has_normalized_output(self): + """单 clip 音频输出也应统一格式（一致性保障）。""" + 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), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 5.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + + # 单 clip 简单路径应该有 -ar 48000 和 -ac 2 + assert "-ar" in cmd, "单 clip 音频应指定采样率" + ar_idx = cmd.index("-ar") + assert cmd[ar_idx + 1] == "48000", "采样率应为 48000" + assert "-ac" in cmd, "单 clip 音频应指定声道数" + ac_idx = cmd.index("-ac") + assert cmd[ac_idx + 1] == "2", "声道数应为 2（stereo）" + + def test_independent_audio_track_has_aformat(self): + """独立音频轨在 amix 前也有 aformat 归一化。""" + clips = [ + _make_clip("c1", "main", order=0, duration=5.0), + _make_clip( + "audio1", + "main", + order=0, + duration=5.0, + config={"role": "audio", "volume": 0.5}, + ), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_audio1.mp4": Path("/tmp/asset_audio1.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 5.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + + # 应该有 aformat + assert "aformat=" in cmd_str + assert "sample_rates=48000" in cmd_str + assert "channel_layouts=stereo" in cmd_str + # amix 应该存在 + assert "amix" in cmd_str + + +class TestConcatNormalizeAudioCodec: + """音频编码归一化：统一为 aac。 + + 对应 4 项 normalize 之：音频编码（统一为 aac） + """ + + def test_multi_clip_output_is_aac(self): + """多 clip concat 后输出编码为 aac。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.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), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 4.5) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + assert "aac" in cmd, "音频输出编码应为 aac" + assert "-b:a" in cmd, "应指定音频码率" + + def test_single_clip_output_is_aac(self): + """单 clip 音频输出编码为 aac。""" + 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), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 5.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + assert "aac" in cmd + + +class TestConcatNormalizeFourItemsComplete: + """4 项 normalize 完整性验证：one_take 多素材 concat 场景全链路检查。 + + 模拟 one_take 模式（多个 main clip）的完整渲染路径， + 验证 concat 前所有 4 项 normalize 都已到位。 + """ + + def _make_one_take_clips(self): + """构造 one_take 模式的典型素材：3 个 main clip（模拟横屏+竖屏混合）。""" + return [ + _make_clip("c1", "main", order=0, duration=5.0), + _make_clip("c2", "main", order=1, duration=4.0), + _make_clip("c3", "main", order=2, duration=3.0), + ] + + def test_video_resolution_normalized(self): + """[1/4] 视频分辨率：所有 clip 都有 pad 到统一分辨率。""" + clips = self._make_one_take_clips() + asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)} + svc = _make_service(clips, asset_paths) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + # 3 个 clip 都应该有 pad（decrease + black padding） + pad_count = fc.count("pad=") + decrease_count = fc.count("force_original_aspect_ratio=decrease") + assert pad_count >= 3, f"3个 clip 都应有 pad，实际 {pad_count} 个" + assert decrease_count >= 3, f"3个 clip 都应用 decrease 模式，实际 {decrease_count} 个" + + def test_video_fps_normalized(self): + """[2/4] 视频帧率：所有 clip 都有 fps 滤镜。""" + clips = self._make_one_take_clips() + asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)} + svc = _make_service(clips, asset_paths, output_fps=30) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + fps_count = fc.count("fps=30") + assert fps_count >= 3, f"3个 clip 都应有 fps=30，实际 {fps_count} 个" + + def test_audio_format_normalized(self): + """[3/4] 音频格式：所有 clip concat 前都有 aformat 归一化。""" + clips = self._make_one_take_clips() + asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=10.0), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 12.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + + aformat_count = cmd_str.count("aformat=") + assert aformat_count >= 3, f"3个 clip 都应有 aformat，实际 {aformat_count} 个" + assert "sample_rates=48000" in cmd_str + assert "channel_layouts=stereo" in cmd_str + assert "sample_fmts=fltp" in cmd_str + + def test_audio_codec_aac(self): + """[4/4] 音频编码：输出为 aac。""" + clips = self._make_one_take_clips() + asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=10.0), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() + mix_audio(ctx, layers, 12.0) + + assert mock_run.called + cmd = mock_run.call_args[0][0] + assert "aac" in cmd + assert "concat=n=3:v=0:a=1" in " ".join(cmd)