apps/worker/worker_app/tasks/generation.py

@@ -178,10 +178,136 @@ class _VirtualClip:
     config: dict[str, Any] = field(default_factory=dict)
 
 
+def _load_template_clip_configs(template_id: str) -> list:
+    """从数据库读取模板的片段配置列表。
+
+    失败返回空列表，不阻断主流程。
+    """
+    if not template_id:
+        return []
+    try:
+        from worker_app.db import SessionLocal
+
+        from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
+            SQLAlchemyTemplateClipConfigRepository,
+        )
+
+        session = SessionLocal()
+        try:
+            repo = SQLAlchemyTemplateClipConfigRepository(session)
+            configs = repo.list_by_template(template_id, limit=200)
+            logger.info("读取模板片段配置: template_id=%s count=%d", template_id, len(configs))
+            return configs
+        finally:
+            session.close()
+    except Exception as e:
+        logger.warning("读取模板片段配置失败，跳过效果层映射: template_id=%s error=%s", template_id, e)
+        return []
+
+
+def _extract_intro_outro_from_clip_configs(clip_configs: list) -> dict[str, Any]:
+    """从模板的 intro/outro 类型 clip_config 中提取 plan 级 intro_outro 配置。
+
+    UnifiedRenderService 已支持 plan.config.intro_outro 路径，
+    这里把 intro/outro 片段配置转为统一格式注入。
+    """
+    intro_configs = [
+        c for c in clip_configs if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) == "intro"
+    ]
+    outro_configs = [
+        c for c in clip_configs if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) == "outro"
+    ]
+
+    result: dict[str, Any] = {}
+
+    if intro_configs:
+        intro = intro_configs[0]
+        intro_cfg = intro.config or {}
+        result["has_intro"] = True
+        result["intro_type"] = intro_cfg.get("intro_type", "text")
+        result["intro_duration"] = intro.default_duration or 3.0
+        if intro.text_template:
+            result["intro_text"] = intro.text_template
+        # 透传额外配置
+        for key in ("intro_text_color", "intro_bg_color", "intro_font_size", "intro_video_url", "intro_video_path"):
+            if key in intro_cfg:
+                result[key] = intro_cfg[key]
+
+    if outro_configs:
+        outro = outro_configs[0]
+        outro_cfg = outro.config or {}
+        result["has_outro"] = True
+        result["outro_type"] = outro_cfg.get("outro_type", "text")
+        result["outro_duration"] = outro.default_duration or 3.0
+        if outro.text_template:
+            result["outro_text"] = outro.text_template
+        for key in ("outro_text_color", "outro_bg_color", "outro_font_size", "outro_follow_text"):
+            if key in outro_cfg:
+                result[key] = outro_cfg[key]
+
+    return result
+
+
+def _apply_template_clip_effects(
+    clips: list[_VirtualClip],
+    clip_configs: list,
+    mode: str,
+) -> None:
+    """将模板的 clip 级效果层映射到素材 clips 上（就地修改）。
+
+    映射规则：
+    - 只对素材主体 clips 做映射（ONE_TAKE: main, PIP: main+overlay, VOICE_OVER: main, VOICE_PIP: background+b_roll）
+    - 从模板中筛选 main 类型的 clip_config 作为效果模板
+    - 素材 clips 按顺序循环匹配模板 clip_config（素材多的话重复使用最后一个模板配置）
+    - 映射字段：transition_effect, config.color_grade, config.speed
+    """
+    if not clip_configs or not clips:
+        return
+
+    # 筛选 main 类型的模板配置（作为效果模板池）
+    main_configs = [
+        c
+        for c in clip_configs
+        if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) in ("main", "showcase", "b_roll")
+    ]
+    if not main_configs:
+        return
+
+    # 确定需要映射的素材 clips（排除 corner_voice 等特殊层）
+    target_clips = [c for c in clips if c.clip_type not in ("corner_voice",)]
+
+    for i, clip in enumerate(target_clips):
+        # 循环匹配：素材多了用最后一个模板配置
+        cfg_idx = min(i, len(main_configs) - 1)
+        template_cfg = main_configs[cfg_idx]
+
+        # 1. 转场效果
+        transition = (
+            template_cfg.transition_effect.value
+            if hasattr(template_cfg.transition_effect, "value")
+            else template_cfg.transition_effect
+        )
+        if transition and transition != "cut":
+            clip.transition_effect = transition
+
+        # 2. clip 级效果配置（滤镜、调速等）
+        template_clip_config = template_cfg.config or {}
+        if template_clip_config:
+            # 合并到 clip.config（保留已有配置如 role 等）
+            existing_config = clip.config or {}
+            # 需要从模板复制的效果层 key
+            effect_keys = ("color_grade", "speed", "playback_speed", "reverse", "chroma_key", "filter")
+            for key in effect_keys:
+                if key in template_clip_config:
+                    existing_config[key] = template_clip_config[key]
+            clip.config = existing_config
+
+
 def _build_plan_and_clips_from_task(
     task_id: str,
     downloaded_paths: list[Path],
     mode: str,
+    template_id: str = "",
 ) -> tuple[_VirtualPlan, list[_VirtualClip], dict[str, Path]]:
     """根据模式和下载的素材路径，构建虚拟 plan + clips + asset_path_map。
 
@@ -269,6 +395,25 @@ def _build_plan_and_clips_from_task(
                 )
             )
 
+    # ── P1: 模板效果层映射 ──
+    if template_id:
+        clip_configs = _load_template_clip_configs(template_id)
+        if clip_configs:
+            # 1. clip级效果层（转场、滤镜、调速等）
+            _apply_template_clip_effects(clips, clip_configs, mode)
+
+            # 2. 片头片尾（从 intro/outro 类型 clip 提取 plan 级配置）
+            intro_outro_config = _extract_intro_outro_from_clip_configs(clip_configs)
+            if intro_outro_config:
+                plan_config = plan.config or {}
+                plan_config["intro_outro"] = intro_outro_config
+                plan.config = plan_config
+                logger.info(
+                    "模板片头片尾配置已注入: has_intro=%s has_outro=%s",
+                    intro_outro_config.get("has_intro", False),
+                    intro_outro_config.get("has_outro", False),
+                )
+
     return plan, clips, asset_path_map
 
 
@@ -1030,6 +1175,7 @@ def _render_video(
         task_id=task_id,
         downloaded_paths=downloaded_videos,
         mode=editing_mode.value,
+        template_id=template_id,
     )
 
     total_duration = sum(c.duration for c in virtual_clips)

tests/unit/test_oneclick_gen_p0_fixes.py

@@ -264,3 +264,147 @@ class TestP1Validations:
         with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
             with pytest.raises(ValueError, match="模板不存在"):
                 _validate_template_exists("tmpl_nonexistent")
+
+
+# ── P1: 一键生成 clip 级效果层映射 ───────────────────────────────────────────
+
+
+class TestTemplateClipEffectMapping:
+    """P1: 模板 clip 级效果层映射到一键生成素材 clips."""
+
+    def _make_virtual_clip(self, idx: int, clip_type: str = "main", config: dict | None = None):
+        from dataclasses import dataclass, field
+
+        _clip_type_val = clip_type
+
+        @dataclass
+        class FakeClip:
+            id: str = f"vc_{idx:03d}"
+            plan_id: str = "task_001"
+            clip_type: str = _clip_type_val
+            order: int = idx
+            asset_id: str = f"asset_{idx}"
+            duration: float = 5.0
+            transition_effect: str = "cut"
+            config: dict = field(default_factory=dict)
+
+        return FakeClip(config=config or {})
+
+    def _make_template_clip_config(self, clip_type: str = "main", transition: str = "cut", config: dict | None = None):
+        mock = MagicMock()
+        mock.clip_type = clip_type
+        mock.transition_effect = transition
+        mock.config = config or {}
+        mock.default_duration = 3.0
+        mock.text_template = ""
+        return mock
+
+    def test_transition_effect_mapped(self):
+        """转场效果正确映射到素材 clips."""
+        from worker_app.tasks.generation import _apply_template_clip_effects
+
+        clips = [self._make_virtual_clip(i) for i in range(3)]
+        clip_configs = [
+            self._make_template_clip_config("main", transition="fade"),
+            self._make_template_clip_config("main", transition="dissolve"),
+        ]
+
+        _apply_template_clip_effects(clips, clip_configs, "one_take")
+
+        # 前两个按顺序映射，第三个用最后一个模板配置
+        assert clips[0].transition_effect == "fade"
+        assert clips[1].transition_effect == "dissolve"
+        assert clips[2].transition_effect == "dissolve"  # 复用最后一个
+
+    def test_color_grade_mapped(self):
+        """滤镜配置正确映射到 clip.config.color_grade."""
+        from worker_app.tasks.generation import _apply_template_clip_effects
+
+        clips = [self._make_virtual_clip(i) for i in range(2)]
+        clip_configs = [
+            self._make_template_clip_config(
+                "main", config={"color_grade": {"enabled": True, "filter": "vintage", "brightness": 0.1}}
+            ),
+        ]
+
+        _apply_template_clip_effects(clips, clip_configs, "one_take")
+
+        assert clips[0].config["color_grade"]["filter"] == "vintage"
+        assert clips[0].config["color_grade"]["brightness"] == 0.1
+        # 第二个素材复用第一个模板配置
+        assert clips[1].config["color_grade"]["filter"] == "vintage"
+
+    def test_existing_config_preserved(self):
+        """已有 clip.config 内容（如 role）被保留."""
+        from worker_app.tasks.generation import _apply_template_clip_effects
+
+        clips = [self._make_virtual_clip(0, config={"role": "b_roll"})]
+        clip_configs = [
+            self._make_template_clip_config("main", config={"color_grade": {"enabled": True, "filter": "warm"}}),
+        ]
+
+        _apply_template_clip_effects(clips, clip_configs, "voice_over")
+
+        assert clips[0].config["role"] == "b_roll"  # 保留原有配置
+        assert clips[0].config["color_grade"]["filter"] == "warm"  # 新增滤镜配置
+
+    def test_empty_clip_configs_no_change(self):
+        """空模板配置时 clips 保持不变."""
+        from worker_app.tasks.generation import _apply_template_clip_effects
+
+        clips = [self._make_virtual_clip(i) for i in range(2)]
+        _apply_template_clip_effects(clips, [], "one_take")
+
+        assert clips[0].transition_effect == "cut"
+        assert clips[1].transition_effect == "cut"
+
+    def test_cut_transition_not_overwritten(self):
+        """模板转场为 cut 时不覆盖（保持默认）."""
+        from worker_app.tasks.generation import _apply_template_clip_effects
+
+        clips = [self._make_virtual_clip(0)]
+        clips[0].transition_effect = "fade"  # 已有非默认值
+        clip_configs = [
+            self._make_template_clip_config("main", transition="cut"),
+        ]
+
+        _apply_template_clip_effects(clips, clip_configs, "one_take")
+
+        # 模板是 cut 时，保留原有值（避免无意义覆盖）
+        assert clips[0].transition_effect == "fade"
+
+    def test_intro_outro_extracted(self):
+        """intro/outro 类型 clip_config 正确提取为 plan 级 intro_outro 配置."""
+        from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs
+
+        clip_configs = [
+            self._make_template_clip_config("intro", config={"intro_type": "text", "intro_text_color": "#ffffff"}),
+            self._make_template_clip_config("main"),
+            self._make_template_clip_config("outro", config={"outro_type": "follow", "outro_follow_text": "关注我们"}),
+        ]
+        # 设置 intro/outro 的 text_template
+        clip_configs[0].text_template = "精彩视频"
+        clip_configs[0].default_duration = 2.5
+
+        result = _extract_intro_outro_from_clip_configs(clip_configs)
+
+        assert result["has_intro"] is True
+        assert result["intro_type"] == "text"
+        assert result["intro_text"] == "精彩视频"
+        assert result["intro_duration"] == 2.5
+        assert result["intro_text_color"] == "#ffffff"
+        assert result["has_outro"] is True
+        assert result["outro_type"] == "follow"
+        assert result["outro_follow_text"] == "关注我们"
+
+    def test_intro_outro_empty_when_none(self):
+        """没有 intro/outro 时返回空 dict."""
+        from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs
+
+        clip_configs = [
+            self._make_template_clip_config("main"),
+            self._make_template_clip_config("main"),
+        ]
+
+        result = _extract_intro_outro_from_clip_configs(clip_configs)
+        assert result == {}