diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 329cb2d62..7dd31da46 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -512,6 +512,9 @@ def mix_with_independent_audio( clip_filters.append("asetpts=PTS-STARTPTS") else: clip_filters.append("asetpts=PTS-STARTPTS") + vol = _clip_volume(clip) + if abs(vol - 1.0) >= 1e-6: + clip_filters.append(f"volume={vol:.4f}") # aformat 归一化：concat/amix 前统一音频参数，否则不同采样率/声道会失败 clip_filters.append(AFORMAT) filter_parts.append(f"[{input_idx}:a]{','.join(clip_filters)}[ma{input_idx}]") diff --git a/packages/domain/ass_subtitle_builder.py b/packages/domain/ass_subtitle_builder.py index e80b19ff2..0948bccd5 100755 --- a/packages/domain/ass_subtitle_builder.py +++ b/packages/domain/ass_subtitle_builder.py @@ -197,26 +197,33 @@ def _wrap_title_text( if available_width <= 0: return text - lines: list[str] = [] - current_line = "" - current_width = 0.0 + # 先按已有 \N 分段，每段独立自动换行，最后用 \N 拼回 + segments = text.split("\\N") + wrapped_segments: list[str] = [] - for ch in text: - # CJK 字符按全角估算，其他按半角 - char_width = float(font_size) if ord(ch) > 0x2E80 else font_size * 0.55 + for seg in segments: + lines: list[str] = [] + current_line = "" + current_width = 0.0 - if current_width + char_width > available_width and current_line: + for ch in seg: + # CJK 字符按全角估算，其他按半角 + char_width = float(font_size) if ord(ch) > 0x2E80 else font_size * 0.55 + + if current_width + char_width > available_width and current_line: + lines.append(current_line) + current_line = ch + current_width = char_width + else: + current_line += ch + current_width += char_width + + if current_line: lines.append(current_line) - current_line = ch - current_width = char_width - else: - current_line += ch - current_width += char_width - if current_line: - lines.append(current_line) + wrapped_segments.append("\\N".join(lines)) - return "\\N".join(lines) + return "\\N".join(wrapped_segments) def build_ass_content( diff --git a/tests/unit/test_ass_subtitle_builder.py b/tests/unit/test_ass_subtitle_builder.py index b2f6e2da4..739f28454 100755 --- a/tests/unit/test_ass_subtitle_builder.py +++ b/tests/unit/test_ass_subtitle_builder.py @@ -504,6 +504,35 @@ class TestWrapTitleText: """字号为0时直接返回原文。""" assert _wrap_title_text("测试", 1080, 0) == "测试" + def test_preserves_explicit_newline(self): + """已有的 \\N 换行标记应保留，不被当普通字符算宽度。""" + text = "第一行\\N第二行" + result = _wrap_title_text(text, video_width=1080, font_size=48) + assert result == text + + def test_explicit_newline_each_segment_wraps_independently(self): + """\\N 分段后，每段各自自动换行。""" + # 480px 宽，48px 字号，可用 400px，每段约8个中文字 + text = "这是第一段很长很长很长的内容\\N这是第二段也很长很长的内容" + result = _wrap_title_text(text, video_width=480, font_size=48) + # 应该有多个 \N：用户手动的 + 自动换行的 + assert "\\N" in result + segments = result.split("\\N") + # 至少3行（两段都需要换行） + assert len(segments) >= 3 + # 验证包含两段的文字 + joined = result.replace("\\N", "") + assert "第一段" in joined + assert "第二段" in joined + + def test_multiple_explicit_newlines(self): + """多个 \\N 分段都应保留。""" + text = "A\\NB\\NC" + result = _wrap_title_text(text, video_width=1080, font_size=48) + assert result == text + assert result.count("\\N") == 2 + + def test_build_ass_content_integration(self): """集成测试：build_ass_content 中的标题应该自动换行。""" long_title = "这是一段非常长的标题文字用于测试自动换行功能是否正常工作" diff --git a/tests/unit/test_render_audio_pure.py b/tests/unit/test_render_audio_pure.py index bedd77eb9..336291a74 100755 --- a/tests/unit/test_render_audio_pure.py +++ b/tests/unit/test_render_audio_pure.py @@ -11,6 +11,7 @@ from video_processing.render_audio import ( RenderContext, clip_effective_duration, clip_has_audio, + mix_with_independent_audio, ) from video_processing.unified_render_service import ResolvedClip @@ -153,3 +154,93 @@ class TestRenderContext: """音频缓存初始为空.""" ctx = _make_ctx() assert ctx._audio_cache == {} + + +class TestMixWithIndependentAudioVolume: + """测试 mix_with_independent_audio 中主视频 clip 音量滤镜是否生效。""" + + def _make_clip(self, volume=None, clip_id="c1", duration=5.0): + """创建测试用 ResolvedClip。""" + config = {} + if volume is not None: + config["volume"] = volume + return ResolvedClip( + clip_id=clip_id, + asset_id=f"a_{clip_id}", + local_path=Path(f"/tmp/{clip_id}.mp4"), + clip_type="main", + order=0, + duration=duration, + actual_duration=duration, + config=config, + ) + + def test_main_clip_volume_zero_applied_in_filter(self): + """volume=0 的主视频 clip 应在 filter_complex 中包含 volume=0.0000。""" + from unittest.mock import patch + + ctx = _make_ctx() + main_clip = self._make_clip(volume=0, clip_id="m1") + captured_cmd = {} + + def fake_run_ffmpeg(cmd): + captured_cmd["cmd"] = cmd + + with patch("video_processing.render_audio.run_ffmpeg", side_effect=fake_run_ffmpeg): + mix_with_independent_audio( + ctx=ctx, + main_clips=[main_clip], + audio_clips=[], + output_path=Path("/tmp/out.m4a"), + video_duration=5.0, + ) + + filter_complex = captured_cmd["cmd"][captured_cmd["cmd"].index("-filter_complex") + 1] + assert "volume=0.0000" in filter_complex, f"volume filter missing in: {filter_complex}" + + def test_main_clip_default_volume_no_filter(self): + """默认 volume=1.0 时不应添加 volume 滤镜。""" + from unittest.mock import patch + + ctx = _make_ctx() + main_clip = self._make_clip(clip_id="m1") # no volume set + captured_cmd = {} + + def fake_run_ffmpeg(cmd): + captured_cmd["cmd"] = cmd + + with patch("video_processing.render_audio.run_ffmpeg", side_effect=fake_run_ffmpeg): + mix_with_independent_audio( + ctx=ctx, + main_clips=[main_clip], + audio_clips=[], + output_path=Path("/tmp/out.m4a"), + video_duration=5.0, + ) + + filter_complex = captured_cmd["cmd"][captured_cmd["cmd"].index("-filter_complex") + 1] + # 默认音量不应出现 volume= 滤镜 + assert "volume=" not in filter_complex, f"unexpected volume filter in: {filter_complex}" + + def test_main_clip_partial_volume_applied(self): + """volume=0.5 的主视频 clip 应包含 volume=0.5000。""" + from unittest.mock import patch + + ctx = _make_ctx() + main_clip = self._make_clip(volume=0.5, clip_id="m1") + captured_cmd = {} + + def fake_run_ffmpeg(cmd): + captured_cmd["cmd"] = cmd + + with patch("video_processing.render_audio.run_ffmpeg", side_effect=fake_run_ffmpeg): + mix_with_independent_audio( + ctx=ctx, + main_clips=[main_clip], + audio_clips=[], + output_path=Path("/tmp/out.m4a"), + video_duration=5.0, + ) + + filter_complex = captured_cmd["cmd"][captured_cmd["cmd"].index("-filter_complex") + 1] + assert "volume=0.5000" in filter_complex, f"volume filter missing in: {filter_complex}"