fix(ci): auto-fix lint/format issues

根因：API层水印配置存在 config.export.watermark_enabled/text（扁平格式），
但渲染引擎只从 config.watermark（嵌套格式）读取，两套格式不兼容导致
水印配置永远读不到，所有模式下水印都静默失效，单clip直通模式更隐蔽。

修复：
1. 新增 _resolve_watermark_config 静态方法，统一解析水印配置
2. 同时支持两种存储格式：
   - 嵌套格式（优先）：config.watermark = {enabled, mode, ...}
   - 扁平格式：config.export.watermark_enabled + watermark_text
3. _can_use_pass_through 和 _build_filter_complex 两处均使用统一方法
4. 补充 9 个单测覆盖两种格式及边界情况

apps/worker/video_processing/unified_render_service.py

@@ -843,6 +843,44 @@ class UnifiedRenderService:
             logger.warning("TTS 配音异常，跳过: %s", e)
             return False
 
+    @staticmethod
+    def _resolve_watermark_config(plan_config: dict[str, Any] | None) -> WatermarkConfig | None:
+        """从 plan config 中解析水印配置，兼容两种存储格式.
+
+        支持格式：
+        1. 嵌套格式：config.watermark = {enabled, mode, text, image_path, ...}
+        2. 扁平格式（导出配置）：config.export.watermark_enabled + config.export.watermark_text
+
+        Returns:
+            WatermarkConfig 或 None（未启用水印时）
+        """
+        if not plan_config or not isinstance(plan_config, dict):
+            return None
+
+        # 格式1: 嵌套 watermark 对象（优先）
+        wm_data = plan_config.get("watermark")
+        if isinstance(wm_data, dict) and wm_data:
+            config = WatermarkConfig.from_dict(wm_data)
+            if config is not None:
+                return config
+
+        # 格式2: 扁平 export.watermark_enabled + export.watermark_text
+        export_cfg = plan_config.get("export")
+        if isinstance(export_cfg, dict) and export_cfg:
+            enabled = export_cfg.get("watermark_enabled", False)
+            text = export_cfg.get("watermark_text", "") or ""
+            if enabled and text:
+                return WatermarkConfig(
+                    mode="text",
+                    text=str(text),
+                    position=export_cfg.get("watermark_position", "bottom_right"),
+                    opacity=float(export_cfg.get("watermark_opacity", 0.6)),
+                    font_size=int(export_cfg.get("watermark_font_size", 24)),
+                    font_color=str(export_cfg.get("watermark_font_color", "white")),
+                )
+
+        return None
+
     def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool:
         """判断是否可以走直通优化路径。
 
@@ -864,15 +902,11 @@ class UnifiedRenderService:
         if isinstance(plan_config, dict) and plan_config.get("stickers"):
             return False
 
-        # 有水印时禁用直通（图片水印需要额外输入，统一走 filter_complex）
-        try:
-            from video_processing.watermark_engine import WatermarkConfig
-
-            wm_config = WatermarkConfig.from_dict(plan_config.get("watermark"))
-            if wm_config is not None and wm_config.validate()[0]:
-                return False
-        except Exception:
-            pass
+        # 有水印时禁用直通（图片水印需要额外输入，统一走 filter_complex；
+        # 文字水印虽然可以 -vf 叠加，但为了保持路径统一也走 filter_complex）
+        wm_config = UnifiedRenderService._resolve_watermark_config(plan_config)
+        if wm_config is not None and wm_config.validate()[0]:
+            return False
 
         # 有调速时仍然可以走直通（视频调速通过 setpts 实现，单输入即可）
 
@@ -1117,11 +1151,11 @@ class UnifiedRenderService:
             filters.append(f"scale={pip_w}:{pip_h}")
         elif role == "background":
             # background: 铺满裁剪（作为底图，覆盖全屏）
-            filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase")
+            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"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=decrease")
             filters.append(f"pad={self.output_width}:{self.output_height}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")
 
         # 调色滤镜
@@ -1464,16 +1498,12 @@ class UnifiedRenderService:
                 pip_h = int(self.output_height * _PIP_SCALE)
                 filters.append(f"scale={pip_w}:{pip_h}")
             elif role == "background":
-                filters.append(
-                    f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase"
-                )
+                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: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
                 # concat 要求所有输入分辨率完全一致，pad 模式确保不同宽高比的素材都能正常拼接
-                filters.append(
-                    f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease"
-                )
+                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}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")
 
             # 调色滤镜（每个 clip 独立的 color grade 配置）
@@ -1561,9 +1591,7 @@ class UnifiedRenderService:
                 if role in layer_output_labels:
                     base_label = layer_output_labels[role]
                     combined_label = f"combined_{role}"
-                    filter_parts.append(
-                        f"[{final_video_label}][{base_label}]" f"overlay=(W-w)/2:(H-h)/2[{combined_label}]"
-                    )
+                    filter_parts.append(f"[{final_video_label}][{base_label}]overlay=(W-w)/2:(H-h)/2[{combined_label}]")
                     final_video_label = combined_label
         else:
             # 无 background 时，取 broll 或 main 作为基础
@@ -1587,11 +1615,11 @@ class UnifiedRenderService:
                     20,
                 )
                 combined_label = f"combined_{layer.role}"
-                filter_parts.append(f"[{final_video_label}][{overlay_label}]" f"overlay={x}:{y}[{combined_label}]")
+                filter_parts.append(f"[{final_video_label}][{overlay_label}]overlay={x}:{y}[{combined_label}]")
                 final_video_label = combined_label
 
         # 叠加水印（在字幕之前）
-        watermark_config = WatermarkConfig.from_dict((self.plan.config or {}).get("watermark"))
+        watermark_config = UnifiedRenderService._resolve_watermark_config(self.plan.config)
         if watermark_config is not None:
             wm_valid, wm_err = watermark_config.validate()
             if wm_valid:

tests/unit/test_watermark_intro_outro.py

@@ -28,12 +28,14 @@ class TestWatermarkConfig(unittest.TestCase):
 
     def test_from_dict_text_mode(self):
         """文字水印模式."""
-        cfg = WatermarkConfig.from_dict({
-            "enabled": True,
-            "mode": "text",
-            "text": "hello world",
-            "position": "top_left",
-        })
+        cfg = WatermarkConfig.from_dict(
+            {
+                "enabled": True,
+                "mode": "text",
+                "text": "hello world",
+                "position": "top_left",
+            }
+        )
         self.assertIsNotNone(cfg)
         self.assertEqual(cfg.mode, "text")
         self.assertEqual(cfg.text, "hello world")
@@ -41,18 +43,22 @@ class TestWatermarkConfig(unittest.TestCase):
 
     def test_from_dict_image_missing_path(self):
         """图片水印缺路径 → None."""
-        cfg = WatermarkConfig.from_dict({
-            "enabled": True,
-            "mode": "image",
-        })
+        cfg = WatermarkConfig.from_dict(
+            {
+                "enabled": True,
+                "mode": "image",
+            }
+        )
         self.assertIsNone(cfg)
 
     def test_from_dict_text_missing_text(self):
         """文字水印缺文字 → None."""
-        cfg = WatermarkConfig.from_dict({
-            "enabled": True,
-            "mode": "text",
-        })
+        cfg = WatermarkConfig.from_dict(
+            {
+                "enabled": True,
+                "mode": "text",
+            }
+        )
         self.assertIsNone(cfg)
 
     def test_validate_text_valid(self):
@@ -98,9 +104,7 @@ class TestWatermarkEnginePosition(unittest.TestCase):
         self.my = 20
 
     def test_top_left(self):
-        x, y = WatermarkEngine.calc_position(
-            "top_left", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
-        )
+        x, y = WatermarkEngine.calc_position("top_left", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my)
         self.assertEqual(x, 20)
         self.assertEqual(y, 20)
 
@@ -126,9 +130,7 @@ class TestWatermarkEnginePosition(unittest.TestCase):
         self.assertEqual(y, (1080 - 100) // 2)
 
     def test_center(self):
-        x, y = WatermarkEngine.calc_position(
-            "center", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
-        )
+        x, y = WatermarkEngine.calc_position("center", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my)
         self.assertEqual(x, (1920 - 200) // 2)
         self.assertEqual(y, (1080 - 100) // 2)
 
@@ -162,9 +164,7 @@ class TestWatermarkEnginePosition(unittest.TestCase):
 
     def test_default_fallback(self):
         """非法位置默认右下角."""
-        x, y = WatermarkEngine.calc_position(
-            "unknown", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
-        )
+        x, y = WatermarkEngine.calc_position("unknown", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my)
         self.assertEqual(x, 1920 - 200 - 20)
         self.assertEqual(y, 1080 - 100 - 20)
 
@@ -188,9 +188,7 @@ class TestWatermarkEngineFilters(unittest.TestCase):
             margin_x=10,
             margin_y=10,
         )
-        result = WatermarkEngine.build_text_watermark_filter(
-            "[in]", "[out]", cfg, 1920, 1080
-        )
+        result = WatermarkEngine.build_text_watermark_filter("[in]", "[out]", cfg, 1920, 1080)
         self.assertTrue(result.startswith("[in]drawtext="))
         self.assertIn("text='hello'", result)
         self.assertIn("fontsize=24", result)
@@ -206,9 +204,7 @@ class TestWatermarkEngineFilters(unittest.TestCase):
             scroll=True,
             scroll_speed=60,
         )
-        result = WatermarkEngine.build_text_watermark_filter(
-            "[in]", "[out]", cfg, 1920, 1080
-        )
+        result = WatermarkEngine.build_text_watermark_filter("[in]", "[out]", cfg, 1920, 1080)
         self.assertIn("mod(60*t", result)
 
 
@@ -224,16 +220,18 @@ class TestIntroOutroConfig(unittest.TestCase):
 
     def test_from_dict_intro_text(self):
         """文字片头配置."""
-        cfg = IntroOutroConfig.from_dict({
-            "enabled": True,
-            "intro": {
-                "type": "text",
-                "title": "欢迎观看",
-                "subtitle": "精彩内容马上开始",
-                "duration": 3.0,
-                "background": "#1a1a2e",
-            },
-        })
+        cfg = IntroOutroConfig.from_dict(
+            {
+                "enabled": True,
+                "intro": {
+                    "type": "text",
+                    "title": "欢迎观看",
+                    "subtitle": "精彩内容马上开始",
+                    "duration": 3.0,
+                    "background": "#1a1a2e",
+                },
+            }
+        )
         self.assertTrue(cfg.enabled)
         self.assertTrue(cfg.has_intro)
         self.assertFalse(cfg.has_outro)
@@ -243,14 +241,16 @@ class TestIntroOutroConfig(unittest.TestCase):
 
     def test_from_dict_outro_video(self):
         """视频片尾配置."""
-        cfg = IntroOutroConfig.from_dict({
-            "enabled": True,
-            "outro": {
-                "type": "video",
-                "video_path": "/tmp/outro.mp4",
-                "duration": 5.0,
-            },
-        })
+        cfg = IntroOutroConfig.from_dict(
+            {
+                "enabled": True,
+                "outro": {
+                    "type": "video",
+                    "video_path": "/tmp/outro.mp4",
+                    "duration": 5.0,
+                },
+            }
+        )
         self.assertTrue(cfg.has_outro)
         self.assertEqual(cfg.outro_type, "video")
         self.assertEqual(cfg.outro_video_path, "/tmp/outro.mp4")
@@ -314,9 +314,7 @@ class TestIntroOutroEngineConcat(unittest.TestCase):
             # 创建空文件模拟
             main_video.write_bytes(b"fake video data")
 
-            result = IntroOutroEngine.concat_with_intro_outro(
-                main_video, None, None, output
-            )
+            result = IntroOutroEngine.concat_with_intro_outro(main_video, None, None, output)
             self.assertTrue(result)
             self.assertTrue(output.exists())
             self.assertEqual(main_video.read_bytes(), output.read_bytes())
@@ -333,12 +331,91 @@ class TestIntroOutroEngineConcat(unittest.TestCase):
             # intro 路径不存在
             intro = Path(tmpdir) / "nonexistent.mp4"
 
-            result = IntroOutroEngine.concat_with_intro_outro(
-                main_video, intro, None, output
-            )
+            result = IntroOutroEngine.concat_with_intro_outro(main_video, intro, None, output)
             self.assertTrue(result)
             self.assertTrue(output.exists())
 
 
+class TestResolveWatermarkConfig(unittest.TestCase):
+    """UnifiedRenderService._resolve_watermark_config 兼容性测试."""
+
+    def _resolve(self, plan_config):
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        return UnifiedRenderService._resolve_watermark_config(plan_config)
+
+    def test_none_or_empty_config(self):
+        """空配置 → None."""
+        self.assertIsNone(self._resolve(None))
+        self.assertIsNone(self._resolve({}))
+        self.assertIsNone(self._resolve([]))  # 非dict安全处理
+
+    def test_nested_format_enabled(self):
+        """嵌套格式 config.watermark 正常解析."""
+        cfg = {"watermark": {"enabled": True, "mode": "text", "text": "测试水印"}}
+        result = self._resolve(cfg)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.mode, "text")
+        self.assertEqual(result.text, "测试水印")
+
+    def test_nested_format_disabled(self):
+        """嵌套格式未启用 → None."""
+        cfg = {"watermark": {"enabled": False, "mode": "text", "text": "测试"}}
+        self.assertIsNone(self._resolve(cfg))
+
+    def test_flat_export_format_enabled(self):
+        """扁平格式 config.export.watermark_enabled + text 正常解析."""
+        cfg = {"export": {"watermark_enabled": True, "watermark_text": "我的水印"}}
+        result = self._resolve(cfg)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.mode, "text")
+        self.assertEqual(result.text, "我的水印")
+        self.assertEqual(result.position, "bottom_right")
+
+    def test_flat_export_format_disabled(self):
+        """扁平格式未启用 → None."""
+        cfg = {"export": {"watermark_enabled": False, "watermark_text": "测试"}}
+        self.assertIsNone(self._resolve(cfg))
+
+    def test_flat_export_format_no_text(self):
+        """扁平格式启用但无文字 → None."""
+        cfg = {"export": {"watermark_enabled": True, "watermark_text": ""}}
+        self.assertIsNone(self._resolve(cfg))
+
+    def test_nested_takes_priority(self):
+        """嵌套格式存在时优先使用嵌套格式（忽略扁平格式）."""
+        cfg = {
+            "watermark": {"enabled": True, "mode": "text", "text": "嵌套水印"},
+            "export": {"watermark_enabled": True, "watermark_text": "扁平水印"},
+        }
+        result = self._resolve(cfg)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.text, "嵌套水印")
+
+    def test_flat_with_custom_position(self):
+        """扁平格式支持自定义位置、透明度等参数."""
+        cfg = {
+            "export": {
+                "watermark_enabled": True,
+                "watermark_text": "自定义水印",
+                "watermark_position": "top_left",
+                "watermark_opacity": 0.5,
+                "watermark_font_size": 32,
+                "watermark_font_color": "red",
+            }
+        }
+        result = self._resolve(cfg)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.position, "top_left")
+        self.assertAlmostEqual(result.opacity, 0.5)
+        self.assertEqual(result.font_size, 32)
+        self.assertEqual(result.font_color, "red")
+
+    def test_no_export_key(self):
+        """没有 export 字段时不报错."""
+        cfg = {"other": "value"}
+        self.assertIsNone(self._resolve(cfg))
+
+
 if __name__ == "__main__":
     unittest.main()