fix: 模板配置字段非dict类型崩溃防护 (bool/str/int) (#1384)

apps/worker/video_processing/unified_render_service.py

@@ -558,7 +558,11 @@ class UnifiedRenderService:
         """
         config = self.plan.config or {}
         title_cfg = config.get("title", {}) or {}
+        if not isinstance(title_cfg, dict):
+            title_cfg = {}
         subtitle_cfg = config.get("subtitle", {}) or {}
+        if not isinstance(subtitle_cfg, dict):
+            subtitle_cfg = {}
 
         title_enabled = title_cfg.get("enabled", True)
         subtitle_enabled = subtitle_cfg.get("enabled", True)
@@ -789,6 +793,8 @@ class UnifiedRenderService:
         config = self.plan.config or {}
         tts_cfg = config.get("tts", {}) or {}
         subtitle_cfg = config.get("subtitle", {}) or {}
+        if not isinstance(subtitle_cfg, dict):
+            subtitle_cfg = {}
         use_subtitle_align = False  # 是否使用字幕对齐模式
 
         # 兼容前端顶层字段：voice_id / custom_text / voice_clone_profile_id

apps/worker/worker_app/tasks/generation.py

@@ -984,9 +984,9 @@ def _load_template_plan_config(template_id: str) -> dict:
 
             # 从独立字段组装成 plan.config 格式
             plan_config: dict[str, Any] = {}
-            title_cfg = template.title_config or {}
-            subtitle_cfg = template.subtitle_config or {}
-            bgm_cfg = template.bgm_config or {}
+            title_cfg = template.title_config if isinstance(template.title_config, dict) else {}
+            subtitle_cfg = template.subtitle_config if isinstance(template.subtitle_config, dict) else {}
+            bgm_cfg = template.bgm_config if isinstance(template.bgm_config, dict) else {}
 
             if title_cfg:
                 plan_config["title"] = title_cfg
@@ -1177,7 +1177,8 @@ def _render_video(
     # 2. 纯文本格式（旧）：直接作为标题文本使用
     if custom_title and custom_title.strip():
         plan_cfg = dict(virtual_plan.config or {})
-        title_cfg = dict(plan_cfg.get("title", {}) or {})
+        _raw_title = plan_cfg.get("title", {}) or {}
+        title_cfg = dict(_raw_title) if isinstance(_raw_title, dict) else {}
         ct_stripped = custom_title.strip()
         parsed_config = None
         if ct_stripped.startswith("{"):
@@ -1245,6 +1246,8 @@ def _render_video(
         plan_cfg = dict(virtual_plan.config or {})
         plan_cfg["voice_id"] = voice_ids[0]
         subtitle_cfg = plan_cfg.get("subtitle", {}) or {}
+        if not isinstance(subtitle_cfg, dict):
+            subtitle_cfg = {}
         subtitle_cfg["auto_generated"] = True
         subtitle_cfg["enabled"] = True  # 确保 ASR 字幕路径被触发，标题叠加也依赖此路径
         plan_cfg["subtitle"] = subtitle_cfg

tests/unit/test_bool_config_defense.py

@@ -0,0 +1,130 @@
+"""
+测试：模板 config 字段存储了非 dict 值（如 True / False / str）时，
+渲染链路不会崩溃（'bool' object has no attribute 'get'）。
+
+覆盖两个关键文件：
+1. generation.py — _load_template_plan_config 旧系统路径
+2. unified_render_service.py — _maybe_generate_ass
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Add worker app to path
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+
+class TestLoadTemplatePlanConfigBoolDefense:
+    """_load_template_plan_config 旧系统路径对非 dict 值的防护。"""
+
+    def _call_old_path(self, title_cfg, subtitle_cfg, bgm_cfg):
+        """通过 mock 新模板系统返回 None，强制走旧模板系统 fallback 路径。"""
+        from worker_app.tasks.generation import _load_template_plan_config
+
+        mock_old_template = MagicMock()
+        mock_old_template.title_config = title_cfg
+        mock_old_template.subtitle_config = subtitle_cfg
+        mock_old_template.bgm_config = bgm_cfg
+
+        mock_session = MagicMock()
+        # 旧系统 query 返回 mock template
+        mock_session.query.return_value.filter.return_value.first.return_value = mock_old_template
+
+        # Mock 新模板系统 repo.get() 返回 None（强制走 fallback）
+        mock_repo_cls = MagicMock()
+        mock_repo_cls.return_value.get.return_value = None
+
+        with (
+            patch("worker_app.tasks.generation.SessionLocal", return_value=mock_session),
+            patch("packages.adapters.sqlalchemy_impl.SQLAlchemyEditTemplateRepository", mock_repo_cls),
+            patch("packages.adapters.sqlalchemy_impl.SQLAlchemyTemplateClipConfigRepository", MagicMock()),
+        ):
+            return _load_template_plan_config("fake-id")
+
+    def test_bool_values_return_empty(self):
+        """title_config=True / subtitle_config=False / bgm_config='str' → 全部过滤掉"""
+        result = self._call_old_path(True, False, "not_a_dict")
+        assert isinstance(result, dict)
+        assert "title" not in result
+        assert "subtitle" not in result
+        assert "bgm" not in result
+
+    def test_valid_dict_passes_through(self):
+        """正常 dict 正常传递"""
+        result = self._call_old_path(
+            {"text": "标题", "enabled": True},
+            {"text": "副标题"},
+            {"enabled": True, "source": "test.mp3"},
+        )
+        assert result["title"] == {"text": "标题", "enabled": True}
+        assert result["subtitle"] == {"text": "副标题"}
+        assert result["bgm"] == {"enabled": True, "source": "test.mp3"}
+
+    def test_none_returns_empty(self):
+        """None → 空 dict"""
+        result = self._call_old_path(None, None, None)
+        assert result == {}
+
+    def test_mixed_valid_and_invalid(self):
+        """部分有效、部分无效时只保留有效的"""
+        result = self._call_old_path({"text": "OK"}, True, None)
+        assert "title" in result
+        assert "subtitle" not in result
+        assert "bgm" not in result
+
+    def test_int_and_list_also_filtered(self):
+        """int / list 类型也被过滤"""
+        result = self._call_old_path(42, [1, 2, 3], 0)
+        assert result == {}
+
+
+class TestUnifiedRenderBoolConfigDefense:
+    """_maybe_generate_ass 对 plan.config 中非 dict title/subtitle 的防护。"""
+
+    def _make_service(self, config):
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        service = UnifiedRenderService.__new__(UnifiedRenderService)
+        mock_plan = MagicMock()
+        mock_plan.config = config
+        service.plan = mock_plan
+        service.task_id = "test-task"
+        return service
+
+    def test_bool_title_does_not_crash(self):
+        """config['title']=True → 不崩溃，返回 None"""
+        service = self._make_service({"title": True, "subtitle": {}})
+        result = service._maybe_generate_ass(video_duration=10.0)
+        assert result is None
+
+    def test_bool_subtitle_does_not_crash(self):
+        """config['subtitle']=False → 不崩溃，返回 None"""
+        service = self._make_service({"title": {}, "subtitle": False})
+        result = service._maybe_generate_ass(video_duration=10.0)
+        assert result is None
+
+    def test_str_title_does_not_crash(self):
+        """config['title']='plain string' → 不崩溃"""
+        service = self._make_service({"title": "plain string", "subtitle": {}})
+        result = service._maybe_generate_ass(video_duration=10.0)
+        assert result is None
+
+    def test_none_config_does_not_crash(self):
+        """config=None → 不崩溃"""
+        service = self._make_service(None)
+        result = service._maybe_generate_ass(video_duration=10.0)
+        assert result is None
+
+    def test_int_title_does_not_crash(self):
+        """config['title']=42 → 不崩溃"""
+        service = self._make_service({"title": 42, "subtitle": 0})
+        result = service._maybe_generate_ass(video_duration=10.0)
+        assert result is None