apps/api/app/services/video_compose_service.py

@@ -422,7 +422,11 @@ class VideoComposeService:
         # 2. pad: 居中+留黑边到目标分辨率（保持原始比例，不裁剪内容）
         filters.append(f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black")
 
-        # 3. fps: 统一帧率（concat 要求所有输入帧率一致）
+        # 3. format: 统一像素格式为 yuv420p（H.264 标准格式，concat 要求所有输入像素格式一致）
+        # 不同素材可能是 yuv420p / yuv422p / yuv444p / nv12 等，必须统一
+        filters.append("format=yuv420p")
+
+        # 4. fps: 统一帧率（concat 要求所有输入帧率一致）
         # 放在 pad 之后、setpts 之前，确保分辨率和帧率都已统一
         if fps and fps > 0:
             filters.append(f"fps={fps}")
@@ -623,13 +627,22 @@ def _build_xfade_filter(
     # 总时长需要减去转场重叠部分
     total_duration -= transition_duration * (n - 1)
 
-    # 音频 crossfade（简化处理：使用 adelay + amix）
-    audio_labels = [c.audio_label for c in clip_chains if c.audio_label]
-    if len(audio_labels) >= 2:
-        # 简单拼接音频（不做 crossfade）
-        audio_inputs = "".join(f"[{label}]" for label in audio_labels)
-        parts.append(f"{audio_inputs}concat=n={len(audio_labels)}:v=0:a=1[outa]")
-    elif len(audio_labels) == 1:
-        parts.append(f"[{audio_labels[0]}]acopy[outa]")
+    # 音频：先 aformat 归一化再 concat（不同采样率/声道/采样格式会导致concat失败）
+    audio_chains_with_label = [(c, c.audio_label) for c in clip_chains if c.audio_label]
+    if len(audio_chains_with_label) >= 2:
+        normalized_audio_labels: list[str] = []
+        for chain, _ in audio_chains_with_label:
+            norm_label = f"anorm_{chain.video_label}"
+            audio_filters = [
+                "aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp",
+                f"atrim=0:{chain.duration}",
+                "asetpts=PTS-STARTPTS",
+            ]
+            parts.append(f"[{chain.audio_label}]{','.join(audio_filters)}[{norm_label}]")
+            normalized_audio_labels.append(norm_label)
+        audio_inputs = "".join(f"[{label}]" for label in normalized_audio_labels)
+        parts.append(f"{audio_inputs}concat=n={len(normalized_audio_labels)}:v=0:a=1[outa]")
+    elif len(audio_chains_with_label) == 1:
+        parts.append(f"[{audio_chains_with_label[0][0].audio_label}]acopy[outa]")
 
     return ";".join(parts), max(0.0, total_duration)

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -361,14 +361,37 @@ def _render_with_legacy(
         fps=fps,
     )
 
-    logger.info("执行 FFmpeg (legacy): plan_id=%s", plan_id)
+    logger.info("执行 FFmpeg (legacy): plan_id=%s cmd=%s", plan_id, " ".join(compose_cmd.command)[:500])
     try:
         from video_processing.ffmpeg_utils import run_ffmpeg
 
         run_ffmpeg(compose_cmd.command, timeout=3600)
     except Exception as e:
-        error_msg = f"FFmpeg 执行失败: {str(e)[:500]}"
-        logger.error("FFmpeg 执行失败(legacy): %s — %s", plan_id, error_msg)
+        # 提取完整 stderr（如果是 CalledProcessError）
+        stderr_text = ""
+        if hasattr(e, "stderr"):
+            stderr_raw = e.stderr
+            if isinstance(stderr_raw, bytes):
+                stderr_text = stderr_raw.decode("utf-8", errors="replace")
+            elif isinstance(stderr_raw, str):
+                stderr_text = stderr_raw
+
+        # 完整命令（截断前2000字符，避免日志过大）
+        full_cmd = " ".join(compose_cmd.command)
+        cmd_preview = full_cmd[:2000] + ("..." if len(full_cmd) > 2000 else "")
+
+        # 拼接完整错误信息：命令 + 异常 + stderr最后1500字符
+        error_parts = [f"FFmpeg渲染失败(exit={getattr(e, 'returncode', 'unknown')})"]
+        error_parts.append("--- cmd ---")
+        error_parts.append(cmd_preview)
+        if stderr_text:
+            # 取最后1500字符，通常错误信息在末尾
+            stderr_preview = stderr_text[-1500:] if len(stderr_text) > 1500 else stderr_text
+            error_parts.append("--- stderr (last 1500 chars) ---")
+            error_parts.append(stderr_preview)
+        error_msg = "\n".join(error_parts)
+
+        logger.error("FFmpeg 执行失败(legacy): plan_id=%s\n%s", plan_id, error_msg)
         _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg)
         return {"status": "error", "message": error_msg}
 

tests/unit/test_video_compose_service.py

@@ -359,6 +359,20 @@ class TestBuildComposeCommand(TestCase):
         self.assertIn("black", filter_text)
         self.assertIn("trim=", filter_text)
 
+    def test_filter_chain_contains_format_yuv420p(self):
+        """滤镜链包含 format=yuv420p，统一像素格式避免 concat 失败。"""
+        plan = _StubPlan()
+        clips = [_make_ready_clip(plan_id=plan.id)]
+        svc = _make_service(plan, clips)
+        cmd = svc.build_compose_command(plan.id, "/tmp/out.mp4")
+
+        chain = cmd.clip_chains[0]
+        filter_text = ",".join(chain.filters)
+        # format=yuv420p 必须在 pad 之后、fps 之后（像素格式统一放在分辨率之后）
+        pad_idx = filter_text.index("pad=")
+        fmt_idx = filter_text.index("format=yuv420p")
+        self.assertGreater(fmt_idx, pad_idx, "format 应该在 pad 之后")
+
     def test_filter_chain_contains_fps(self):
         """滤镜链包含 fps 滤镜，用于统一帧率避免 concat 失败。"""
         plan = _StubPlan()
@@ -617,6 +631,32 @@ class TestBuildXfadeFilter(TestCase):
         # 总时长 = 15 - 0.5*2 = 14.0
         self.assertAlmostEqual(duration, 14.0, places=2)
 
+    def test_xfade_with_audio_aformat_normalization(self):
+        """xfade 路径下多片段音频 concat 前必须经过 aformat 归一化。"""
+        from app.services.video_compose_service import ClipFilterChain
+
+        chains = [
+            ClipFilterChain(clip_id="c1", input_index=0, video_label="v0", audio_label="a0", filters=[], duration=5.0),
+            ClipFilterChain(clip_id="c2", input_index=1, video_label="v1", audio_label="a1", filters=[], duration=8.0),
+        ]
+        filter_str, _ = _build_xfade_filter(chains, transition_duration=0.5, transitions=["fade"])
+        # 两个音频流都必须经过 aformat 归一化（48000Hz + stereo + fltp）
+        aformat_count = filter_str.count("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")
+        self.assertEqual(aformat_count, 2, "两个音频片段都应该有aformat归一化")
+        # 最终音频 concat
+        self.assertIn("concat=n=2:v=0:a=1[outa]", filter_str)
+
+    def test_xfade_single_audio_passthrough(self):
+        """xfade 路径下只有一个音频片段时直接 acopy。"""
+        from app.services.video_compose_service import ClipFilterChain
+
+        chains = [
+            ClipFilterChain(clip_id="c1", input_index=0, video_label="v0", audio_label="a0", filters=[], duration=5.0),
+            ClipFilterChain(clip_id="c2", input_index=1, video_label="v1", audio_label=None, filters=[], duration=8.0),
+        ]
+        filter_str, _ = _build_xfade_filter(chains, transition_duration=0.5, transitions=["fade"])
+        self.assertIn("[a0]acopy[outa]", filter_str)
+
 
 class TestHasAudioTitleSubtitleFix(TestCase):
     """P0 修复验证：title/subtitle 片段不应有音频流。"""