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}]")

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(

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 = "这是一段非常长的标题文字用于测试自动换行功能是否正常工作"

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}"