chore: 清理剪辑模板编辑器生成视频死代码 + 重命名为剪辑模板 (#1472)

前端：
- 删除 GenerationProgressModal/GenerationHistoryModal 组件及目录
- 删除 generateEditPlan/getGenerationStatus/getEditPlanGenerations 三个死API函数
- 删除对应类型 GenerateResponse/EditPlanGeneration/GenerationStatusResponse
- 清理测试文件中的 mock 和引用
- 用户可见文案 '剪辑编辑器' 统一改为 '剪辑模板'
- 不改变量名/组件名/路由路径/API路径

后端：
- 删除 templates_editor/generation.py 死接口文件
- 清理 _fallback.py、dependencies.py、schemas.py 中的死代码
- 删除对应无用测试文件

apps/api/app/api/routes/generation_cover.py

@@ -63,7 +63,15 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
-def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> str:
+def _persist_cover_frame(
+    frame_url: str,
+    plan_id: str,
+    title_text: str = "",
+    *,
+    title_color: str = "#ffffff",
+    title_position: str = "bottom",
+    title_font_size: int | None = None,
+) -> str:
     """下载 MediaKit 返回的临时帧图，可选叠加标题后转存到 OSS covers/ 路径。
 
     Args:
@@ -71,6 +79,9 @@ def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") ->
         plan_id: 剪辑计划 ID（生成 OSS key）
         title_text: 非空时用 Pillow 在帧上叠加标题（用于 E2 从源素材抽帧，
             因为源素材本身没有烧录标题）
+        title_color: 标题字体颜色（#RRGGBB）
+        title_position: 标题位置 top/center/bottom
+        title_font_size: 标题字号，None 时自动计算
     """
     import tempfile
     import uuid
@@ -94,7 +105,13 @@ def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") ->
             try:
                 from packages.shared.title_overlay import apply_title_to_image
 
-                applied = apply_title_to_image(tmp_path, title_text)
+                applied = apply_title_to_image(
+                    tmp_path,
+                    title_text,
+                    color=title_color,
+                    position=title_position,
+                    font_size=title_font_size,
+                )
                 if applied:
                     logger.info("[封面生成] E2 帧图已叠加标题: plan_id=%s", plan_id)
             except Exception:
@@ -418,11 +435,15 @@ def generate_cover(
             asset_repo = SQLAlchemyAssetRepository(db)
             storage_svc = get_shared_storage_service()
             mk_client = get_mediakit_client()
-            # 从 plan.config 读取标题，E2 从源素材抽帧时叠加（源素材本身无标题）
+            # 从 plan.config 读取完整标题样式，E2 从源素材抽帧时叠加（源素材本身无标题）
             _e2_title_cfg = (plan.config or {}).get("title", {}) or {}
             if not isinstance(_e2_title_cfg, dict):
                 _e2_title_cfg = {}
             _e2_title_text = (_e2_title_cfg.get("text", "") or "").strip() if _e2_title_cfg.get("enabled", True) else ""
+            # 读取标题样式：前端可能传 color 或 font_color，都兼容
+            _e2_title_color = _e2_title_cfg.get("color") or _e2_title_cfg.get("font_color") or "#ffffff"
+            _e2_title_position = _e2_title_cfg.get("position", "bottom") or "bottom"
+            _e2_title_font_size = _e2_title_cfg.get("font_size") or _e2_title_cfg.get("size")
             if mk_client.is_available:
                 for aid in body.asset_ids:
                     try:
@@ -452,7 +473,14 @@ def generate_cover(
                         if snapshots:
                             raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
                             if raw:
-                                cover_url_from_task = _persist_cover_frame(raw, plan_id, title_text=_e2_title_text)
+                                cover_url_from_task = _persist_cover_frame(
+                                    raw,
+                                    plan_id,
+                                    title_text=_e2_title_text,
+                                    title_color=_e2_title_color,
+                                    title_position=_e2_title_position,
+                                    title_font_size=_e2_title_font_size,
+                                )
                                 logger.info(
                                     "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s",
                                     plan_id,

apps/api/app/api/routes/templates_editor/__init__.py

@@ -1,10 +1,9 @@
 """模板编辑器 API 路由包.
 
-将原来 2560 行的 templates_editor.py 巨无霸拆分为 12 个模块：
+模块拆分：
 - schemas.py:        所有 Pydantic model
 - dependencies.py:   依赖注入
 - _utils.py:         工具函数
-- _fallback.py:      自动兜底逻辑
 - draft.py:          草稿管理（详情/更新/发布/版本/回滚）
 - clips.py:          片段管理（CRUD/分割/合并/重排/批量删除/从素材创建）
 - adjustments.py:    片段调整（速度/音量/裁剪/批量调速）
@@ -13,7 +12,6 @@
 - export.py:         导出配置
 - subtitles.py:      字幕管理
 - ai_features.py:    AI 推荐
-- generation.py:     生成（触发/进度/记录）
 - timeline.py:       时间线
 
 挂载路径: /api/v1/templates/{template_id}/editor/
@@ -34,7 +32,6 @@ from .dependencies import get_draft_plan_id, get_editor_services  # noqa: F401
 from .draft import router as draft_router
 from .effects import router as effects_router
 from .export import router as export_router
-from .generation import router as generation_router
 from .subtitles import router as subtitles_router
 from .timeline import router as timeline_router
 
@@ -51,7 +48,6 @@ _sub_routers = [
     export_router,
     subtitles_router,
     ai_features_router,
-    generation_router,
     timeline_router,
 ]
 

apps/api/app/api/routes/templates_editor/_fallback.py

@@ -1,227 +0,0 @@
-"""模板编辑器自动兜底逻辑.
-
-generate_editor_draft 触发生成前的自动修复流程：
-1. draft → editing 状态迁移
-2. 无片段时从模板复制片段配置
-3. 为无素材片段分配指定素材
-4. 项目有素材库时自动选素材
-"""
-
-from __future__ import annotations
-
-import logging
-import random
-from typing import Any
-
-from app.services.edit_plan_service import EditPlanService
-from sqlalchemy.orm import Session
-
-from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
-    SQLAlchemyTemplateClipConfigRepository,
-)
-from packages.adapters.sqlalchemy_impl.template_repository import (
-    SQLAlchemyTemplateRepository,
-)
-from packages.domain.edit_plan import EditPlanStatus
-
-logger = logging.getLogger(__name__)
-
-
-def _auto_fallback_draft_to_editing(svc: EditPlanService, plan_id: str, plan_check) -> None:
-    """自动兜底 1: draft → editing"""
-    if plan_check.status == EditPlanStatus.DRAFT:
-        logger.info("模板编辑器自动兜底: plan=%s draft→editing", plan_id)
-        svc.transition_status(plan_id, EditPlanStatus.EDITING)
-
-
-def _auto_fallback_copy_template_clips(svc: EditPlanService, plan_id: str, plan_check, db: Session) -> None:
-    """自动兜底 2: 无片段 + 有 template_id → 从模板复制片段配置"""
-    existing_clips = svc.count_clips(plan_id)
-    if existing_clips == 0 and plan_check.template_id:
-        logger.info(
-            "模板编辑器自动兜底: plan=%s 无片段，从模板 %s 复制片段配置",
-            plan_id,
-            plan_check.template_id,
-        )
-        clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db)
-        configs = clip_config_repo.list_by_template(plan_check.template_id)
-        if configs:
-            for cfg in configs:
-                svc.create_clip(
-                    plan_id=plan_id,
-                    clip_type=cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
-                    order=cfg.order,
-                    template_clip_config_id=cfg.id,
-                    duration=cfg.default_duration,
-                    transition_effect=(
-                        cfg.transition_effect.value
-                        if hasattr(cfg.transition_effect, "value")
-                        else cfg.transition_effect
-                    ),
-                )
-            logger.info(
-                "模板编辑器自动兜底: plan=%s 从 template_clip_configs 复制了 %d 个片段",
-                plan_id,
-                len(configs),
-            )
-        else:
-            tpl_repo = SQLAlchemyTemplateRepository(db)
-            segments = tpl_repo.list_segments(plan_check.template_id)
-            for seg in segments:
-                avg_duration = (seg.duration_min + seg.duration_max) / 2
-                svc.create_clip(
-                    plan_id=plan_id,
-                    clip_type="main",
-                    order=seg.segment_order,
-                    duration=avg_duration,
-                    config={
-                        "material_type": seg.material_type or "",
-                        "template_segment_id": seg.id,
-                    },
-                )
-            logger.info(
-                "模板编辑器自动兜底: plan=%s 从旧模板 segments 复制了 %d 个片段",
-                plan_id,
-                len(segments),
-            )
-
-
-def _auto_fallback_assign_assets(svc: EditPlanService, plan_id: str, plan_check) -> list:
-    """自动兜底 3: 为没有素材的片段分配素材。返回剩余无素材片段列表。"""
-    all_clips = svc.list_clips(plan_id)
-    clips_without_asset = [c for c in all_clips if not c.asset_id]
-    config_asset_ids = (plan_check.config or {}).get("asset_ids", [])
-
-    logger.info(
-        "模板编辑器自动兜底3 诊断: plan=%s total_clips=%d " "clips_without_asset=%d config_asset_ids=%r",
-        plan_id,
-        len(all_clips),
-        len(clips_without_asset),
-        config_asset_ids[:5] if config_asset_ids else [],
-    )
-
-    if clips_without_asset and config_asset_ids:
-        logger.info(
-            "模板编辑器自动兜底3: plan=%s 为 %d 个无素材片段分配 %d 个指定素材",
-            plan_id,
-            len(clips_without_asset),
-            len(config_asset_ids),
-        )
-        assigned = 0
-        for i, clip in enumerate(clips_without_asset):
-            asset_idx = i % len(config_asset_ids)
-            try:
-                svc.assign_asset(clip.id, config_asset_ids[asset_idx])
-                assigned += 1
-            except Exception as exc:
-                logger.error(
-                    "模板编辑器自动兜底3: plan=%s clip=%s 分配素材 %s 失败: %s",
-                    plan_id,
-                    clip.id,
-                    config_asset_ids[asset_idx],
-                    exc,
-                )
-        logger.info(
-            "模板编辑器自动兜底3: plan=%s 素材分配完成 assigned=%d/%d",
-            plan_id,
-            assigned,
-            len(clips_without_asset),
-        )
-        # 重新检查剩余无素材片段
-        all_clips_after = svc.list_clips(plan_id)
-        clips_without_asset = [c for c in all_clips_after if not c.asset_id]
-        if clips_without_asset:
-            logger.warning(
-                "模板编辑器自动兜底3: plan=%s 仍有 %d 个片段无素材",
-                plan_id,
-                len(clips_without_asset),
-            )
-    elif not clips_without_asset:
-        logger.info("模板编辑器自动兜底3: plan=%s 所有片段已有素材，跳过", plan_id)
-    elif not config_asset_ids:
-        logger.info(
-            "模板编辑器自动兜底3: plan=%s config.asset_ids 为空，跳过分配",
-            plan_id,
-        )
-
-    return clips_without_asset
-
-
-def _auto_fallback_auto_material_mode(
-    svc: EditPlanService,
-    plan_id: str,
-    plan_check,
-    clips_without_asset: list,
-    asset_library_repo: Any,
-    asset_repo: Any,
-    user_id: str = "",
-) -> None:
-    """自动兜底 4: 自动选素材分配给无素材片段
-
-    查找策略（按优先级）：
-    1. plan 有 project_id → 从项目素材库查找
-    2. plan 无 project_id 但有 user_id → 从用户上传的素材中查找
-    """
-    if not clips_without_asset:
-        return
-
-    ready_videos: list = []
-    source_desc = ""
-
-    # 策略 1: 通过 project_id 查找项目素材库
-    if plan_check.project_id:
-        libs = asset_library_repo.find_by_project(plan_check.project_id)
-        video_lib = None
-        for lib in libs:
-            lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
-            if lib_kind == "video":
-                video_lib = lib
-                break
-        if video_lib:
-            assets = asset_repo.find_by_library(video_lib.id)
-            ready_videos = [
-                a
-                for a in assets
-                if (a.status.value if hasattr(a.status, "value") else a.status) == "ready"
-                and a.mime_type
-                and a.mime_type.startswith("video")
-            ]
-            source_desc = f"素材库 {video_lib.name}"
-
-    # 策略 2: 通过 user_id 查找用户上传的素材
-    if not ready_videos and user_id and hasattr(asset_repo, "find_ready_videos_by_user"):
-        logger.info(
-            "模板编辑器自动兜底4: plan=%s project_id 为空，尝试通过 user_id=%s 查找素材",
-            plan_id,
-            user_id,
-        )
-        ready_videos = asset_repo.find_ready_videos_by_user(user_id)
-        source_desc = f"用户上传 (user_id={user_id[:8]}...)"
-
-    if not ready_videos:
-        logger.warning(
-            "模板编辑器自动兜底4: plan=%s 未找到可用素材 (project_id=%s, user_id=%s)",
-            plan_id,
-            plan_check.project_id or "(empty)",
-            user_id[:8] + "..." if user_id else "(empty)",
-        )
-        return
-
-    logger.info(
-        "模板编辑器自动兜底4: plan=%s 自动选素材分配给 %d 个无素材片段 (来源: %s, 共 %d 个)",
-        plan_id,
-        len(clips_without_asset),
-        source_desc,
-        len(ready_videos),
-    )
-    random.shuffle(ready_videos)
-    for i, clip in enumerate(clips_without_asset):
-        asset = ready_videos[i % len(ready_videos)]
-        svc.assign_asset(clip.id, asset.id)
-    logger.info(
-        "模板编辑器自动兜底4: plan=%s 从 %s 分配了 %d 个素材给 %d 个片段",
-        plan_id,
-        source_desc,
-        len(ready_videos),
-        len(clips_without_asset),
-    )

apps/api/app/api/routes/templates_editor/dependencies.py

@@ -3,7 +3,6 @@
 核心依赖：
 - get_editor_services: 获取模板+计划服务
 - get_draft_plan_id: 根据 template_id 获取或创建草稿，返回 plan_id
-- _check_queue_limits: 生成队列限流检查
 """
 
 from __future__ import annotations
@@ -11,7 +10,6 @@ from __future__ import annotations
 import logging
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.core.task_enqueue import GLOBAL_PENDING_LIMIT, USER_PENDING_LIMIT
 from app.dependencies import get_db_session
 from app.services.edit_plan_service import EditPlanService
 from app.services.edit_template_service import EditTemplateService
@@ -113,29 +111,3 @@ def get_draft_plan_id(
         user_id,
     )
     return plan.id
-
-
-def _check_queue_limits(gen_task_repo, user_id: str) -> None:
-    """队列限流预检查"""
-    try:
-        has_count = (
-            hasattr(gen_task_repo, "count_pending_by_user")
-            and hasattr(gen_task_repo, "count_pending_total")
-        )
-        if has_count:
-            user_pending = gen_task_repo.count_pending_by_user(user_id)
-            global_pending = gen_task_repo.count_pending_total()
-            if user_pending >= USER_PENDING_LIMIT:
-                raise HTTPException(
-                    status_code=429,
-                    detail=f"您的待处理任务过多（当前 {user_pending}/{USER_PENDING_LIMIT}），请等待完成后再提交",
-                )
-            if global_pending >= GLOBAL_PENDING_LIMIT:
-                raise HTTPException(
-                    status_code=503,
-                    detail="系统繁忙，请稍后再试",
-                )
-    except HTTPException:
-        raise
-    except Exception as e:
-        logger.warning("[模板编辑器队列限流] 检查失败，跳过: %s", e)

apps/api/app/api/routes/templates_editor/draft.py

@@ -17,6 +17,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
 
 from .dependencies import get_draft_plan_id, get_editor_services
 from .schemas import (
+    EditorClipBatchUpdateRequest,
+    EditorClipBatchUpdateResponse,
     EditorDraftResponse,
     EditorPublishResponse,
     EditorRollbackRequest,
@@ -126,11 +128,7 @@ def list_template_versions(
             clip_count=len(v.clip_configs),
             change_note=v.change_note,
             published_by=v.published_by,
-            created_at=(
-                v.created_at.isoformat()
-                if hasattr(v.created_at, "isoformat")
-                else str(v.created_at)
-            ),
+            created_at=(v.created_at.isoformat() if hasattr(v.created_at, "isoformat") else str(v.created_at)),
         )
         for v in versions
     ]
@@ -162,3 +160,35 @@ def rollback_template(
         new_version=tpl.version,
         clip_count=len(clip_configs),
     )
+
+
+@router.put("/clips", response_model=EditorClipBatchUpdateResponse)
+def batch_update_clips(
+    template_id: str,
+    req: EditorClipBatchUpdateRequest,
+    plan_id: str = Depends(get_draft_plan_id),
+    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
+    _: AuthenticatedUser = Depends(get_current_user),
+):
+    """批量替换草稿clips（全量覆盖，用于前端选择素材后同步片段）
+
+    事务保证：清空→创建→标记ready 在同一数据库事务内完成，
+    任何步骤失败时自动回滚，避免数据不一致。
+    """
+    _, plan_svc = services
+    plan_svc.get_plan_or_raise(plan_id)
+
+    clips_data = []
+    for clip_item in req.clips:
+        item = {
+            "asset_id": clip_item.asset_id,
+            "start_time": clip_item.start_time,
+            "duration": clip_item.duration,
+        }
+        if clip_item.order is not None:
+            item["order"] = clip_item.order
+        clips_data.append(item)
+
+    plan_svc.replace_all_clips_transactional(plan_id, clips_data)
+
+    return EditorClipBatchUpdateResponse(plan_id=plan_id, clip_count=len(req.clips))

apps/api/app/api/routes/templates_editor/generation.py

@@ -1,385 +0,0 @@
-"""草稿生成路由.
-
-端点：
-- POST /generate              触发生成
-- GET  /generation-status     生成进度
-- GET  /generations           生成记录列表
-"""
-
-from __future__ import annotations
-
-import json
-import logging
-from typing import Any, Optional
-
-from app.auth import AuthenticatedUser, get_current_user
-from app.core.celery_app import celery_app
-from app.core.storage import OSSStorageService, get_storage_service
-from app.dependencies import (
-    get_asset_library_repository,
-    get_asset_repository,
-    get_db_session,
-    get_generated_video_repository,
-)
-from app.schemas.generation_task import GenerationTaskResponse
-from app.services.edit_plan_service import EditPlanService
-from app.services.edit_template_service import EditTemplateService
-from fastapi import APIRouter, Depends, HTTPException, status
-from sqlalchemy.orm import Session
-
-from packages.adapters.sqlalchemy_impl.generation_task_repository import (
-    SQLAlchemyGenerationTaskRepository,
-)
-from packages.application.generated_videos import ListGeneratedVideosByTaskUseCase
-from packages.application.generation_tasks import (
-    CreateGenerationTaskCommand,
-    CreateGenerationTaskUseCase,
-)
-from packages.domain.edit_plan import EditPlanStatus
-
-from ._fallback import (
-    _auto_fallback_assign_assets,
-    _auto_fallback_auto_material_mode,
-    _auto_fallback_copy_template_clips,
-    _auto_fallback_draft_to_editing,
-)
-from .dependencies import _check_queue_limits, get_draft_plan_id, get_editor_services
-from .schemas import (
-    ClipStatusItem,
-    EditPlanGenerateRequest,
-    EditPlanGenerateResponse,
-    EditPlanGenerationsResponse,
-    EditPlanGenerationStatusResponse,
-)
-
-logger = logging.getLogger(__name__)
-router = APIRouter(tags=["Template Editor"])
-
-
-# DEPRECATED: 前端已改用 /generation/tasks 体系，此路由保留仅供旧版兼容，计划下线
-@router.post("/generate", response_model=EditPlanGenerateResponse)
-def generate_editor_draft(
-    template_id: str,
-    request: Optional[EditPlanGenerateRequest] = None,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    db: Session = Depends(get_db_session),
-    current_user: AuthenticatedUser = Depends(get_current_user),
-    asset_library_repo: Any = Depends(get_asset_library_repository),
-    asset_repo: Any = Depends(get_asset_repository),
-) -> EditPlanGenerateResponse:
-    """触发模板草稿渲染生成"""
-    req = request or EditPlanGenerateRequest()
-    _, plan_svc = services
-    plan_check = plan_svc.get_plan_or_raise(plan_id)
-
-    # 自动兜底流程
-    _auto_fallback_draft_to_editing(plan_svc, plan_id, plan_check)
-    _auto_fallback_copy_template_clips(plan_svc, plan_id, plan_check, db)
-    clips_without_asset = _auto_fallback_assign_assets(plan_svc, plan_id, plan_check)
-    _auto_fallback_auto_material_mode(
-        plan_svc,
-        plan_id,
-        plan_check,
-        clips_without_asset,
-        asset_library_repo,
-        asset_repo,
-        user_id=str(current_user.user.id),
-    )
-
-    # 检查是否可复用已完成的预览产物（预览品质已与正式一致）
-    gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-    reusable_task = _find_reusable_preview_task(gen_task_repo, plan_id, plan_check)
-    if reusable_task:
-        # 复用预览产物：标记为正式产出，跳过渲染
-        # 如果前端传了 title_config，需要创建新任务（因为预览任务的 custom_title 可能不同）
-        title_config_reuse = req.title_config or {}
-        title_text_reuse = (title_config_reuse.get("text") or "").strip()
-        existing_custom_title = getattr(reusable_task, "custom_title", "") or ""
-        if title_text_reuse and existing_custom_title:
-            # 如果新标题和已有标题不同，不能复用，走新建任务流程
-            new_title_json = json.dumps(title_config_reuse, ensure_ascii=False)
-            if new_title_json != existing_custom_title:
-                logger.info(
-                    "[模板生成] 标题已变更，跳过复用: task_id=%s",
-                    reusable_task.id,
-                )
-                reusable_task = None
-        elif title_text_reuse and not existing_custom_title:
-            # 原来没标题，现在有标题，不能复用
-            logger.info(
-                "[模板生成] 新增标题，跳过复用: task_id=%s",
-                reusable_task.id,
-            )
-            reusable_task = None
-        elif not title_text_reuse and existing_custom_title:
-            # 原来有标题，现在移除了，不能复用
-            logger.info(
-                "[模板生成] 移除标题，跳过复用: task_id=%s",
-                reusable_task.id,
-            )
-            reusable_task = None
-
-    if reusable_task:
-        # 复用预览产物：标记为正式产出，跳过渲染
-        reusable_task.mark_confirmed()
-        gen_task_repo.update(reusable_task)
-
-        # 将产物 URL 写入 plan config
-        rendered_url = _get_task_output_url(reusable_task, gen_task_repo, db)
-        plan_svc.update_plan_config(
-            plan_id,
-            {
-                "generation_task_id": reusable_task.id,
-                "rendered_storage_key": rendered_url,  # 统一用 rendered_storage_key
-            },
-        )
-        plan_svc.transition_status(plan_id, EditPlanStatus.COMPLETED)
-
-        updated_plan = plan_svc.get_plan_or_raise(plan_id)
-        logger.info(
-            "模板编辑器复用预览产物: template_id=%s plan_id=%s task_id=%s by user=%s",
-            template_id,
-            plan_id,
-            reusable_task.id,
-            current_user.user.id,
-        )
-        return EditPlanGenerateResponse(
-            plan_id=plan_id,
-            plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
-            generation_task_id=reusable_task.id,
-            clip_count=len((plan_check.config or {}).get("clips", [])),
-        )
-
-    # 检查是否可生成（含最后防线自动修复 + 诊断日志）
-    try:
-        can_gen, reason = plan_svc.can_generate(plan_id)
-    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
-    if not can_gen:
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=reason)
-
-    try:
-        clip_count = plan_svc.mark_clips_ready(plan_id)
-
-        user_id = current_user.user.id
-        _check_queue_limits(gen_task_repo, user_id)
-
-        gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
-        plan = plan_svc.get_plan_or_raise(plan_id)
-        config_asset_ids = (plan.config or {}).get("asset_ids", [])
-        # 从 plan config 读取封面 URL（由 generate-cover 保存）
-        cover_url_from_config = (plan.config or {}).get("cover", {}).get("image_url", "")
-
-        # 处理标题配置：序列化 title_config 为 JSON 存入 custom_title
-        title_config = req.title_config or {}
-        title_text = (title_config.get("text") or "").strip()
-        custom_title_value = ""
-        if title_text:
-            custom_title_value = json.dumps(title_config, ensure_ascii=False)
-            logger.info(
-                "[模板生成] 标题配置: text=%s, config_keys=%s",
-                title_text[:30],
-                list(title_config.keys()),
-            )
-
-        gen_task = gen_task_use_case.execute(
-            CreateGenerationTaskCommand(
-                project_id=plan.project_id or "",
-                template_id=plan.template_id,
-                created_by_user_id=current_user.user.id,
-                source_edit_plan_id=plan_id,
-                asset_ids=list(config_asset_ids) if config_asset_ids else [],
-                cover_url=cover_url_from_config,
-                custom_title=custom_title_value,
-            ),
-        )
-
-        plan_svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id})
-        plan_svc.transition_status(plan_id, EditPlanStatus.RENDERING)
-        celery_app.send_task("worker.render_edit_plan", args=[plan_id])
-
-        updated_plan = plan_svc.get_plan_or_raise(plan_id)
-
-        logger.info(
-            "模板编辑器触发生成: template_id=%s plan_id=%s gen_task_id=%s clips=%d by user=%s",
-            template_id,
-            plan_id,
-            gen_task.id,
-            clip_count,
-            current_user.user.id,
-        )
-
-        return EditPlanGenerateResponse(
-            plan_id=plan_id,
-            plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
-            generation_task_id=gen_task.id,
-            clip_count=clip_count,
-        )
-    except HTTPException:
-        raise
-    except Exception as _e:
-        logger.exception(
-            "模板编辑器触发生成失败: template_id=%s plan_id=%s",
-            template_id,
-            plan_id,
-        )
-        try:
-            plan_svc.transition_status(plan_id, EditPlanStatus.FAILED)
-        except Exception:
-            pass
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="生成失败，请稍后重试",
-        ) from _e
-
-
-def _find_reusable_preview_task(gen_task_repo, plan_id: str, plan) -> "object | None":
-    """查找该 plan 关联的已完成预览任务，判断是否可复用。
-
-    复用条件：
-    1. 存在 source_edit_plan_id == plan_id 的已完成预览任务
-    2. plan 在预览完成后未被修改（updated_at <= 预览完成时间）
-
-    Returns:
-        可复用的 GenerationTask，或 None
-    """
-    try:
-        tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
-    except Exception:
-        return None
-
-    for task in tasks:
-        if not getattr(task, "is_preview", False):
-            continue
-        if not task.is_completed:
-            continue
-        # 检查 plan 是否在预览完成后被修改
-        completed_at = getattr(task, "completed_at", None)
-        if completed_at and hasattr(plan, "updated_at"):
-            plan_updated = plan.updated_at
-            # 如果 plan.updated_at 为空，无法判断是否修改过，跳过
-            if plan_updated is None:
-                continue
-            # 如果 plan 在预览完成后又被修改了，不能复用
-            if plan_updated > completed_at:
-                continue
-        return task
-    return None
-
-
-def _get_task_output_url(task, gen_task_repo, db) -> str:
-    """获取任务的输出视频 URL。"""
-    try:
-        video_repo = get_generated_video_repository(db)
-        use_case = ListGeneratedVideosByTaskUseCase(video_repo)
-        videos = use_case.execute(task.id)
-        if videos:
-            url = getattr(videos[0], "file_url", "") or ""
-            # 规范化：合并路径中的双斜杠（保留协议头 ://）
-            if url:
-                import re as _re
-                url = _re.sub(r"(?<!:)//", "/", url)
-            return url
-    except Exception:
-        pass
-    return ""
-
-
-# DEPRECATED: 前端已改用 /generation/tasks 体系，此路由保留仅供旧版兼容，计划下线
-@router.get("/generation-status", response_model=EditPlanGenerationStatusResponse)
-def get_editor_generation_status(
-    template_id: str,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    storage_service: OSSStorageService = Depends(get_storage_service),
-    _: AuthenticatedUser = Depends(get_current_user),
-) -> EditPlanGenerationStatusResponse:
-    """查询草稿生成进度"""
-    _, plan_svc = services
-    try:
-        gen_status = plan_svc.get_generation_status(plan_id)
-    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
-
-    plan = gen_status["plan"]
-    clips = gen_status["clips"]
-
-    clip_items = [
-        ClipStatusItem(
-            clip_id=c.id,
-            clip_type=c.clip_type,
-            order=c.order,
-            status=c.status.value if hasattr(c.status, "value") else c.status,
-            asset_id=c.asset_id or "",
-            text_content=c.text_content or "",
-            duration=c.duration,
-        )
-        for c in clips
-    ]
-
-    raw_video_url = (plan.config or {}).get("rendered_storage_key", "") or (plan.config or {}).get("rendered_url", "")
-    video_url = ""
-    if raw_video_url:
-        if raw_video_url.startswith("http"):
-            video_url = raw_video_url  # 已经是完整 URL
-        else:
-            try:
-                video_url = storage_service.get_url(raw_video_url)  # storage_key -> 完整 URL
-            except Exception as e:
-                logger.warning("生成视频URL获取失败: template_id=%s error=%s", template_id, e)
-                video_url = raw_video_url
-
-    progress = gen_status.get("progress", 0.0)
-    error_message = gen_status.get("error_message", "")
-    gen_task_status = gen_status.get("generation_task_status")
-    plan_status_val = plan.status.value if hasattr(plan.status, "value") else plan.status
-    if plan_status_val == "completed" and progress < 100:
-        progress = 100.0
-
-    return EditPlanGenerationStatusResponse(
-        plan_id=plan_id,
-        plan_status=plan_status_val,
-        generation_task_id=gen_status["generation_task_id"],
-        generation_task_status=gen_task_status,
-        progress=progress,
-        video_url=video_url,
-        error_message=error_message,
-        clips=clip_items,
-    )
-
-
-@router.get("/generations", response_model=EditPlanGenerationsResponse)
-def list_editor_generations(
-    template_id: str,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    db: Session = Depends(get_db_session),
-    _: AuthenticatedUser = Depends(get_current_user),
-) -> EditPlanGenerationsResponse:
-    """查询草稿关联的生成记录列表"""
-    _, plan_svc = services
-    plan_svc.get_plan_or_raise(plan_id)
-
-    gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-    tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
-    items = [
-        GenerationTaskResponse(
-            id=t.id,
-            project_id=t.project_id,
-            asset_library_id=t.asset_library_id,
-            strategy_id=t.strategy_id,
-            voice_library_id=t.voice_library_id,
-            template_id=t.template_id,
-            asset_ids=t.asset_ids,
-            title_ids=t.title_ids,
-            voice_ids=t.voice_ids,
-            source_edit_plan_id=t.source_edit_plan_id or "",
-            status=t.status.value if hasattr(t.status, "value") else t.status,
-            progress=t.progress,
-            result_count=t.result_count,
-            error_message=t.error_message,
-        )
-        for t in tasks
-    ]
-    return EditPlanGenerationsResponse(items=items, total=len(items))

apps/api/app/api/routes/templates_editor/schemas.py

@@ -6,9 +6,8 @@
 from __future__ import annotations
 
 import re as _re
-from typing import Any, Dict, List, Optional
+from typing import Any, List, Optional
 
-from app.schemas.generation_task import GenerationTaskResponse
 from pydantic import BaseModel, Field, validator
 
 _EXPORT_RESOLUTION_PATTERN = _re.compile(r"^\d+x\d+$")
@@ -16,58 +15,6 @@ _EXPORT_VALID_QUALITY_PRESETS = {"ultra_fast", "fast", "balanced", "high", "best
 _EXPORT_VALID_FORMATS = {"mp4", "mov"}
 
 
-# ── 生成状态相关 ────────────────────────────────────────────────────────────
-
-
-class ClipStatusItem(BaseModel):
-    """片段生成状态"""
-
-    clip_id: str
-    clip_type: str
-    order: int
-    status: str
-    asset_id: str
-    text_content: str
-    duration: float
-
-
-class EditPlanGenerationStatusResponse(BaseModel):
-    """剪辑计划生成进度响应体"""
-
-    plan_id: str
-    plan_status: str
-    generation_task_id: Optional[str] = None
-    generation_task_status: Optional[str] = None
-    progress: float = 0.0
-    video_url: str = ""
-    error_message: str = ""
-    clips: List[ClipStatusItem]
-
-
-class EditPlanGenerateRequest(BaseModel):
-    """模板编辑器触发生成请求体"""
-    title_config: Optional[Dict[str, Any]] = Field(
-        default_factory=dict,
-        description="标题配置（可选），渲染时烧录到视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow",
-    )
-
-
-class EditPlanGenerateResponse(BaseModel):
-    """剪辑计划触发生成响应体"""
-
-    plan_id: str
-    plan_status: str
-    generation_task_id: str
-    clip_count: int
-
-
-class EditPlanGenerationsResponse(BaseModel):
-    """剪辑计划关联的生成记录列表响应体"""
-
-    items: List[GenerationTaskResponse]
-    total: int
-
-
 # ── AI 推荐 ────────────────────────────────────────────────────────────────
 
 
@@ -75,12 +22,8 @@ class AIRecommendRequest(BaseModel):
     """AI 推荐片段方案请求体"""
 
     asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表")
-    editing_mode: str = Field(
-        default="one_take", description="剪辑模式: one_take / pip / voice_over / voice_pip"
-    )
-    target_duration: float = Field(
-        default=30.0, ge=1.0, le=600.0, description="目标时长（秒）"
-    )
+    editing_mode: str = Field(default="one_take", description="剪辑模式: one_take / pip / voice_over / voice_pip")
+    target_duration: float = Field(default=30.0, ge=1.0, le=600.0, description="目标时长（秒）")
 
 
 class AIRecommendClipItem(BaseModel):
@@ -107,8 +50,6 @@ class AIRecommendResponse(BaseModel):
     confidence: float = Field(..., ge=0.0, le=1.0, description="AI 推荐置信度 (0~1)")
 
 
-
-
 # ── BGM ────────────────────────────────────────────────────────────────────
 
 
@@ -224,9 +165,7 @@ class ClipBatchDeleteResponse(BaseModel):
 class ClipsFromAssetsRequest(BaseModel):
     """从素材批量创建片段请求"""
 
-    asset_ids: List[str] = Field(
-        ..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾"
-    )
+    asset_ids: List[str] = Field(..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾")
     clip_type: str = Field(default="main", description="片段类型，默认 main")
 
 
@@ -501,6 +440,28 @@ class EditorClipUpdateRequest(BaseModel):
     config: Optional[dict[str, Any]] = None
 
 
+class EditorClipBatchItem(BaseModel):
+    """批量更新clips的单个片段"""
+
+    asset_id: str = Field(default="", max_length=100, description="关联素材ID，可为空（占位片段）")
+    start_time: float = Field(default=0.0, ge=0.0)
+    duration: float = Field(default=0.0, ge=0.0)
+    order: Optional[int] = Field(default=None, ge=0, description="排序，None表示按数组顺序")
+
+
+class EditorClipBatchUpdateRequest(BaseModel):
+    """批量替换clips请求（全量覆盖）"""
+
+    clips: List[EditorClipBatchItem] = Field(default_factory=list)
+
+
+class EditorClipBatchUpdateResponse(BaseModel):
+    """批量更新clips响应"""
+
+    plan_id: str
+    clip_count: int
+
+
 class EditorPublishResponse(BaseModel):
     """发布草稿响应"""
 

apps/api/app/services/edit_plan_service.py

@@ -371,6 +371,85 @@ class EditPlanService:
         logger.info("删除所有片段: plan_id=%s count=%d", plan_id, count)
         return count
 
+    def replace_all_clips_transactional(
+        self,
+        plan_id: str,
+        clips_data: list[dict],
+    ) -> int:
+        """事务性地替换所有片段：清空→创建→标记ready，单事务保证原子性。
+
+        Args:
+            plan_id: 计划 ID
+            clips_data: 片段数据列表，每项包含 asset_id/start_time/duration/order
+
+        Returns:
+            int: 创建的片段数量
+
+        Raises:
+            Exception: 任何步骤失败时自动回滚
+        """
+        from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel
+
+        db = self._clip_repo.session
+        try:
+            # 1. 清空现有 clips（不 commit）
+            deleted_count = db.query(EditPlanClipModel).filter(EditPlanClipModel.plan_id == plan_id).delete()
+
+            # 2. 批量创建新 clips（不 commit）
+            for i, clip_item in enumerate(clips_data):
+                order = clip_item.get("order") or i
+                clip = EditPlanClip.create(
+                    plan_id=plan_id,
+                    clip_type="main",
+                    order=order,
+                    asset_id=clip_item.get("asset_id", ""),
+                    start_time=clip_item.get("start_time", 0.0),
+                    duration=clip_item.get("duration", 0.0),
+                )
+                model = EditPlanClipModel(
+                    id=clip.id,
+                    plan_id=clip.plan_id,
+                    clip_type=clip.clip_type,
+                    order=clip.order,
+                    asset_id=clip.asset_id,
+                    text_content=clip.text_content,
+                    start_time=clip.start_time,
+                    duration=clip.duration,
+                    transition_effect=clip.transition_effect,
+                    transition_duration=clip.transition_duration,
+                    playback_speed=clip.playback_speed,
+                    status=clip.status.value,
+                    config=clip.config,
+                )
+                db.add(model)
+
+            # 3. 标记有 asset_id 的 clips 为 ready（不 commit）
+            pending_with_asset = (
+                db.query(EditPlanClipModel)
+                .filter(
+                    EditPlanClipModel.plan_id == plan_id,
+                    EditPlanClipModel.status == "pending",
+                    EditPlanClipModel.asset_id != "",
+                )
+                .all()
+            )
+            for m in pending_with_asset:
+                m.status = "ready"
+
+            # 4. 一次性提交
+            db.commit()
+            logger.info(
+                "事务性替换片段: plan_id=%s deleted=%d created=%d",
+                plan_id,
+                deleted_count,
+                len(clips_data),
+            )
+            return len(clips_data)
+        except Exception:
+            db.rollback()
+            logger.exception("事务性替换片段失败: plan_id=%s", plan_id)
+            raise
+
     # ── 片段分割与合并 ──────────────────────────────────────────────────────
 
     def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]:

apps/web/src/api/template-editor/editPlans.ts

@@ -1,15 +1,8 @@
 /**
- * 模板草稿 CRUD + 生成相关 API
+ * 模板草稿 CRUD API
  */
 import apiClient from "../client"
-import type {
-  EditPlan,
-  UpdateEditPlanRequest,
-  GenerateResponse,
-  GenerationStatusResponse,
-  EditPlanGeneration,
-  GeneratedVideo,
-} from "./types"
+import type { EditPlan, UpdateEditPlanRequest, GeneratedVideo } from "./types"
 
 /** 获取单个模板草稿 */
 export async function getEditPlan(templateId: string): Promise<EditPlan> {
@@ -27,26 +20,34 @@ export async function updateEditPlan(
   return response.data
 }
 
-/** 触发生成 */
-export async function generateEditPlan(templateId: string): Promise<GenerateResponse> {
-  const response = await apiClient.post(`/templates/${templateId}/editor/generate`)
-  return response.data
-}
-
-/** 获取生成状态（轮询用） */
-export async function getGenerationStatus(templateId: string): Promise<GenerationStatusResponse> {
-  const response = await apiClient.get(`/templates/${templateId}/editor/generation-status`)
-  return response.data
-}
-
-/** 获取模板草稿关联的生成记录 */
-export async function getEditPlanGenerations(templateId: string): Promise<EditPlanGeneration[]> {
-  const response = await apiClient.get(`/templates/${templateId}/editor/generations`)
-  return response.data.items || []
-}
-
 /** 获取生成任务的视频结果列表 */
 export async function getGenerationTaskResults(taskId: string): Promise<GeneratedVideo[]> {
   const response = await apiClient.get(`/generation/tasks/${taskId}/results`)
   return response.data.items || response.data || []
 }
+
+/** ── 草稿 clips 批量更新 ── */
+
+export interface EditPlanClipInput {
+  asset_id: string
+  start_time: number
+  duration: number
+  order: number
+}
+
+/**
+ * 批量替换草稿的 clips（先全删再批量插入）
+ * 后端路由：PUT /templates/{template_id}/editor/clips
+ */
+export async function updateEditPlanClips(
+  templateId: string,
+  clips: EditPlanClipInput[],
+  signal?: AbortSignal,
+): Promise<{ count: number }> {
+  const response = await apiClient.put(
+    `/templates/${templateId}/editor/clips`,
+    { clips },
+    { signal },
+  )
+  return response.data
+}

apps/web/src/api/template-editor/index.ts

@@ -16,10 +16,7 @@ export type {
   EditPlanConfig,
   EditPlan,
   UpdateEditPlanRequest,
-  GenerateResponse,
-  EditPlanGeneration,
   ClipStatusItem,
-  GenerationStatusResponse,
   GeneratedVideo,
   AIRecommendRequest,
   AIRecommendClipItem,
@@ -51,11 +48,10 @@ export {
 export {
   getEditPlan,
   updateEditPlan,
-  generateEditPlan,
-  getGenerationStatus,
-  getEditPlanGenerations,
+  updateEditPlanClips,
   getGenerationTaskResults,
 } from "./editPlans"
+export type { EditPlanClipInput } from "./editPlans"
 
 // 片段 CRUD + 批量操作
 export {

apps/web/src/api/template-editor/types.ts

@@ -187,31 +187,6 @@ export interface EditPlanListResponse {
 
 /* ── 生成相关 ── */
 
-/** 生成响应 */
-export interface GenerateResponse {
-  plan_id: string
-  plan_status: EditPlanStatus
-  generation_task_id: string
-  clip_count: number
-}
-
-/** 模板草稿关联的生成记录 */
-export interface EditPlanGeneration {
-  id: string
-  source_edit_plan_id: string
-  template_id: string
-  asset_ids: string[]
-  status: EditPlanStatus
-  progress: number
-  result_count: number
-  error_message: string
-  error_info: Record<string, unknown>
-  logs: Array<Record<string, unknown>>
-  retry_count: number
-  created_at?: string
-  updated_at?: string
-}
-
 /** 片段生成状态 */
 export interface ClipStatusItem {
   clip_id: string
@@ -224,17 +199,6 @@ export interface ClipStatusItem {
   error_message?: string
 }
 
-/** 生成状态轮询响应 */
-export interface GenerationStatusResponse {
-  plan_id: string
-  plan_status: EditPlanStatus
-  generation_task_id?: string
-  error_message?: string
-  clips: ClipStatusItem[]
-  error?: string
-  message?: string
-}
-
 /** 生成视频详情 */
 export interface GeneratedVideo {
   id: string

apps/web/src/config/navigation.ts

@@ -71,7 +71,7 @@ export const NAV_ITEMS: NavItem[] = [
   },
   {
     key: "editing-planner",
-    label: "剪辑编辑器",
+    label: "剪辑模板",
     path: "/app/editing-planner",
     icon: React.createElement(EditOutlined),
   },
@@ -133,7 +133,7 @@ export const NAV_GROUPS: NavGroup[] = [
       },
       {
         key: "editing-planner",
-        label: "剪辑编辑器",
+        label: "剪辑模板",
         path: "/app/editing-planner",
         icon: React.createElement(EditOutlined),
       },

apps/web/src/pages/editing-planner/components/GenerationHistoryModal.tsx

@@ -1,116 +0,0 @@
-/**
- * 生成历史弹窗 — 展示当前模板草稿的生成任务记录
- * 从 EditingPlanner 拆分，避免主文件过大
- */
-import React from "react"
-import { CloseOutlined, InboxOutlined } from "@ant-design/icons"
-import type { EditPlanGeneration } from "@/api/template-editor"
-import { PLAN_STATUS_LABELS } from "@/api/template-editor"
-
-interface GenerationHistoryModalProps {
-  open: boolean
-  loading: boolean
-  history: EditPlanGeneration[]
-  onClose: () => void
-  onCancel?: (taskId: string) => void
-  cancelLoading?: boolean
-}
-
-const GenerationHistoryModal: React.FC<GenerationHistoryModalProps> = ({
-  open,
-  loading,
-  history,
-  onClose,
-  onCancel,
-  cancelLoading,
-}) => {
-  if (!open) return null
-
-  return (
-    <div className="ep-modal-overlay" onClick={onClose}>
-      <div className="ep-modal ep-gh-modal" onClick={(e) => e.stopPropagation()}>
-        <div className="ep-modal-header">
-          <h3>生成历史</h3>
-          <button className="ep-modal-close" onClick={onClose}>
-            <CloseOutlined />
-          </button>
-        </div>
-        <div className="ep-modal-body ep-gh-body">
-          {loading ? (
-            <div className="ep-gh-empty">
-              <div className="ep-skeleton">
-                <div className="ep-skeleton-item ep-skeleton-item--header" />
-                <div className="ep-skeleton-item" />
-                <div className="ep-skeleton-item" />
-                <div className="ep-skeleton-item" />
-              </div>
-            </div>
-          ) : history.length === 0 ? (
-            <div className="ep-gh-empty">
-              <InboxOutlined style={{ fontSize: 32, opacity: 0.4 }} />
-              <span>暂无生成记录</span>
-            </div>
-          ) : (
-            <table className="ep-gh-table">
-              <thead>
-                <tr className="ep-gh-table-header-row">
-                  <th className="ep-gh-th">任务ID</th>
-                  <th className="ep-gh-th">状态</th>
-                  <th className="ep-gh-th">创建时间</th>
-                  <th className="ep-gh-th">更新时间</th>
-                  {onCancel && <th className="ep-gh-th">操作</th>}
-                </tr>
-              </thead>
-              <tbody>
-                {history.map((gen) => {
-                  const statusClass = `ep-gh-status-tag--${gen.status}`
-                  const canCancel = gen.status === "rendering" || gen.status === "editing"
-                  return (
-                    <tr key={gen.id} className="ep-gh-table-row">
-                      <td className="ep-gh-td ep-gh-td-id">
-                        {gen.id ? `${gen.id.slice(0, 8)}...` : "—"}
-                      </td>
-                      <td className="ep-gh-td">
-                        <span className={`ep-gh-status-tag ${statusClass}`}>
-                          {PLAN_STATUS_LABELS[gen.status] || gen.status}
-                        </span>
-                      </td>
-                      <td className="ep-gh-td ep-gh-td-time">
-                        {gen.created_at ? new Date(gen.created_at).toLocaleString("zh-CN") : "—"}
-                      </td>
-                      <td className="ep-gh-td ep-gh-td-time">
-                        {gen.updated_at ? new Date(gen.updated_at).toLocaleString("zh-CN") : "—"}
-                      </td>
-                      {onCancel && (
-                        <td className="ep-gh-td ep-gh-td-action">
-                          {canCancel ? (
-                            <button
-                              className="ep-gh-cancel-btn"
-                              onClick={() => onCancel(gen.id)}
-                              disabled={cancelLoading}
-                            >
-                              取消
-                            </button>
-                          ) : (
-                            <span className="ep-gh-action-placeholder">—</span>
-                          )}
-                        </td>
-                      )}
-                    </tr>
-                  )
-                })}
-              </tbody>
-            </table>
-          )}
-        </div>
-        <div className="ep-modal-footer">
-          <button className="ep-btn ep-btn-secondary" onClick={onClose}>
-            关闭
-          </button>
-        </div>
-      </div>
-    </div>
-  )
-}
-
-export default GenerationHistoryModal

apps/web/src/pages/editing-planner/components/GenerationProgressModal.tsx

@@ -1,6 +0,0 @@
-/**
- * 生成进度弹窗 — 入口文件（向后兼容）
- * 实际实现已移至 ./generation-progress-modal/ 目录
- */
-export { default } from "./generation-progress-modal"
-export type { GenPhase, GenerationProgressModalProps } from "./generation-progress-modal"

apps/web/src/pages/editing-planner/components/generation-progress-modal/ProgressPhase.tsx

@@ -1,65 +0,0 @@
-import React from "react"
-import { Modal } from "@/components/ui"
-import type { TaskItem } from "@/api/tasks"
-import { getStepLabel, getStatusColor } from "./constants"
-
-interface ProgressPhaseProps {
-  open: boolean
-  task: TaskItem | null
-  onCancel: () => void
-}
-
-/** progress（进度轮询）阶段弹窗 */
-export const ProgressPhase: React.FC<ProgressPhaseProps> = ({ open, task, onCancel }) => {
-  const progress = task?.progress ?? 0
-  const status = task?.status ?? ""
-  const currentStep = task?.current_step ?? ""
-  const userMessage = task?.user_message ?? ""
-  const stepColor = getStatusColor(status, currentStep)
-
-  return (
-    <Modal open={open} title="视频生成中" footer={null} onCancel={onCancel} closable width={480}>
-      <div className="ep-gen-progress">
-        {/* 进度环 */}
-        <div className="ep-gen-progress-ring-wrap">
-          <svg className="ep-gen-progress-ring" viewBox="0 0 120 120">
-            <circle className="ep-gen-progress-ring-bg" cx="60" cy="60" r="52" />
-            <circle
-              className="ep-gen-progress-ring-fill"
-              cx="60"
-              cy="60"
-              r="52"
-              style={{
-                strokeDasharray: `${2 * Math.PI * 52}`,
-                strokeDashoffset: `${2 * Math.PI * 52 * (1 - progress / 100)}`,
-                stroke: stepColor,
-              }}
-            />
-          </svg>
-          <span className="ep-gen-progress-pct" style={{ color: stepColor }}>
-            {progress}%
-          </span>
-        </div>
-
-        {/* 当前步骤 */}
-        <div className="ep-gen-step-text">
-          {userMessage || getStepLabel(currentStep) || "处理中…"}
-        </div>
-
-        {/* 进度条 */}
-        <div className="ep-gen-progress-bar">
-          <div
-            className="ep-gen-progress-bar-fill"
-            style={{
-              width: `${progress}%`,
-              backgroundColor: stepColor,
-            }}
-          />
-        </div>
-
-        {/* 任务 ID */}
-        {task?.id && <div className="ep-gen-task-id">任务 ID: {task.id}</div>}
-      </div>
-    </Modal>
-  )
-}

apps/web/src/pages/editing-planner/components/generation-progress-modal/ResultPhase.tsx

@@ -1,82 +0,0 @@
-import React from "react"
-import { Modal, Button } from "@/components/ui"
-import type { TaskItem } from "@/api/tasks"
-
-interface ResultPhaseProps {
-  open: boolean
-  phase: "completed" | "failed"
-  task: TaskItem | null
-  onCancel: () => void
-  onRetry?: () => void
-  onClose?: () => void
-}
-
-/** completed / failed（结果）阶段弹窗 */
-export const ResultPhase: React.FC<ResultPhaseProps> = ({
-  open,
-  phase,
-  task,
-  onCancel,
-  onRetry,
-  onClose,
-}) => {
-  const userMessage = task?.user_message ?? ""
-  const errorMessage = task?.error_message ?? ""
-  const retryable = task?.retryable ?? false
-  const handleClose = onClose || onCancel
-
-  if (phase === "completed") {
-    return (
-      <Modal
-        open={open}
-        title="✅ 生成完成"
-        footer={null}
-        onCancel={handleClose}
-        closable
-        width={440}
-      >
-        <div className="ep-gen-result">
-          <div className="ep-gen-result-icon">🎉</div>
-          <div className="ep-gen-result-title">视频生成完成！</div>
-          {userMessage && <div className="ep-gen-result-msg">{userMessage}</div>}
-          <div className="ep-gen-result-actions">
-            <Button buttonType="primary" onClick={handleClose}>
-              查看结果
-            </Button>
-          </div>
-        </div>
-      </Modal>
-    )
-  }
-
-  return (
-    <Modal
-      open={open}
-      title="❌ 生成失败"
-      footer={null}
-      onCancel={handleClose}
-      closable
-      width={440}
-    >
-      <div className="ep-gen-result ep-gen-result--error">
-        <div className="ep-gen-result-icon">😥</div>
-        <div className="ep-gen-result-title">视频生成失败</div>
-        {(errorMessage || userMessage) && (
-          <div className="ep-gen-result-msg ep-gen-result-msg--error">
-            {errorMessage || userMessage}
-          </div>
-        )}
-        <div className="ep-gen-result-actions">
-          {retryable && onRetry && (
-            <Button buttonType="primary" onClick={onRetry}>
-              🔄 重试
-            </Button>
-          )}
-          <Button buttonType="secondary" onClick={handleClose}>
-            关闭
-          </Button>
-        </div>
-      </div>
-    </Modal>
-  )
-}

apps/web/src/pages/editing-planner/components/generation-progress-modal/SetupPhase.tsx

@@ -1,53 +0,0 @@
-import React from "react"
-import { Modal } from "@/components/ui"
-
-interface SetupPhaseProps {
-  open: boolean
-  voiceoverDuration: number | null
-  estimatedDuration: number
-  submitting: boolean
-  onDurationChange: (v: number | null) => void
-  onGenerate: () => void
-  onCancel: () => void
-}
-
-/** setup（配置）阶段弹窗 */
-export const SetupPhase: React.FC<SetupPhaseProps> = ({
-  open,
-  voiceoverDuration,
-  estimatedDuration,
-  submitting,
-  onDurationChange,
-  onGenerate,
-  onCancel,
-}) => (
-  <Modal
-    open={open}
-    title="使用模板生成视频"
-    confirmLoading={submitting}
-    onOk={onGenerate}
-    onCancel={onCancel}
-    okText="开始生成"
-    cancelText="取消"
-    width={440}
-  >
-    <div className="ep-gen-setup">
-      <label className="ep-gen-field-label">配音时长（秒）</label>
-      <input
-        type="number"
-        className="ep-gen-duration-input"
-        placeholder="请输入配音时长"
-        value={voiceoverDuration ?? ""}
-        onChange={(e) => {
-          const v = e.target.value ? Number(e.target.value) : null
-          onDurationChange(v)
-        }}
-        min={1}
-        max={600}
-      />
-      <div className="ep-gen-estimate">
-        预估总时长：<strong>{estimatedDuration}s</strong>
-      </div>
-    </div>
-  </Modal>
-)

apps/web/src/pages/editing-planner/components/generation-progress-modal/constants.ts

@@ -1,31 +0,0 @@
-/* ──────────── 步骤文案映射 ──────────── */
-
-export const STEP_LABELS: Record<string, string> = {
-  queued: "排队中…",
-  preparing: "准备素材…",
-  generating_video: "渲染视频中…",
-  adding_effects: "添加特效…",
-  composing: "合成中…",
-  encoding: "编码输出中…",
-  completed: "生成完成！",
-  failed: "生成失败",
-}
-
-export const getStepLabel = (step: string) => STEP_LABELS[step] || step.replace(/_/g, " ")
-
-/* ──────────── 状态徽标颜色 ──────────── */
-
-export const STATUS_COLOR: Record<string, string> = {
-  queued: "#6b7280",
-  pending: "#6b7280",
-  preparing: "#f59e0b",
-  generating_video: "#4f46e5",
-  adding_effects: "#7c3aed",
-  composing: "#2563eb",
-  encoding: "#0891b2",
-  completed: "#10b981",
-  failed: "#ef4444",
-}
-
-export const getStatusColor = (status: string, currentStep: string) =>
-  STATUS_COLOR[status] || STATUS_COLOR[currentStep] || "#4f46e5"

apps/web/src/pages/editing-planner/components/generation-progress-modal/index.tsx

@@ -1,70 +0,0 @@
-/**
- * 生成进度弹窗 — 任务 2.17
- * 三阶段 UI：setup（配置）→ progress（进度轮询）→ completed / failed（结果）
- * V21 设计系统，CSS 类名前缀 ep-gen-
- */
-import React, { useEffect, useRef } from "react"
-import type { GenPhase, GenerationProgressModalProps } from "./types"
-import { SetupPhase } from "./SetupPhase"
-import { ProgressPhase } from "./ProgressPhase"
-import { ResultPhase } from "./ResultPhase"
-
-/* 重新导出类型，保持向后兼容 */
-export type { GenPhase, GenerationProgressModalProps }
-
-const GenerationProgressModal: React.FC<GenerationProgressModalProps> = ({
-  open,
-  phase,
-  voiceoverDuration,
-  estimatedDuration,
-  onDurationChange,
-  onGenerate,
-  task,
-  submitting,
-  onCancel,
-  onRetry,
-  onClose,
-}) => {
-  /* 关闭弹窗时重置（避免下次打开残留旧状态） */
-  const prevOpen = useRef(false)
-  useEffect(() => {
-    if (prevOpen.current && !open) {
-      /* modal just closed — parent handles reset */
-    }
-    prevOpen.current = open
-  }, [open])
-
-  /* setup 阶段 */
-  if (phase === "setup") {
-    return (
-      <SetupPhase
-        open={open}
-        voiceoverDuration={voiceoverDuration}
-        estimatedDuration={estimatedDuration}
-        submitting={submitting}
-        onDurationChange={onDurationChange}
-        onGenerate={onGenerate}
-        onCancel={onCancel}
-      />
-    )
-  }
-
-  /* progress 阶段 */
-  if (phase === "progress") {
-    return <ProgressPhase open={open} task={task} onCancel={onCancel} />
-  }
-
-  /* completed / failed 阶段 */
-  return (
-    <ResultPhase
-      open={open}
-      phase={phase as "completed" | "failed"}
-      task={task}
-      onCancel={onCancel}
-      onRetry={onRetry}
-      onClose={onClose}
-    />
-  )
-}
-
-export default GenerationProgressModal

apps/web/src/pages/editing-planner/components/generation-progress-modal/types.ts

@@ -1,23 +0,0 @@
-import type { TaskItem } from "@/api/tasks"
-
-export type GenPhase = "setup" | "progress" | "completed" | "failed"
-
-export interface GenerationProgressModalProps {
-  open: boolean
-  phase: GenPhase
-
-  /* setup 阶段 */
-  voiceoverDuration: number | null
-  estimatedDuration: number
-  onDurationChange: (v: number | null) => void
-  onGenerate: () => void
-
-  /* progress / 结果阶段 */
-  task: TaskItem | null
-
-  /* 通用 */
-  submitting: boolean
-  onCancel: () => void
-  onRetry?: () => void
-  onClose?: () => void
-}

apps/web/src/pages/generate/components/GenerateStepContent.tsx

@@ -123,6 +123,10 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
     presetVoices,
   } = props
 
+  /* 当前模板的 segments，传给 Step2 构建 clips */
+  const currentTemplate = userTemplates.find((t) => t.id === selectedTemplate)
+  const templateSegments = currentTemplate?.segments
+
   switch (currentStep) {
     case 1:
       return (
@@ -142,6 +146,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
           smartSelectedIds={smartSelectedIds}
           onSmartSelectedIdsChange={onSmartSelectedIdsChange}
           selectedTemplate={selectedTemplate}
+          templateSegments={templateSegments}
         />
       )
     case 3:

apps/web/src/pages/generate/components/Step2MaterialSelect.tsx

@@ -2,6 +2,7 @@
  * Step 2 素材选择组件
  */
 import React from "react"
+import type { TemplateSegment } from "@/api/templates/types"
 import { useStep2Materials } from "../hooks/useStep2Materials"
 import MaterialModeTabs from "./material/MaterialModeTabs"
 import ManualMaterialList from "./material/ManualMaterialList"
@@ -17,6 +18,8 @@ interface Step2MaterialSelectProps {
   onSmartSelectedIdsChange: (ids: string[]) => void
   /** 当前选中的模板/草稿 ID，用于自动保存 */
   selectedTemplate?: string
+  /** 当前模板的 segments（用于构建 clips duration） */
+  templateSegments?: TemplateSegment[]
 }
 
 const Step2MaterialSelect: React.FC<Step2MaterialSelectProps> = (props) => {

apps/web/src/pages/generate/hooks/useStep2Materials.ts

@@ -3,7 +3,11 @@
  * 组合素材库加载 + 智能匹配两个子 Hook
  */
 import { useCallback, useEffect, useRef } from "react"
+import type { AssetItem } from "@/api/assets"
+import type { TemplateSegment } from "@/api/templates/types"
+import { updateEditPlanClips } from "@/api/template-editor"
 import { formatDuration } from "../utils/formatDuration"
+import { buildClipsFromAssets } from "../utils/buildClipsFromAssets"
 import { useMaterialLibrary } from "./step2-materials/useMaterialLibrary"
 import { useSmartMatch } from "./step2-materials/useSmartMatch"
 import { useDraftAutoSave } from "./useDraftAutoSave"
@@ -17,6 +21,8 @@ interface UseStep2MaterialsProps {
   onSmartSelectedIdsChange: (ids: string[]) => void
   /** 当前选中的模板/草稿 ID，用于自动保存 */
   selectedTemplate?: string
+  /** 当前模板的 segments（用于构建 clips duration） */
+  templateSegments?: TemplateSegment[]
 }
 
 export function useStep2Materials({
@@ -27,6 +33,7 @@ export function useStep2Materials({
   smartSelectedIds,
   onSmartSelectedIdsChange,
   selectedTemplate,
+  templateSegments,
 }: UseStep2MaterialsProps) {
   const { libraries, selectedLibraryId, setSelectedLibraryId, materials, materialsLoading } =
     useMaterialLibrary()
@@ -57,7 +64,7 @@ export function useStep2Materials({
     handleSmartMatch()
   }, [selectedLibraryId, materialMode, materialsLoading, materials.items, handleSmartMatch])
 
-  /* ── Step2 选择素材后自动保存草稿（防抖 500ms，失败静默） ── */
+  /* ── Step2 选择素材后自动保存草稿 asset_ids（防抖 500ms，失败静默） ── */
   const { scheduleSave } = useDraftAutoSave(selectedTemplate)
   useEffect(() => {
     if (!selectedTemplate) return
@@ -65,6 +72,61 @@ export function useStep2Materials({
     scheduleSave({ asset_ids: ids }, 500)
   }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, scheduleSave])
 
+  /* ── Step2 选择素材后同步写入 edit_plan_clips（防抖 800ms，失败静默） ── */
+  const clipsTimerRef = useRef<ReturnType<typeof setTimeout>>()
+  const clipsAbortRef = useRef<AbortController | null>(null)
+  const templateSegmentsRef = useRef(templateSegments)
+  templateSegmentsRef.current = templateSegments
+  const selectedTemplateRef = useRef(selectedTemplate)
+  selectedTemplateRef.current = selectedTemplate
+  const materialsRef = useRef(materials)
+  materialsRef.current = materials
+  const smartMatchedRef = useRef<AssetItem[]>(smartMatch.smartMatchedResults)
+  smartMatchedRef.current = smartMatch.smartMatchedResults
+
+  useEffect(() => {
+    const tid = selectedTemplateRef.current
+    if (!tid) return
+    const ids = materialMode === "auto" ? smartSelectedIds : selectedMaterials
+    if (!ids.length) return
+
+    if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
+    clipsTimerRef.current = setTimeout(async () => {
+      // 取消上一次未完成的请求
+      if (clipsAbortRef.current) clipsAbortRef.current.abort()
+      const controller = new AbortController()
+      clipsAbortRef.current = controller
+
+      const clips = buildClipsFromAssets({
+        selectedIds: ids,
+        materials: materialsRef.current.items,
+        smartMatchedAssets: smartMatchedRef.current,
+        templateSegments: templateSegmentsRef.current || [],
+      })
+
+      try {
+        await updateEditPlanClips(tid, clips, controller.signal)
+      } catch (err) {
+        const name = (err as { name?: string })?.name
+        if (name !== "CanceledError" && name !== "AbortError") {
+          console.warn("[useStep2Materials] 写入 clips 失败:", err)
+        }
+      }
+    }, 800)
+
+    return () => {
+      if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
+    }
+  }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, templateSegments])
+
+  // 组件卸载时取消未完成请求
+  useEffect(() => {
+    return () => {
+      if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
+      if (clipsAbortRef.current) clipsAbortRef.current.abort()
+    }
+  }, [])
+
   /* ── 手动选择素材 ── */
   const handleToggleMaterial = useCallback(
     (materialId: string) => {

apps/web/src/pages/generate/utils/buildClipsFromAssets.ts

@@ -0,0 +1,57 @@
+/**
+ * 将选中素材 + 模板 segments 构建为 edit_plan_clips 写入数据。
+ *
+ * 逻辑必须与 FrontendPreviewPlayer.tsx 中 buildPlaybackSegments 完全一致：
+ *   assetDuration = asset.duration || asset.metadata?.duration || 30
+ *   tplSeg = templateSegments[i] || lastSegment
+ *   segDuration = clamp(assetDuration, tplSeg.duration_min, tplSeg.duration_max)
+ *   start_time = 0
+ *   duration = segDuration
+ */
+import type { AssetItem } from "@/api/assets"
+import type { TemplateSegment } from "@/api/templates/types"
+import type { EditPlanClipInput } from "@/api/template-editor"
+
+interface BuildClipsOptions {
+  /** 选中的素材 ID 列表（按选择顺序） */
+  selectedIds: string[]
+  /** 已加载的素材列表（用于查 duration） */
+  materials: AssetItem[]
+  /** 智能匹配返回的素材（auto 模式下可能不在 materials 列表中） */
+  smartMatchedAssets?: AssetItem[]
+  /** 模板 segments */
+  templateSegments?: TemplateSegment[]
+}
+
+export function buildClipsFromAssets({
+  selectedIds,
+  materials,
+  smartMatchedAssets = [],
+  templateSegments = [],
+}: BuildClipsOptions): EditPlanClipInput[] {
+  if (!selectedIds.length) return []
+
+  // 合并两个素材来源，建立 id → asset 索引
+  const assetMap = new Map<string, AssetItem>()
+  for (const a of materials) assetMap.set(a.id, a)
+  for (const a of smartMatchedAssets) assetMap.set(a.id, a)
+
+  const lastSeg = templateSegments[templateSegments.length - 1]
+
+  return selectedIds.map((assetId, i) => {
+    const asset = assetMap.get(assetId)
+    const assetDuration = asset?.duration || asset?.metadata?.duration || 30
+
+    const tplSeg = templateSegments[i] || lastSeg
+    const segDuration = tplSeg
+      ? Math.min(tplSeg.duration_max, Math.max(tplSeg.duration_min, assetDuration))
+      : Math.min(assetDuration, 10)
+
+    return {
+      asset_id: assetId,
+      start_time: 0,
+      duration: segDuration,
+      order: i,
+    }
+  })
+}

apps/web/src/pages/templates/hooks/useTemplateLibrary.ts

@@ -87,7 +87,7 @@ export const useTemplateLibrary = () => {
     [copyMutation],
   )
 
-  /* 操作：使用模板 → 跳转剪辑编辑器 */
+  /* 操作：使用模板 → 跳转剪辑模板 */
   const handleUse = useCallback(
     (template: TemplateItem) => {
       navigate(`/app/editing-planner?templateId=${template.id}`)

apps/web/src/test/api/edit-plans.test.ts

@@ -2,10 +2,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest"
 import {
   getEditPlan,
   updateEditPlan,
-  generateEditPlan,
-  getGenerationStatus,
   aiRecommendClips,
-  getEditPlanGenerations,
   getGenerationTaskResults,
   getEditPlanClips,
   getEditPlanClip,
@@ -80,38 +77,6 @@ describe("editPlans API", () => {
     })
   })
 
-  describe("generateEditPlan", () => {
-    it("should resolve successfully", async () => {
-      await expect(generateEditPlan("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(generateEditPlan("test-planId")).rejects.toThrow()
-    })
-  })
-
-  describe("getGenerationStatus", () => {
-    it("should resolve successfully", async () => {
-      await expect(getGenerationStatus("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(getGenerationStatus("test-planId")).rejects.toThrow()
-    })
-  })
-
   describe("aiRecommendClips", () => {
     it("should resolve successfully", async () => {
       await expect(aiRecommendClips("test-planId")).resolves.not.toThrow()
@@ -128,22 +93,6 @@ describe("editPlans API", () => {
     })
   })
 
-  describe("getEditPlanGenerations", () => {
-    it("should resolve successfully", async () => {
-      await expect(getEditPlanGenerations("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(getEditPlanGenerations("test-planId")).rejects.toThrow()
-    })
-  })
-
   describe("getGenerationTaskResults", () => {
     it("should resolve successfully", async () => {
       await expect(getGenerationTaskResults("test-taskId")).resolves.not.toThrow()

apps/web/src/test/pages/EditingPlanner.test.tsx

@@ -148,11 +148,8 @@ vi.mock("@/api/editing-planner", () => ({
 
 vi.mock("@/api/template-editor", () => ({
   getMediaAssets: vi.fn().mockResolvedValue({ items: [] }),
-  getEditPlanGenerations: vi.fn().mockResolvedValue({ items: [] }),
   getEditPlan: vi.fn().mockResolvedValue({}),
   updateEditPlan: vi.fn().mockResolvedValue({}),
-  generateEditPlan: vi.fn().mockResolvedValue({ task_id: "test-task" }),
-  getGenerationStatus: vi.fn().mockResolvedValue({ status: "completed" }),
   getGenerationTaskResults: vi.fn().mockResolvedValue({ items: [] }),
   getEditPlanClips: vi.fn().mockResolvedValue({ items: [] }),
   createEditPlanClip: vi.fn().mockResolvedValue({}),
@@ -223,12 +220,6 @@ vi.mock("@/pages/editing-planner/components/StickerPanel", () => ({
 vi.mock("@/pages/editing-planner/components/SaveModal", () => ({
   default: () => React.createElement("div", { "data-testid": "SaveModal" }),
 }))
-vi.mock("@/pages/editing-planner/components/GenerationHistoryModal", () => ({
-  default: () => React.createElement("div", { "data-testid": "GenerationHistoryModal" }),
-}))
-vi.mock("@/pages/editing-planner/components/GenerationProgressModal", () => ({
-  default: () => React.createElement("div", { "data-testid": "GenerationProgressModal" }),
-}))
 
 // === useUndoRedo hook mock ===
 vi.mock("@/pages/editing-planner/hooks/useUndoRedo", () => ({

apps/web/src/test/pages/GeneratePage.test.tsx

@@ -214,12 +214,6 @@ vi.mock("@/api/titles", () => ({
 }))
 
 vi.mock("@/api/template-editor", () => ({
-  generateEditPlan: vi.fn().mockResolvedValue({
-    plan_id: "test-plan",
-    generation_task_id: "test-task",
-    plan_status: "processing",
-    clip_count: 5,
-  }),
   updateEditPlan: vi.fn().mockResolvedValue({ plan_id: "test-plan", template_id: "test-template" }),
   getEditPlan: vi.fn().mockResolvedValue({
     plan_id: "test-plan",
@@ -228,9 +222,6 @@ vi.mock("@/api/template-editor", () => ({
     config: {},
     status: "draft",
   }),
-  getGenerationStatus: vi
-    .fn()
-    .mockResolvedValue({ plan_status: "completed", generation_task_id: "test-task", clips: [] }),
   getGenerationTaskResults: vi.fn().mockResolvedValue({ items: [] }),
 }))
 

apps/web/src/test/pages/editing-planner/components/GenerationHistoryModal.test.tsx

@@ -1,39 +0,0 @@
-import { describe, expect, it, vi } from "vitest"
-import { render } from "@testing-library/react"
-import GenerationHistoryModal from "@/pages/editing-planner/components/GenerationHistoryModal"
-
-const defaultProps = {
-  open: true,
-  loading: false,
-  history: [],
-  onClose: vi.fn(),
-  onCancel: vi.fn(),
-  cancelLoading: false,
-}
-
-describe("GenerationHistoryModal", () => {
-  it("should render without crashing", () => {
-    const { container } = render(<GenerationHistoryModal {...defaultProps} />)
-    expect(container).toBeTruthy()
-  })
-
-  it("should not render when open is false", () => {
-    const { container } = render(<GenerationHistoryModal {...defaultProps} open={false} />)
-    expect(container.firstChild).toBeNull()
-  })
-
-  it("should render with history items", () => {
-    const history = [
-      {
-        id: "1",
-        status: "completed",
-        created_at: "2024-01-01T00:00:00Z",
-        duration: 60,
-      },
-    ]
-    const { container } = render(
-      <GenerationHistoryModal {...defaultProps} history={history as any} />,
-    )
-    expect(container).toBeTruthy()
-  })
-})

apps/web/src/test/pages/editing-planner/components/GenerationProgressModal.test.tsx

@@ -1,102 +0,0 @@
-import { describe, expect, it, vi } from "vitest"
-import { render } from "@testing-library/react"
-import GenerationProgressModal from "@/pages/editing-planner/components/GenerationProgressModal"
-
-const baseProps = {
-  open: true,
-  voiceoverDuration: null,
-  estimatedDuration: 60,
-  onDurationChange: vi.fn(),
-  onGenerate: vi.fn(),
-  task: null,
-  submitting: false,
-  onCancel: vi.fn(),
-  onRetry: vi.fn(),
-  onClose: vi.fn(),
-}
-
-describe("GenerationProgressModal", () => {
-  it("should render setup phase", () => {
-    const { container } = render(<GenerationProgressModal {...baseProps} phase="setup" />)
-    expect(container).toBeTruthy()
-  })
-
-  it("should render progress phase without task", () => {
-    const { container } = render(<GenerationProgressModal {...baseProps} phase="progress" />)
-    expect(container).toBeTruthy()
-  })
-
-  it("should render progress phase with task data", () => {
-    const { container } = render(
-      <GenerationProgressModal
-        {...baseProps}
-        phase="progress"
-        task={
-          {
-            id: "task-123",
-            status: "generating_video",
-            progress: 50,
-            current_step: "generating_video",
-            user_message: "正在生成视频",
-          } as any
-        }
-      />,
-    )
-    expect(container).toBeTruthy()
-  })
-
-  it("should render completed phase", () => {
-    const { container } = render(
-      <GenerationProgressModal
-        {...baseProps}
-        phase="completed"
-        task={{ id: "task-1", status: "completed", progress: 100 } as any}
-      />,
-    )
-    expect(container).toBeTruthy()
-  })
-
-  it("should render failed phase with retry", () => {
-    const { container } = render(
-      <GenerationProgressModal
-        {...baseProps}
-        phase="failed"
-        task={
-          {
-            id: "task-1",
-            status: "failed",
-            progress: 30,
-            error_message: "生成失败",
-            retryable: true,
-          } as any
-        }
-      />,
-    )
-    expect(container).toBeTruthy()
-  })
-
-  it("should render failed phase without retry", () => {
-    const { container } = render(
-      <GenerationProgressModal
-        {...baseProps}
-        phase="failed"
-        task={
-          {
-            id: "task-1",
-            status: "failed",
-            progress: 30,
-            retryable: false,
-          } as any
-        }
-      />,
-    )
-    expect(container).toBeTruthy()
-  })
-
-  it("should not render when closed", () => {
-    const { container } = render(
-      <GenerationProgressModal {...baseProps} phase="setup" open={false} />,
-    )
-    expect(container).toBeTruthy()
-  })
-})

apps/web/src/test/pages/editing-planner/smoke.test.tsx

@@ -31,8 +31,6 @@ import "@/pages/editing-planner/components/ClipPropertiesPanel"
 import "@/pages/editing-planner/components/EditorClipList"
 import "@/pages/editing-planner/components/EditingDrawers"
 import "@/pages/editing-planner/components/FilterPanel"
-import "@/pages/editing-planner/components/GenerationHistoryModal"
-import "@/pages/editing-planner/components/GenerationProgressModal"
 import "@/pages/editing-planner/components/GreenScreenPanel"
 import "@/pages/editing-planner/components/IntroOutroPanel"
 import "@/pages/editing-planner/components/MediaPanel"

apps/worker/video_processing/__init__.py

@@ -7,7 +7,7 @@ VideoProcessor 等）按需从子模块导入，避免 __init__ 阶段引入
 packages / DB 等重依赖。
 """
 
-# 共享工具模块（零外部依赖，供 editing_modes / generation / edit_plan_generation 等复用）
+# 共享工具模块（零外部依赖，供 editing_modes / generation 等复用）
 from . import dedup_helpers, ffmpeg_utils, oss_helpers, url_security
 
 __all__ = [

apps/worker/video_processing/dedup_helpers.py

@@ -1,6 +1,6 @@
 """查重辅助函数 — 从 generation.py 提取的 GeneratedVideo 记录 + 查重逻辑.
 
-供 render_edit_plan 和 generate_video 共同复用，
+供 generate_video 共同复用，
 创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。
 """
 

apps/worker/video_processing/oss_helpers.py

@@ -1,4 +1,4 @@
-"""OSS 工具函数 — 从 generation.py / edit_plan_generation.py 提取的共享 OSS 操作.
+"""OSS 工具函数 — 从 generation.py 提取的共享 OSS 操作.
 
 提供 OSS 配置读取、Bucket 创建、素材上传/下载、asset_id → 本地路径解析
 等能力，供 render_edit_plan 和 generate_video 共同复用。

apps/worker/video_processing/render_adapter.py

@@ -128,6 +128,7 @@ class RenderAdapter:
         job_id: str = "",
         work_dir: Path | None = None,
         progress_cb: ProgressCallback | None = None,
+        voiceover_audio_path: str | None = None,
     ) -> RenderAdapterResult:
         """渲染一个 EditPlan。
 
@@ -142,6 +143,7 @@ class RenderAdapter:
             job_id: 关联的 Job ID（用于结果存储路径）
             work_dir: 工作目录，不传则使用临时目录
             progress_cb: 进度回调函数
+            voiceover_audio_path: 配音音频本地路径（一键生成场景使用）
 
         Returns:
             RenderAdapterResult
@@ -208,6 +210,7 @@ class RenderAdapter:
                 progress_cb=progress_cb,
                 rendered_clip_ids=rendered_clip_ids,
                 failed_clip_ids=failed_clip_ids,
+                voiceover_audio_path=voiceover_audio_path,
             )
 
         except subprocess.CalledProcessError as exc:

apps/worker/video_processing/thumbnail_generator.py

@@ -21,12 +21,13 @@ def apply_title_overlay(
     image_path: str,
     title_text: str,
     *,
+    color: str = "#ffffff",
     position: str = "bottom",
     font_size: int | None = None,
     margin_ratio: float = 0.06,
     stroke_width_ratio: float = 0.04,
 ) -> str:
-    """在图片上绘制标题文字（白色 + 黑色描边/阴影）。
+    """在图片上绘制标题文字（指定颜色 + 黑色描边/阴影）。
 
     委托给 packages.shared.title_overlay.apply_title_to_image，
     保持 Worker 内调用方式不变。title_text 为空时直接返回原路径。
@@ -38,6 +39,7 @@ def apply_title_overlay(
     result = apply_title_to_image(
         image_path,
         title_text,
+        color=color,
         position=position,
         font_size=font_size,
         margin_ratio=margin_ratio,
@@ -208,6 +210,9 @@ def extract_and_upload_cover_frames(
     *,
     num_frames: int = 3,
     title_text: str = "",
+    title_color: str = "#ffffff",
+    title_position: str = "bottom",
+    title_font_size: int | None = None,
 ) -> list[dict]:
     """从视频中抽取多帧作为封面候选，上传到 OSS。
 
@@ -215,8 +220,11 @@ def extract_and_upload_cover_frames(
         video_path: 视频文件路径
         plan_id: 编辑计划 ID（用于生成 storage key）
         num_frames: 抽取帧数（默认 3）
-        title_text: 标题文字；非空时用 Pillow 叠加到每帧（白色 + 黑色描边）。
+        title_text: 标题文字；非空时用 Pillow 叠加到每帧。
             从已渲染视频抽帧时通常传空（标题已烧录）；从源素材抽帧时传标题。
+        title_color: 标题字体颜色（#RRGGBB）
+        title_position: 标题位置 top/center/bottom
+        title_font_size: 标题字号，None 时自动计算
 
     Returns:
         封面候选列表，每项包含 {"url": str, "position": float}
@@ -244,7 +252,13 @@ def extract_and_upload_cover_frames(
             )
             # 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过
             if title_text and title_text.strip():
-                apply_title_overlay(frame_path, title_text)
+                apply_title_overlay(
+                    frame_path,
+                    title_text,
+                    color=title_color,
+                    position=title_position,
+                    font_size=title_font_size,
+                )
             storage_key = f"covers/{plan_id}/frame_{i}.jpg"
             url = upload_to_oss(frame_path, storage_key)
             if url:

apps/worker/worker_app/celery_app.py

@@ -14,8 +14,6 @@ celery_app.conf.imports = (
     "worker_app.tasks.voice_extraction",
     "worker_app.tasks.voice_clone",
     "worker_app.tasks.tts_synthesis",
-    "worker_app.tasks.edit_plan_generation",
-    "worker_app.tasks.compose_video",
     "worker_app.tasks.batch_download",
     "worker_app.tasks._startup",
     "apps.worker.video_processing.dedup",

apps/worker/worker_app/tasks/__init__.py

@@ -25,10 +25,6 @@ def __getattr__(name: str):
         from .voice_extraction import extract_voice_task
 
         return extract_voice_task
-    elif name == "compose_video":
-        from .compose_video import compose_video
-
-        return compose_video
     elif name == "extract_background_task":
         from .voice_extraction import extract_background_task
 
@@ -58,7 +54,6 @@ def __getattr__(name: str):
 
 __all__ = [
     "classify_asset",
-    "compose_video",
     "generate_video",
     "healthcheck",
     "ingest_asset",

apps/worker/worker_app/tasks/compose_video.py

@@ -1,198 +0,0 @@
-"""视频合成 Celery 任务 — Phase 8 任务 2.10.
-
-使用 JobService 管理任务生命周期，通过 RenderAdapter 调用 UnifiedRenderService 执行合成。
-"""
-
-from __future__ import annotations
-
-import os
-import subprocess  # pragma: no cover
-import tempfile
-from pathlib import Path
-
-from celery.exceptions import SoftTimeLimitExceeded  # pragma: no cover
-from celery.utils.log import get_task_logger
-from worker_app.celery_app import celery_app
-from worker_app.db import SessionLocal
-
-logger = get_task_logger(__name__)
-
-# 任务超时时间（秒）：超过此时间 Celery 会抛出 SoftTimeLimitExceeded
-RENDER_TASK_SOFT_TIME_LIMIT = 600  # pragma: no cover  # 10 分钟
-# 硬超时：超过此时间进程会被强制 kill
-RENDER_TASK_TIME_LIMIT = 660  # pragma: no cover  # 10 分钟 + 1 分钟清理缓冲
-
-
-def _get_job_service():
-    """延迟导入 JobService，避免循环依赖。"""
-    from apps.api.app.services.job_service import JobService
-    from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository
-
-    db = SessionLocal()
-    repo = SQLAlchemyJobRepository(db)
-    return JobService(repo), db
-
-
-@celery_app.task(
-    name="worker.compose_video",
-    bind=True,
-    max_retries=3,
-    default_retry_delay=60,
-    soft_time_limit=RENDER_TASK_SOFT_TIME_LIMIT,
-    time_limit=RENDER_TASK_TIME_LIMIT,
-)
-def compose_video(self, job_id: str, **kwargs):  # pragma: no cover
-    """视频合成任务。
-
-    使用 UnifiedRenderService（图层架构）进行渲染。
-
-    Args:
-        job_id: JobService 中的任务 ID
-        **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等）
-    """
-    job_service, db = _get_job_service()
-
-    try:
-        job = job_service.get_job(job_id)
-        if job is None:
-            logger.error("Job not found: %s", job_id)
-            return {"status": "error", "message": f"Job {job_id} not found"}
-
-        plan_id = job.payload.get("plan_id", "")
-        if not plan_id:
-            job_service.fail_job(job_id, "Missing plan_id in job payload")
-            return {"status": "error", "message": "Missing plan_id"}
-
-        # 使用 unified 渲染引擎
-        return _compose_with_unified_engine(self, job_service, job, plan_id, db)
-
-    except SoftTimeLimitExceeded:
-        # Celery 软超时：任务执行超过 soft_time_limit
-        error_msg = f"渲染任务超时（超过 {RENDER_TASK_SOFT_TIME_LIMIT // 60} 分钟）"
-        logger.error("视频合成超时: job_id=%s", job_id)
-        try:
-            job_service.fail_job(job_id, error_msg)
-        except Exception:
-            logger.exception("更新 Job 超时失败状态时出错")
-        # 超时不重试
-        return {"status": "error", "message": error_msg, "error_type": "timeout"}
-
-    except subprocess.TimeoutExpired as exc:
-        # FFmpeg 子进程超时
-        error_msg = f"FFmpeg 渲染超时（{exc.timeout}s）"
-        logger.error("视频合成 FFmpeg 超时: job_id=%s timeout=%s", job_id, exc.timeout)
-        try:
-            job_service.fail_job(job_id, error_msg)
-        except Exception:
-            logger.exception("更新 Job 超时失败状态时出错")
-        # 超时不重试
-        return {"status": "error", "message": error_msg, "error_type": "ffmpeg_timeout"}
-
-    except self.retry_exc as exc:
-        logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
-        raise
-
-    except subprocess.CalledProcessError as exc:
-        # FFmpeg 执行失败，提取有意义的错误信息
-        from video_processing.video_validation import get_exit_code_message
-
-        exit_msg = get_exit_code_message(exc.returncode)
-        stderr_text = (exc.stderr or "").strip()
-        stderr_tail = stderr_text[-300:] if len(stderr_text) > 300 else stderr_text
-        error_msg = f"渲染失败: {exit_msg}"
-        if stderr_tail:
-            error_msg += f" | {stderr_tail[:200]}"
-
-        logger.error("视频合成 FFmpeg 失败: job_id=%s %s", job_id, exit_msg)
-        try:
-            job_service.fail_job(job_id, error_msg[:500])
-        except Exception:
-            logger.exception("更新 Job 失败状态时出错")
-        # FFmpeg 错误不重试（通常是素材或配置问题）
-        return {"status": "error", "message": error_msg, "error_type": "ffmpeg_error", "exit_code": exc.returncode}
-
-    except Exception as exc:
-        logger.exception("视频合成异常: job_id=%s", job_id)
-        try:
-            job_service.fail_job(job_id, str(exc)[:500])
-        except Exception:
-            logger.exception("更新 Job 失败状态时出错")
-        raise self.retry(exc=exc, countdown=60) from exc
-    finally:
-        db.close()
-
-
-def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict:  # pragma: no cover
-    """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。"""
-    job_id = job.id
-
-    # 标记为 running
-    job_service.update_progress(job_id, progress=10.0, current_stage="初始化统一渲染引擎")
-
-    from video_processing.render_adapter import RenderAdapter
-
-    adapter = RenderAdapter(db)
-
-    # 校验合成条件
-    job_service.update_progress(job_id, progress=15.0, current_stage="校验合成条件")
-    valid, errors, warnings, ready_count, total_count = adapter.validate_plan(plan_id)
-    if not valid:
-        error_msg = "; ".join(errors)
-        job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
-        return {"status": "error", "message": error_msg}
-
-    # 进度回调
-    def progress_cb(progress: float, stage: str) -> None:
-        try:
-            job_service.update_progress(job_id, progress=progress, current_stage=stage)
-        except Exception:
-            logger.exception("更新进度失败")
-
-    # 执行渲染
-    job_service.update_progress(job_id, progress=20.0, current_stage="开始渲染")
-    logger.info("统一渲染引擎开始: job_id=%s plan_id=%s", job_id, plan_id)
-
-    result = adapter.render_plan(
-        plan_id=plan_id,
-        job_id=job_id,
-        progress_cb=progress_cb,
-    )
-
-    if not result.success:
-        error_msg = f"渲染失败: {result.error_message}"
-        job_service.fail_job(job_id, error_msg[:500])
-        raise RuntimeError(result.error_message)
-
-    # 更新 Job 状态为完成
-    result_data = {
-        "plan_id": plan_id,
-        "output_path": str(result.output_path) if result.output_path else "",
-        "storage_key": f"rendered/{plan_id}/{job_id}.mp4",
-        "output_url": result.output_url,
-        "estimated_duration": result.duration,
-        "clip_count": result.clip_count,
-        "engine": "unified",
-        "width": result.width,
-        "height": result.height,
-        "file_size": result.file_size,
-    }
-    job_service.complete_job(job_id, result=result_data)
-
-    logger.info(
-        "视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs",
-        job_id,
-        plan_id,
-        result.duration,
-    )
-    return {"status": "completed", "job_id": job_id, "result": result_data}
-
-
-def _cleanup_output(job_id: str) -> None:
-    """清理临时输出文件。"""
-    try:
-        _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-        output_path = os.path.join(_output_dir, f"{job_id}.mp4")
-        if Path(output_path).exists():
-            Path(output_path).unlink()
-    except Exception as e:
-        logger.warning(f"清理输出文件失败: {e}", exc_info=True)

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -1,452 +0,0 @@
-"""剪辑计划渲染任务 — 使用 UnifiedRenderService 统一渲染引擎.
-
-Celery 任务 worker.render_edit_plan:
-  1. 加载 EditPlan + EditPlanClips
-  2. 通过 RenderAdapter 调用 UnifiedRenderService 渲染
-  3. 下载各片段素材 + 渲染
-  4. 上传渲染结果到 OSS
-  5. 创建 GeneratedVideo 记录 + 查重
-  6. 更新 EditPlan / EditPlanClip 状态
-  7. 更新 GenerationTask 进度
-"""
-
-from __future__ import annotations
-
-import logging
-from datetime import datetime, timezone
-from pathlib import Path
-
-from worker_app.celery_app import celery_app
-from worker_app.db import SessionLocal
-
-logger = logging.getLogger(__name__)
-
-OUTPUT_WIDTH = 1280
-OUTPUT_HEIGHT = 720
-OUTPUT_FPS = 25.0
-
-
-# ── 共享工具模块导入 ──────────────────────────────────────────────────────────
-
-from video_processing.dedup_helpers import create_video_record_and_dedup
-
-# ── Repository imports (延迟导入避免循环依赖) ─────────────────────────────────
-
-
-def _get_repos():
-    """获取数据库仓储实例"""
-    from packages.adapters.sqlalchemy_impl import SQLAlchemyEditPlanRepository
-    from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import (
-        SQLAlchemyEditPlanClipRepository,
-    )
-    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
-        SQLAlchemyGenerationTaskRepository,
-    )
-
-    db = SessionLocal()
-    try:
-        plan_repo = SQLAlchemyEditPlanRepository(db)
-        clip_repo = SQLAlchemyEditPlanClipRepository(db)
-        gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-        yield plan_repo, clip_repo, gen_task_repo, db
-    finally:
-        db.close()
-
-
-# ── Celery Task ───────────────────────────────────────────────────────────────
-
-
-def _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg: str):
-    """统一的计划失败标记工具。"""
-    plan = plan_repo.get(plan_id)
-    if plan and plan.status.value == "rendering":
-        plan.mark_failed()
-        plan_repo.update(plan)
-    if generation_task_id:
-        gen_task = gen_task_repo.get(generation_task_id)
-        if gen_task and gen_task.status.value != "failed":
-            gen_task.status = "failed"
-            gen_task.error_message = error_msg
-            gen_task.completed_at = datetime.now(timezone.utc)
-            try:
-                gen_task.append_log(
-                    stage="render_failed",
-                    message=error_msg[:500],
-                    level="ERROR",
-                )
-            except Exception:
-                pass
-            gen_task_repo.update(gen_task)
-
-
-def _finalize_render_success(
-    plan,
-    plan_repo,
-    clip_repo,
-    gen_task_repo,
-    db,
-    plan_id: str,
-    output_url: str,
-    storage_key: str,
-    duration: float,
-    file_size: int,
-    width: int,
-    height: int,
-    rendered_clip_ids: list[str],
-    failed_clip_ids: list[str],
-    generation_task_id: str,
-    output_path: Path,
-    engine: str,
-    thumbnail_url: str = "",
-    cover_candidates: list[dict] | None = None,
-) -> dict:
-    """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。"""
-    # 创建 GeneratedVideo 记录 + 查重
-    project_id = plan.project_id or ""
-    batch_id = plan.config.get("batch_id", "")
-    mode = plan.config.get("mode", "edit_plan")
-    # 从 plan.config.title.text 读取视频名称
-    plan_config = plan.config or {}
-    title_cfg = plan_config.get("title", {}) or {}
-    if not isinstance(title_cfg, dict):
-        title_cfg = {}
-    video_name = (title_cfg.get("text") or "").strip() or f"generated-{generation_task_id[:8]}.mp4"
-    if generation_task_id:
-        try:
-            create_video_record_and_dedup(
-                generation_task_id=generation_task_id,
-                project_id=project_id,
-                user_id=plan.created_by_user_id or "",
-                batch_id=batch_id,
-                file_url=output_url or "",
-                file_size=file_size,
-                duration=duration,
-                video_path=str(output_path),
-                mode=mode,
-                session=db,
-                width=width,
-                height=height,
-                fps=OUTPUT_FPS,
-                name=video_name,
-                thumbnail_url=thumbnail_url,
-            )
-        except Exception as dedup_err:
-            logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
-
-    # 更新片段状态为 rendered
-    for clip_id in rendered_clip_ids:
-        clip = clip_repo.get(clip_id)
-        if clip and clip.status.value == "ready":
-            clip.mark_rendered()
-            clip_repo.update(clip)
-
-    # 更新 EditPlan 状态为 completed + 回写实际渲染时长 + 结果数
-    plan.config["rendered_url"] = output_url or ""
-    plan.config["rendered_storage_key"] = storage_key
-    if hasattr(plan, "total_duration") and duration > 0:
-        plan.total_duration = duration
-    plan.mark_completed()
-    plan_repo.update(plan)
-
-    # 更新 GenerationTask 状态为 completed
-    if generation_task_id:
-        gen_task = gen_task_repo.get(generation_task_id)
-        if gen_task:
-            gen_task.status = "completed"
-            gen_task.progress = 100.0
-            # 剪辑计划是多片段合成 1 个成片，result_count = 1
-            gen_task.result_count = 1
-            gen_task.append_log(
-                stage="render_complete",
-                message=f"渲染完成，输出时长 {duration:.1f}s",
-                level="INFO",
-                engine=engine,
-                clip_count=len(rendered_clip_ids),
-            )
-            gen_task.completed_at = datetime.now(timezone.utc)
-
-            # 回写封面 URL 到 GenerationTask，供封面生成接口读取
-            if cover_candidates:
-                first_cover = cover_candidates[0].get("image_url") or cover_candidates[0].get("url") or ""
-                if first_cover:
-                    gen_task.cover_url = first_cover
-                    logger.info(
-                        "预览渲染完成，回写 cover_url: plan_id=%s task_id=%s url=%s",
-                        plan_id,
-                        generation_task_id,
-                        first_cover[:80],
-                    )
-
-            gen_task_repo.update(gen_task)
-
-    logger.info(
-        "剪辑计划渲染完成: plan_id=%s engine=%s rendered=%d failed=%d duration=%.1fs",
-        plan_id,
-        engine,
-        len(rendered_clip_ids),
-        len(failed_clip_ids),
-        duration,
-    )
-
-    return {
-        "status": "completed",
-        "plan_id": plan_id,
-        "rendered_count": len(rendered_clip_ids),
-        "failed_count": len(failed_clip_ids),
-        "output_url": output_url,
-        "duration": duration,
-    }
-
-
-def _render_with_unified(
-    plan,
-    clips,
-    plan_id: str,
-    generation_task_id: str,
-    plan_repo,
-    clip_repo,
-    gen_task_repo,
-    db,
-) -> dict:
-    """统一渲染引擎路径（通过 RenderAdapter 调用 UnifiedRenderService）。
-
-    RenderAdapter 内部处理：素材下载、BGM 准备、ASR 自动字幕、渲染执行、OSS 上传。
-    本函数只负责：业务状态更新、查重、收尾。
-    """
-    from video_processing.render_adapter import RenderAdapter
-
-    adapter = RenderAdapter(db)
-
-    # 进度回调：更新 GenerationTask 进度
-    def _progress_cb(progress: float, stage: str):
-        if not generation_task_id:
-            return
-        try:
-            gen_task = gen_task_repo.get(generation_task_id)
-            if gen_task:
-                # 映射到 30%~90% 区间（素材下载前已到 30%）
-                mapped_progress = 30.0 + progress * 0.6
-                gen_task.progress = min(mapped_progress, 95.0)
-                gen_task.append_log(
-                    stage="render_progress",
-                    message=stage,
-                    level="INFO",
-                    progress=mapped_progress,
-                )
-                gen_task_repo.update(gen_task)
-        except Exception:
-            pass
-
-    try:
-        result = adapter.render_plan(
-            plan_id=plan_id,
-            job_id=generation_task_id or plan_id,
-            progress_cb=_progress_cb,
-        )
-    except Exception as render_err:
-        logger.error("渲染失败(unified): %s — %s", plan_id, render_err)
-        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"渲染失败: {render_err}")
-        return {"status": "error", "message": f"渲染失败: {render_err}"}
-
-    if not result.success:
-        full_error = result.error_message or "渲染失败"
-        if result.error_detail:
-            full_error = f"{full_error}\n--- stderr ---\n{result.error_detail}"
-        logger.error("渲染失败(unified): %s — %s", plan_id, result.error_message)
-        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, full_error)
-        return {"status": "error", "message": result.error_message or "渲染失败"}
-
-    output_path = result.output_path or Path("")
-    output_url = result.output_url or ""
-    thumbnail_url = result.thumbnail_url or ""
-    # adapter 上传到 rendered/{plan_id}/{job_id}.mp4，从 URL 提取实际 key
-    # 不能用 output.mp4 硬编码，否则 cover 等下游通过 key 构造的 URL 指向不存在的文件
-    if output_url:
-        from packages.shared.storage import get_shared_storage_service
-
-        storage_key = get_shared_storage_service().normalize_storage_key(output_url)
-    else:
-        storage_key = f"rendered/{plan_id}/{generation_task_id or plan_id}.mp4"
-
-    # 用 adapter 返回的 clip 明细（以 adapter 的结果为准）
-    rendered_clip_ids = result.rendered_clip_ids or []
-    failed_clip_ids = result.failed_clip_ids or []
-
-    # 将封面候选帧写入 plan.config（供封面 API 直接使用，跳过 MediaKit 抽帧）
-    if result.cover_candidates:
-        plan_config = plan.config or {}
-        plan_config["cover_candidates"] = result.cover_candidates
-        plan.config = plan_config
-        logger.info(
-            "封面候选帧已写入 plan.config: plan_id=%s count=%d",
-            plan_id,
-            len(result.cover_candidates),
-        )
-
-    return _finalize_render_success(
-        plan=plan,
-        plan_repo=plan_repo,
-        clip_repo=clip_repo,
-        gen_task_repo=gen_task_repo,
-        db=db,
-        plan_id=plan_id,
-        output_url=output_url or "",
-        storage_key=storage_key,
-        duration=result.duration,
-        file_size=result.file_size,
-        width=result.width,
-        height=result.height,
-        rendered_clip_ids=rendered_clip_ids,
-        failed_clip_ids=failed_clip_ids,
-        generation_task_id=generation_task_id,
-        output_path=output_path,
-        engine="unified",
-        thumbnail_url=thumbnail_url,
-        cover_candidates=result.cover_candidates,
-    )
-
-
-@celery_app.task(
-    name="worker.render_edit_plan",
-    bind=True,
-    max_retries=2,
-    soft_time_limit=600,  # 10 分钟软超时
-    time_limit=660,  # 11 分钟硬超时
-)
-def render_edit_plan(self, plan_id: str) -> dict:  # pragma: no cover
-    """渲染剪辑计划
-
-    流程：
-    1. 加载 EditPlan + EditPlanClips
-    2. 通过 RenderAdapter 调用 UnifiedRenderService 渲染
-    3. 下载素材 + 渲染
-    4. 上传渲染结果到 OSS
-    5. 创建 GeneratedVideo 记录 + 查重
-    6. 更新 EditPlan → completed, EditPlanClips → rendered
-    7. 更新 GenerationTask 进度
-    """
-    logger.info("开始渲染剪辑计划: plan_id=%s", plan_id)
-
-    generation_task_id = ""
-
-    for repos in _get_repos():
-        plan_repo, clip_repo, gen_task_repo, db = repos
-
-        try:
-            # 1. 加载 EditPlan
-            plan = plan_repo.get(plan_id)
-            if plan is None:
-                logger.error("剪辑计划不存在: %s", plan_id)
-                return {"status": "error", "message": f"计划不存在: {plan_id}"}
-
-            # 获取 generation_task_id（提前读取，确保 except 块可用）
-            generation_task_id = plan.config.get("generation_task_id", "")
-
-            # 2. 准备渲染（使用 unified 渲染引擎）
-
-            # 3. 加载片段列表（按 order 排序）
-            clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
-            if not clips:
-                logger.warning("剪辑计划没有片段: %s", plan_id)
-                plan.mark_failed()
-                plan_repo.update(plan)
-                return {"status": "error", "message": "没有可渲染的片段"}
-
-            # 更新 GenerationTask 状态为 running
-            if generation_task_id:
-                gen_task = gen_task_repo.get(generation_task_id)
-                if gen_task:
-                    gen_task.status = "running"
-                    gen_task.started_at = datetime.now(timezone.utc)
-                    gen_task.append_log(
-                        stage="render_start",
-                        message=f"开始渲染，片段数 {len(clips)}",
-                        level="INFO",
-                        engine="unified",
-                        clip_count=len(clips),
-                    )
-                    gen_task_repo.update(gen_task)
-
-            # 3. 渲染前取消检查
-            if generation_task_id:
-                current_task = gen_task_repo.get(generation_task_id)
-                if current_task:
-                    task_status = (
-                        current_task.status.value if hasattr(current_task.status, "value") else str(current_task.status)
-                    )
-                    if task_status == "cancelled":
-                        logger.info("任务已被取消，中止渲染: plan_id=%s task_id=%s", plan_id, generation_task_id)
-
-                        if plan.status.value == "rendering":
-                            try:
-                                plan.resume_editing()
-                                plan_repo.update(plan)
-                            except ValueError:
-                                pass
-                        return {"status": "cancelled", "plan_id": plan_id, "message": "任务已取消"}
-
-            # 4. 渲染（unified 引擎：RenderAdapter 统一处理下载 + BGM + ASR + 渲染 + 上传）
-            result = _render_with_unified(
-                plan=plan,
-                clips=clips,
-                plan_id=plan_id,
-                generation_task_id=generation_task_id,
-                plan_repo=plan_repo,
-                clip_repo=clip_repo,
-                gen_task_repo=gen_task_repo,
-                db=db,
-            )
-
-            result["engine"] = "unified"
-            return result
-
-        except Exception as exc:
-            # 超时异常不重试，直接标记失败
-            from celery.exceptions import SoftTimeLimitExceeded
-
-            is_timeout = isinstance(exc, SoftTimeLimitExceeded)
-            if is_timeout:
-                logger.error("渲染剪辑计划超时: plan_id=%s", plan_id)
-            else:
-                logger.exception("渲染剪辑计划异常: %s", plan_id)
-
-            # 尝试标记计划和 GenerationTask 为失败
-            error_msg = "渲染任务超时（超过10分钟）" if is_timeout else f"渲染异常: {type(exc).__name__}: {exc}"
-            try:
-                plan = plan_repo.get(plan_id)
-                if plan and plan.status.value == "rendering":
-                    plan.mark_failed()
-                    plan_repo.update(plan)
-            except Exception as e:
-                logger.warning("标记计划失败时异常: plan_id=%s error=%s", plan_id, e, exc_info=True)
-            # 更新 GenerationTask 状态为 failed，前端轮询能看到失败状态
-            try:
-                if generation_task_id:
-                    gen_task = gen_task_repo.get(generation_task_id)
-                    if gen_task and gen_task.status.value != "failed":
-                        gen_task.status = "failed"
-                        gen_task.error_message = error_msg
-                        gen_task.completed_at = datetime.now(timezone.utc)
-                        try:
-                            log_stage = "render_timeout" if is_timeout else "render_failed"
-                            gen_task.append_log(
-                                stage=log_stage,
-                                message=error_msg[:500],
-                                level="ERROR",
-                                exception_type=type(exc).__name__,
-                            )
-                        except Exception:
-                            pass
-                        gen_task_repo.update(gen_task)
-                        logger.info(
-                            "GenerationTask 已标记为 failed: task_id=%s plan_id=%s",
-                            generation_task_id,
-                            plan_id,
-                        )
-            except Exception as e:
-                logger.warning(
-                    "更新 GenerationTask 失败状态时异常: task_id=%s error=%s", generation_task_id, e, exc_info=True
-                )
-            raise self.retry(exc=exc, countdown=60) from exc
-
-    return {"status": "error", "message": "数据库连接失败"}

apps/worker/worker_app/tasks/generation.py

@@ -223,6 +223,7 @@ def _load_template_segment_durations(template_id: str) -> list[float]:
         return []
 
 
+# DEPRECATED: 仅兼容无 source_edit_plan_id 的旧调用，后续移除
 def _build_plan_and_clips_from_task(
     task_id: str,
     downloaded_paths: list[Path],
@@ -889,6 +890,7 @@ def _load_task_info(task_id: str) -> dict | None:
             "custom_title": getattr(gen_task, "custom_title", "") or "",
             "title_config": dict(getattr(gen_task, "title_config", {}) or {}),
             "voice_ids": list(getattr(gen_task, "voice_ids", []) or []),
+            "source_edit_plan_id": getattr(gen_task, "source_edit_plan_id", "") or "",
         }
     finally:
         session.close()
@@ -976,7 +978,7 @@ def _render_video(
 
     Args:
     Returns:
-        (output_path, render_duration, cover_candidates)
+        (output_path, render_duration, cover_candidates, voiceover_path)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1204,6 +1206,133 @@ def _upload_and_record(
 # ── Celery Task ──────────────────────────────────────────────────────────────
 
 
+def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None:
+    """将 GenerationTask 的配置同步到 EditPlan.config，返回配音本地路径（如果有）。
+
+    包括：title_config、BGM、输出分辨率。配音单独处理（需下载到本地）。
+    """
+    from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
+        SQLAlchemyEditPlanRepository,
+    )
+
+    plan_repo = SQLAlchemyEditPlanRepository(db)
+    plan = plan_repo.get(source_edit_plan_id)
+    if plan is None:
+        logger.error("[task] EditPlan not found: %s", source_edit_plan_id)
+        return None
+
+    plan_config = dict(plan.config or {})
+    changed = False
+
+    # 标题配置
+    title_config = task_info.get("title_config") or {}
+    if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip():
+        cfg = dict(title_config)
+        # 字段名归一化
+        if "font_size" in cfg and "size" not in cfg:
+            cfg["size"] = cfg["font_size"]
+        if "font_color" in cfg and "color" not in cfg:
+            cfg["color"] = cfg["font_color"]
+        plan_config["title"] = cfg
+        changed = True
+        logger.info("[task] title_config synced to plan: %s", cfg.get("text", "")[:30])
+
+    # BGM 配置
+    bgm_config = task_info.get("bgm_config") or {}
+    if bgm_config:
+        from packages.domain.bgm_utils import merge_bgm_config
+
+        existing_bgm = plan_config.get("bgm", {}) or {}
+        plan_config["bgm"] = merge_bgm_config(existing_bgm, bgm_config)
+        changed = True
+
+    # 输出分辨率
+    ow = task_info.get("output_width") or OUTPUT_WIDTH
+    oh = task_info.get("output_height") or OUTPUT_HEIGHT
+    if ow >= 100 and oh >= 100:
+        export_cfg = dict(plan_config.get("export", {}) or {})
+        export_cfg["resolution"] = f"{ow}x{oh}"
+        plan_config["export"] = export_cfg
+        changed = True
+
+    if changed:
+        plan.config = plan_config
+        plan_repo.update(plan)
+        logger.info("[task] plan.config synced: plan_id=%s", source_edit_plan_id)
+
+    # 配音下载
+    voiceover_path: str | None = None
+    voice_library_id = task_info.get("voice_library_id", "")
+    voice_ids = task_info.get("voice_ids", []) or []
+    effective_voice_id = voice_library_id or (voice_ids[0] if voice_ids else "")
+
+    if effective_voice_id:
+        import tempfile
+
+        voice_tmp = Path(tempfile.gettempdir()) / f"voice_{source_edit_plan_id}_{id(task_info)}.mp3"
+        try:
+            if _download_voice_asset(effective_voice_id, voice_tmp):
+                voiceover_path = str(voice_tmp)
+                logger.info("[task] voice downloaded: %s -> %s", effective_voice_id, voiceover_path)
+        except Exception:
+            logger.warning("[task] voice download failed: %s", effective_voice_id, exc_info=True)
+
+    return voiceover_path
+
+
+def _render_from_edit_plan(
+    task_id: str,
+    source_edit_plan_id: str,
+    task_info: dict,
+) -> tuple[Path, float, list[dict] | None]:
+    """从 EditPlan 数据库记录直接渲染（不再内存重建clips）。
+
+    Returns:
+        (output_path, render_duration, cover_candidates, voiceover_path)
+    """
+    from video_processing.render_adapter import RenderAdapter
+    from worker_app.db import SessionLocal
+
+    db = SessionLocal()
+    try:
+        # 同步配置到 plan.config + 下载配音
+        voiceover_path = _sync_task_config_to_plan(source_edit_plan_id, task_info, db)
+
+        # 进度回调
+        def _progress_cb(progress: float, stage: str):
+            mapped = 40.0 + progress * 0.4
+            _update_task_progress(task_id, min(mapped, 80.0), stage)
+
+        adapter = RenderAdapter(db)
+        render_start = time.monotonic()
+        logger.info("[task_id=%s] [渲染] RenderAdapter.render_plan 开始 (plan_id=%s)", task_id, source_edit_plan_id)
+
+        result = adapter.render_plan(
+            plan_id=source_edit_plan_id,
+            job_id=task_id,
+            progress_cb=_progress_cb,
+            voiceover_audio_path=voiceover_path,
+        )
+
+        if not result.success:
+            raise RuntimeError(f"渲染失败: {result.error_message}")
+
+        render_elapsed = time.monotonic() - render_start
+        logger.info(
+            "[task_id=%s] [渲染] RenderAdapter.render_plan 完成: 耗时=%.1fs, 时长=%.2fs",
+            task_id,
+            render_elapsed,
+            result.duration,
+        )
+
+        output_path = result.output_path
+        cover_candidates = getattr(result, "cover_candidates", None)
+
+        return output_path, result.duration, cover_candidates, voiceover_path
+    finally:
+        db.close()
+
+
 @celery_app.task(
     bind=True,
     name="worker.generate_video",
@@ -1294,6 +1423,188 @@ def generate_video(self, task_id: str) -> dict:
         if template_id:
             _validate_template_exists(template_id)
 
+        # ── 新路径：有 source_edit_plan_id 时直接从数据库 EditPlan 渲染 ──
+        source_edit_plan_id = task_info.get("source_edit_plan_id", "")
+        if source_edit_plan_id:
+            voiceover_tmp_path: str | None = None
+            try:
+                logger.info(
+                    "[task_id=%s] 使用 EditPlan 数据库路径渲染: plan_id=%s",
+                    task_id,
+                    source_edit_plan_id,
+                )
+                _update_task_progress(task_id, 30, "加载草稿数据")
+
+                if gen_task:
+                    gen_task.append_log("渲染模式", "从草稿数据渲染（与预览一致）")
+                    _flush_logs(task_id, gen_task)
+
+                output_path, render_duration, cover_candidates, voiceover_tmp_path = _render_from_edit_plan(
+                    task_id=task_id,
+                    source_edit_plan_id=source_edit_plan_id,
+                    task_info=task_info,
+                )
+
+                if gen_task:
+                    gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
+                    _flush_logs(task_id, gen_task)
+
+                _update_task_progress(task_id, 80, "渲染完成")
+
+                # ── 4. 上传 OSS + 查重记录 ───────────────────────────────
+                _update_task_progress(task_id, 85, "开始上传")
+                file_url, duration, file_size, video_count = _upload_and_record(
+                    task_id=task_id,
+                    output_path=output_path,
+                    project_id=project_id,
+                    batch_id=batch_id,
+                    editing_mode=editing_mode,
+                    user_id=user_id,
+                    video_name=task_info.get("video_title", ""),
+                )
+
+                if gen_task:
+                    gen_task.append_log(
+                        "OSS上传",
+                        f"上传成功, 大小={file_size}",
+                        file_size=file_size,
+                        file_url=file_url,
+                    )
+                    _flush_logs(task_id, gen_task)
+
+                _update_task_progress(task_id, 95, "上传完成")
+
+                # ── 4.5 封面帧持久化 ────────────────────────────────────────────
+                try:
+                    if cover_candidates:
+                        first = cover_candidates[0]
+                        cover_frame_url = first.get("image_url") or first.get("url") or ""
+                        if cover_frame_url:
+                            _cover_session = SessionLocal()
+                            try:
+                                from packages.adapters.sqlalchemy_impl.models import (
+                                    GenerationTaskModel,
+                                )
+
+                                _cover_model = (
+                                    _cover_session.query(GenerationTaskModel)
+                                    .filter(GenerationTaskModel.id == task_id)
+                                    .first()
+                                )
+                                if _cover_model:
+                                    _cover_model.cover_url = cover_frame_url
+                                    meta = dict(_cover_model.extra_meta or {})
+                                    meta["cover_candidates"] = cover_candidates
+                                    _cover_model.extra_meta = meta
+                                    _cover_session.commit()
+                            finally:
+                                _cover_session.close()
+                except Exception:
+                    logger.warning("[task_id=%s] 封面帧持久化失败", task_id, exc_info=True)
+
+                # ── 5. 标记完成 ──────────────────────────────────────────────────
+                _update_task_status(task_id, "mark_completed", result_count=video_count)
+
+                # 5.1 更新标题使用次数
+                try:
+                    _title_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                            SQLAlchemyGenerationTaskRepository,
+                        )
+                        from packages.adapters.sqlalchemy_impl.title_library_repository import (
+                            SQLAlchemyTitleLibraryRepository,
+                        )
+
+                        _task_repo = SQLAlchemyGenerationTaskRepository(_title_session)
+                        _gen_task = _task_repo.get(task_id)
+                        if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id:
+                            _title_repo = SQLAlchemyTitleLibraryRepository(_title_session)
+                            for _tid in _gen_task.title_ids:
+                                try:
+                                    _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id)
+                                except Exception:
+                                    logger.warning(
+                                        "[task_id=%s] 更新标题使用次数失败: title_id=%s",
+                                        task_id,
+                                        _tid,
+                                        exc_info=True,
+                                    )
+                    finally:
+                        _title_session.close()
+                except Exception:
+                    logger.warning("[task_id=%s] 更新标题使用次数异常", task_id, exc_info=True)
+
+                # 5.2 更新素材使用次数
+                try:
+                    from worker_app.core.asset_usage import mark_asset_used_for_generation
+
+                    _asset_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.asset_repository import (
+                            SQLAlchemyAssetRepository,
+                        )
+                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                            SQLAlchemyGenerationTaskRepository,
+                        )
+
+                        _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session)
+                        _asset_repo = SQLAlchemyAssetRepository(_asset_session)
+                        _gen_task = _task_repo.get(task_id)
+                        if _gen_task and _gen_task.asset_ids:
+                            for _aid in _gen_task.asset_ids:
+                                try:
+                                    _asset = _asset_repo.get(_aid)
+                                    if _asset:
+                                        mark_asset_used_for_generation(_asset)
+                                        _asset_repo.update(_asset)
+                                except Exception:
+                                    logger.warning(
+                                        "[task_id=%s] 更新素材使用次数失败: asset_id=%s",
+                                        task_id,
+                                        _aid,
+                                        exc_info=True,
+                                    )
+                    finally:
+                        _asset_session.close()
+                except Exception:
+                    logger.warning("[task_id=%s] 更新素材使用次数异常", task_id, exc_info=True)
+
+                if gen_task:
+                    gen_task.append_log(
+                        "任务完成",
+                        f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}",
+                        duration=round(duration, 2),
+                        file_size=file_size,
+                        video_count=video_count,
+                    )
+                    _flush_logs(task_id, gen_task)
+
+                logger.info(
+                    "[task_id=%s] [任务完成] duration=%.2fs file_size=%d (edit_plan path)",
+                    task_id,
+                    duration,
+                    file_size,
+                )
+
+                return {
+                    "status": "completed",
+                    "task_id": task_id,
+                    "output_path": str(output_path),
+                    "file_size": file_size,
+                    "duration": duration,
+                    "mode": editing_mode.value,
+                }
+
+            finally:
+                # 无论任务成功或失败，都清理临时配音文件，避免磁盘泄漏
+                if voiceover_tmp_path:
+                    try:
+                        Path(voiceover_tmp_path).unlink(missing_ok=True)
+                    except OSError:
+                        logger.warning("[task_id=%s] 清理临时配音文件失败: %s", task_id, voiceover_tmp_path)
+
+        # DEPRECATED: 以下为旧路径，仅兼容无 source_edit_plan_id 的旧调用，后续移除
         with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir:
             temp_path = Path(temp_dir)
 
@@ -1369,10 +1680,27 @@ def generate_video(self, task_id: str) -> dict:
                             logger.info("[task_id=%s] MediaKit 视频理解完成: %d 个素材", task_id, len(asset_urls))
 
                     # 保存分析结果到 extra_meta
-                    if asset_analyses and gen_task:
-                        gen_task.extra_meta = {**(gen_task.extra_meta or {}), "asset_analyses": asset_analyses}
-                        _repo.update(gen_task)
-                        _flush_logs(task_id, gen_task)
+                    if asset_analyses:
+                        _meta_session = SessionLocal()
+                        try:
+                            from packages.adapters.sqlalchemy_impl.models import (
+                                GenerationTaskModel,
+                            )
+
+                            _m = (
+                                _meta_session.query(GenerationTaskModel)
+                                .filter(GenerationTaskModel.id == task_id)
+                                .first()
+                            )
+                            if _m:
+                                existing = dict(_m.extra_meta or {})
+                                existing["asset_analyses"] = asset_analyses
+                                _m.extra_meta = existing
+                                _meta_session.commit()
+                        finally:
+                            _meta_session.close()
+                        if gen_task:
+                            _flush_logs(task_id, gen_task)
                 except Exception:
                     logger.warning("[task_id=%s] MediaKit 视频理解失败，继续渲染", task_id, exc_info=True)
 
@@ -1466,10 +1794,10 @@ def generate_video(self, task_id: str) -> dict:
                             )
                             if _cover_model:
                                 _cover_model.cover_url = cover_frame_url
-                                # 持久化完整候选列表到 metadata
-                                meta = dict(_cover_model.metadata or {})
+                                # 持久化完整候选列表到 extra_meta
+                                meta = dict(_cover_model.extra_meta or {})
                                 meta["cover_candidates"] = cover_candidates
-                                _cover_model.metadata = meta
+                                _cover_model.extra_meta = meta
                                 _cover_session.commit()
                                 logger.info(
                                     "[task_id=%s] 封面帧已持久化(ffmpeg本地抽帧): cover_url=%s candidates=%d",

packages/shared/title_overlay.py

@@ -40,6 +40,21 @@ def find_title_font(size: int):
     return ImageFont.load_default()
 
 
+def _parse_hex_color(color: str, fallback=(255, 255, 255)) -> tuple[int, int, int]:
+    "将 #RRGGBB / #RGB 解析为 RGB 元组，失败返回 fallback。"
+    if not color or not isinstance(color, str):
+        return fallback
+    c = color.strip().lstrip("#")
+    try:
+        if len(c) == 6:
+            return (int(c[0:2], 16), int(c[2:4], 16), int(c[4:6], 16))
+        if len(c) == 3:
+            return (int(c[0] * 2, 16), int(c[1] * 2, 16), int(c[2] * 2, 16))
+    except (ValueError, IndexError):
+        pass
+    return fallback
+
+
 def wrap_title_text(text: str, font, max_width: int) -> list[str]:
     """按像素宽度对中英文混合文本自动换行，支持显式 \\n。"""
     lines: list[str] = []
@@ -71,6 +86,7 @@ def apply_title_to_image(
     image_path: str,
     title_text: str,
     *,
+    color: str = "#ffffff",
     position: str = "bottom",
     font_size: Optional[int] = None,
     margin_ratio: float = 0.06,
@@ -81,6 +97,7 @@ def apply_title_to_image(
     Args:
         image_path: 图片路径（处理结果覆盖写回）
         title_text: 标题文字；为空直接返回 None 表示跳过
+        color: 字体颜色（#RRGGBB），默认白色
         position: top / center / bottom
         font_size: 字号，None 时按图片宽度自动计算
         margin_ratio: 边缘留白占短边比例
@@ -109,6 +126,7 @@ def apply_title_to_image(
     if font is None:
         return None
 
+    text_rgb = _parse_hex_color(color)
     stroke_width = max(2, int(font_size * stroke_width_ratio))
     margin = int(min(img_w, img_h) * margin_ratio)
     max_text_width = img_w - 2 * margin
@@ -139,12 +157,12 @@ def apply_title_to_image(
         y = y_start + i * (line_height + line_gap)
         # 阴影
         draw.text((x + 2, y + 2), ln, font=font, fill=(0, 0, 0))
-        # 白色文字 + 黑色描边
+        # 文字（颜色由 color 参数控制）+ 黑色描边
         draw.text(
             (x, y),
             ln,
             font=font,
-            fill=(255, 255, 255),
+            fill=text_rgb,
             stroke_width=stroke_width,
             stroke_fill=(0, 0, 0),
         )

tests/unit/test_batch_clips_endpoint.py

@@ -0,0 +1,135 @@
+"""Tests for PUT /templates/{id}/editor/clips batch update endpoint.
+
+Updated for transactional replace_all_clips_transactional method.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+
+
+@pytest.fixture
+def mock_services():
+    plan_svc = MagicMock()
+    tpl_svc = MagicMock()
+    plan_svc.get_plan_or_raise.return_value = MagicMock(id="plan-1", template_id="tpl-1")
+    plan_svc.replace_all_clips_transactional.return_value = 2
+    return tpl_svc, plan_svc
+
+
+class TestBatchUpdateClips:
+    def test_batch_update_calls_transactional_replace(self, mock_services):
+        """验证批量更新调用事务性替换方法，传入正确的参数。"""
+        from app.api.routes.templates_editor.draft import batch_update_clips
+        from app.api.routes.templates_editor.schemas import (
+            EditorClipBatchItem,
+            EditorClipBatchUpdateRequest,
+        )
+
+        _, plan_svc = mock_services
+        req = EditorClipBatchUpdateRequest(
+            clips=[
+                EditorClipBatchItem(asset_id="a1", start_time=0.0, duration=3.0, order=0),
+                EditorClipBatchItem(asset_id="a2", start_time=3.0, duration=5.0, order=1),
+            ]
+        )
+
+        result = batch_update_clips(
+            template_id="tpl-1",
+            req=req,
+            plan_id="plan-1",
+            services=mock_services,
+            _=MagicMock(),
+        )
+
+        assert result.plan_id == "plan-1"
+        assert result.clip_count == 2
+        plan_svc.replace_all_clips_transactional.assert_called_once()
+        call_args = plan_svc.replace_all_clips_transactional.call_args
+        assert call_args[0][0] == "plan-1"
+        clips_data = call_args[0][1]
+        assert len(clips_data) == 2
+        assert clips_data[0]["asset_id"] == "a1"
+        assert clips_data[0]["start_time"] == 0.0
+        assert clips_data[0]["duration"] == 3.0
+        assert clips_data[1]["asset_id"] == "a2"
+
+    def test_batch_update_empty_clips(self, mock_services):
+        """空 clips 列表也能正常处理。"""
+        from app.api.routes.templates_editor.draft import batch_update_clips
+        from app.api.routes.templates_editor.schemas import EditorClipBatchUpdateRequest
+
+        _, plan_svc = mock_services
+        req = EditorClipBatchUpdateRequest(clips=[])
+
+        result = batch_update_clips(
+            template_id="tpl-1",
+            req=req,
+            plan_id="plan-1",
+            services=mock_services,
+            _=MagicMock(),
+        )
+
+        assert result.clip_count == 0
+        plan_svc.replace_all_clips_transactional.assert_called_once()
+        call_args = plan_svc.replace_all_clips_transactional.call_args
+        assert call_args[0][1] == []
+
+    def test_batch_update_passes_order_correctly(self, mock_services):
+        """验证 order 字段正确传递。"""
+        from app.api.routes.templates_editor.draft import batch_update_clips
+        from app.api.routes.templates_editor.schemas import (
+            EditorClipBatchItem,
+            EditorClipBatchUpdateRequest,
+        )
+
+        _, plan_svc = mock_services
+        req = EditorClipBatchUpdateRequest(
+            clips=[
+                EditorClipBatchItem(asset_id="a1", start_time=0.0, duration=3.0, order=5),
+            ]
+        )
+
+        batch_update_clips(
+            template_id="tpl-1",
+            req=req,
+            plan_id="plan-1",
+            services=mock_services,
+            _=MagicMock(),
+        )
+
+        clips_data = plan_svc.replace_all_clips_transactional.call_args[0][1]
+        assert clips_data[0]["order"] == 5
+        assert clips_data[0]["asset_id"] == "a1"
+        assert clips_data[0]["start_time"] == 0.0
+        assert clips_data[0]["duration"] == 3.0
+
+
+class TestEditorClipBatchItemValidation:
+    """验证 schema 校验规则。"""
+
+    def test_asset_id_empty_string_allowed(self):
+        """asset_id 空字符串允许通过（占位片段场景）。"""
+        from app.api.routes.templates_editor.schemas import EditorClipBatchItem
+
+        item = EditorClipBatchItem(asset_id="", start_time=0.0, duration=3.0, order=0)
+        assert item.asset_id == ""
+
+    def test_asset_id_valid(self):
+        """有效 asset_id 应通过校验。"""
+        from app.api.routes.templates_editor.schemas import EditorClipBatchItem
+
+        item = EditorClipBatchItem(asset_id="abc123", start_time=0.0, duration=3.0, order=0)
+        assert item.asset_id == "abc123"
+
+    def test_order_none_by_default(self):
+        """order 默认为 None，表示按数组顺序。"""
+        from app.api.routes.templates_editor.schemas import EditorClipBatchItem
+
+        item = EditorClipBatchItem(asset_id="a1", start_time=0.0, duration=3.0)
+        assert item.order is None

tests/unit/test_cover_url_finalize.py

@@ -1,132 +0,0 @@
-"""Tests for cover_url backfill to GenerationTask.
-
-Verifies _finalize_render_success correctly writes cover_url
-from cover_candidates to gen_task.cover_url.
-"""
-
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-# Add worker app to sys.path
-_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker"
-if str(_WORKER_ROOT) not in sys.path:
-    sys.path.insert(0, str(_WORKER_ROOT))
-
-
-class FakeGenTask:
-    """Simple stand-in for GenerationTask that tracks attribute assignment."""
-
-    def __init__(self):
-        object.__setattr__(self, "_assigned", {})
-        self.id = "task-1"
-        self.status = MagicMock()
-        self.status.value = "running"
-
-    def __setattr__(self, name, value):
-        if not name.startswith("_"):
-            self._assigned[name] = value
-        object.__setattr__(self, name, value)
-
-    def append_log(self, **kwargs):
-        pass
-
-
-def _make_plan():
-    plan = MagicMock()
-    plan.project_id = "proj-1"
-    plan.created_by_user_id = "user-1"
-    plan.config = {"batch_id": "batch-1", "mode": "edit_plan", "title": {"text": "test"}}
-    plan.mark_completed = MagicMock()
-    return plan
-
-
-def _call_finalize(cover_candidates=None, gen_task=None, plan=None):
-    from worker_app.tasks.edit_plan_generation import _finalize_render_success
-
-    plan = plan or _make_plan()
-    gen_task = gen_task or FakeGenTask()
-
-    plan_repo = MagicMock()
-    clip_repo = MagicMock()
-    gen_task_repo = MagicMock()
-    gen_task_repo.get.return_value = gen_task
-    db = MagicMock()
-
-    with patch("worker_app.tasks.edit_plan_generation.create_video_record_and_dedup"):
-        result = _finalize_render_success(
-            plan=plan,
-            plan_repo=plan_repo,
-            clip_repo=clip_repo,
-            gen_task_repo=gen_task_repo,
-            db=db,
-            plan_id="plan-1",
-            output_url="https://oss.example.com/output.mp4",
-            storage_key="rendered/plan-1/task-1.mp4",
-            duration=10.0,
-            file_size=1024,
-            width=1280,
-            height=720,
-            rendered_clip_ids=["clip-1"],
-            failed_clip_ids=[],
-            generation_task_id="task-1",
-            output_path=Path("/tmp/output.mp4"),
-            engine="unified",
-            thumbnail_url="",
-            cover_candidates=cover_candidates,
-        )
-
-    return result, gen_task, gen_task_repo
-
-
-class TestFinalizeCoverUrl:
-
-    def test_cover_url_set_from_image_url(self):
-        """cover_candidates with image_url should set gen_task.cover_url"""
-        candidates = [
-            {"image_url": "https://oss.example.com/cover1.jpg", "frame_time": 1.5},
-            {"image_url": "https://oss.example.com/cover2.jpg", "frame_time": 3.0},
-        ]
-        _, gen_task, gen_task_repo = _call_finalize(cover_candidates=candidates)
-        assert gen_task.cover_url == "https://oss.example.com/cover1.jpg"
-        gen_task_repo.update.assert_called()
-
-    def test_cover_url_fallback_to_url_key(self):
-        """Should fallback to 'url' key when 'image_url' is absent"""
-        candidates = [{"url": "https://oss.example.com/cover_url_key.jpg"}]
-        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
-        assert gen_task.cover_url == "https://oss.example.com/cover_url_key.jpg"
-
-    def test_cover_url_not_set_when_empty_list(self):
-        """Empty cover_candidates should not set cover_url"""
-        _, gen_task, _ = _call_finalize(cover_candidates=[])
-        assert "cover_url" not in gen_task._assigned
-
-    def test_cover_url_not_set_when_none(self):
-        """None cover_candidates should not set cover_url"""
-        _, gen_task, _ = _call_finalize(cover_candidates=None)
-        assert "cover_url" not in gen_task._assigned
-
-    def test_cover_url_not_set_when_url_empty(self):
-        """Empty URL strings in candidates should not set cover_url"""
-        candidates = [{"image_url": "", "url": ""}]
-        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
-        assert "cover_url" not in gen_task._assigned
-
-    def test_no_generation_task_no_crash(self):
-        """Should not crash when gen_task is None"""
-        candidates = [{"image_url": "https://oss.example.com/cover.jpg"}]
-        gen_task_repo = MagicMock()
-        gen_task_repo.get.return_value = None
-        result, _, _ = _call_finalize(cover_candidates=candidates)
-        assert result["status"] == "completed"
-
-    def test_image_url_priority_over_url(self):
-        """image_url should take priority over url key"""
-        candidates = [{"image_url": "https://a.jpg", "url": "https://b.jpg"}]
-        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
-        assert gen_task.cover_url == "https://a.jpg"

tests/unit/test_edit_plan_worker_failure.py

@@ -1,333 +0,0 @@
-"""P0-2: Celery 任务 render_edit_plan 失败时更新 GenerationTask 状态。
-
-验证：
-  - 异常发生时 GenerationTask 状态更新为 failed
-  - error_message 记录了异常类型和描述
-  - completed_at 被设置
-  - 即使 generation_task_id 为空也不崩溃
-  - 即使更新 GenerationTask 本身失败也不影响 retry
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from dataclasses import dataclass, field
-from types import ModuleType
-from typing import Any, Optional
-from unittest.mock import MagicMock, patch
-
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-
-# ── Mock worker 模块以避免数据库连接 ──────────────────────────────────────────
-# worker_app.db 在 import 时会尝试连接数据库，必须在导入 task 模块前 mock
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-
-# 预注册 mock 模块，阻止真实数据库初始化
-_mock_db_mod = ModuleType("worker_app.db")
-_mock_db_mod.SessionLocal = MagicMock()
-sys.modules.setdefault("worker_app.db", _mock_db_mod)
-
-_mock_celery_mod = ModuleType("worker_app.celery_app")
-_mock_celery_app = MagicMock()
-# 让 @celery_app.task(...) 装饰器透传原始函数，否则函数变成 MagicMock
-_mock_celery_app.task = lambda **kwargs: lambda fn: fn
-_mock_celery_mod.celery_app = _mock_celery_app
-sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod)
-
-
-# ── Stub domain objects ───────────────────────────────────────────────────────
-
-
-@dataclass
-class _StubStatus:
-    value: str
-
-    def __eq__(self, other):
-        if isinstance(other, str):
-            return self.value == other
-        if isinstance(other, _StubStatus):
-            return self.value == other.value
-        return NotImplemented
-
-
-@dataclass
-class StubEditPlan:
-    id: str = "plan-001"
-    template_id: str = "tmpl-001"
-    status: Any = None
-    config: dict = field(default_factory=dict)
-    project_id: str = ""
-    created_by_user_id: str = "user-001"
-
-    def mark_failed(self):
-        self.status = _StubStatus("failed")
-
-    def mark_completed(self):
-        self.status = _StubStatus("completed")
-
-
-@dataclass
-class StubGenerationTask:
-    id: str = "gen-task-001"
-    status: Any = field(default_factory=lambda: _StubStatus("pending"))
-    error_message: str = ""
-    progress: float = 0.0
-    result_count: int = 0
-    started_at: Any = None
-    completed_at: Any = None
-    project_id: str = ""
-    created_by_user_id: str = "user-001"
-
-
-@dataclass
-class StubClip:
-    id: str = "clip-001"
-    plan_id: str = "plan-001"
-    asset_id: str = "assets/video.mp4"
-    order: int = 1
-    status: Any = field(default_factory=lambda: _StubStatus("ready"))
-    transition_effect: str = ""
-    text_content: str = ""
-    clip_type: str = "MAIN"
-    duration: float = 0.0
-
-    def mark_failed(self):
-        self.status = _StubStatus("failed")
-
-    def mark_rendered(self):
-        self.status = _StubStatus("rendered")
-
-
-# ── Stub repositories ─────────────────────────────────────────────────────────
-
-
-class StubPlanRepo:
-    def __init__(self, plan: StubEditPlan):
-        self._plan = plan
-
-    def get(self, plan_id: str) -> Optional[StubEditPlan]:
-        if plan_id == self._plan.id:
-            return self._plan
-        return None
-
-    def update(self, plan: StubEditPlan) -> StubEditPlan:
-        self._plan = plan
-        return plan
-
-
-class StubClipRepo:
-    def __init__(self, clips: list[StubClip] | None = None):
-        self._clips = clips or []
-
-    def list_by_plan(self, plan_id: str, skip: int = 0, limit: int = 10000) -> list[StubClip]:
-        return [c for c in self._clips if c.plan_id == plan_id]
-
-    def get(self, clip_id: str) -> Optional[StubClip]:
-        for c in self._clips:
-            if c.id == clip_id:
-                return c
-        return None
-
-    def update(self, clip: StubClip) -> StubClip:
-        return clip
-
-
-class StubGenTaskRepo:
-    def __init__(self, task: StubGenerationTask | None = None):
-        self._store: dict[str, StubGenerationTask] = {}
-        if task:
-            self._store[task.id] = task
-
-    def get(self, task_id: str) -> Optional[StubGenerationTask]:
-        return self._store.get(task_id)
-
-    def update(self, task: StubGenerationTask) -> StubGenerationTask:
-        self._store[task.id] = task
-        return task
-
-
-# ── Import task module (after mocks are in place) ─────────────────────────────
-
-from worker_app.tasks.edit_plan_generation import render_edit_plan
-
-# ── Tests ─────────────────────────────────────────────────────────────────────
-
-
-class TestRenderEditPlanFailureUpdatesGenTask:
-    """P0-2: render_edit_plan 异常时更新 GenerationTask 状态为 failed"""
-
-    def _make_bound_task(self):
-        """构建绑定的 Celery task mock"""
-        task = MagicMock()
-        task.retry = MagicMock(side_effect=RuntimeError("retry called"))
-        return task
-
-    def test_exception_marks_gen_task_failed(self):
-        """异常时 GenerationTask.status 被设为 failed"""
-        plan = StubEditPlan(status=_StubStatus("rendering"))
-        plan.config["generation_task_id"] = "gen-task-001"
-        gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
-
-        plan_repo = StubPlanRepo(plan)
-        gen_task_repo = StubGenTaskRepo(gen_task)
-
-        # 让 clip_repo 抛异常以触发 except 路径
-        clip_repo_bad = MagicMock()
-        clip_repo_bad.list_by_plan.side_effect = RuntimeError("OSS 连接失败")
-
-        def fake_get_repos():
-            yield plan_repo, clip_repo_bad, gen_task_repo, MagicMock()
-
-        bound_task = self._make_bound_task()
-
-        with patch(
-            "worker_app.tasks.edit_plan_generation._get_repos",
-            side_effect=fake_get_repos,
-        ):
-            with pytest.raises(RuntimeError, match="retry called"):
-                render_edit_plan(bound_task, "plan-001")
-
-        # 核心断言：GenerationTask 状态为 failed（生产代码赋值为字符串）
-        assert gen_task.status == "failed"
-
-    def test_exception_records_error_message(self):
-        """异常时 error_message 包含异常类型和描述"""
-        plan = StubEditPlan(status=_StubStatus("rendering"))
-        plan.config["generation_task_id"] = "gen-task-001"
-        gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
-
-        plan_repo = StubPlanRepo(plan)
-        clip_repo = MagicMock()
-        clip_repo.list_by_plan.side_effect = RuntimeError("DB 查询超时")
-        gen_task_repo = StubGenTaskRepo(gen_task)
-
-        def fake_get_repos():
-            yield plan_repo, clip_repo, gen_task_repo, MagicMock()
-
-        bound_task = self._make_bound_task()
-
-        with patch(
-            "worker_app.tasks.edit_plan_generation._get_repos",
-            side_effect=fake_get_repos,
-        ):
-            with pytest.raises(RuntimeError, match="retry called"):
-                render_edit_plan(bound_task, "plan-001")
-
-        assert gen_task.status == "failed"
-        assert "DB 查询超时" in gen_task.error_message
-        assert "RuntimeError" in gen_task.error_message
-
-    def test_exception_sets_completed_at(self):
-        """异常时 completed_at 被设置"""
-        plan = StubEditPlan(status=_StubStatus("rendering"))
-        plan.config["generation_task_id"] = "gen-task-001"
-        gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
-
-        plan_repo = StubPlanRepo(plan)
-        clip_repo = MagicMock()
-        clip_repo.list_by_plan.side_effect = RuntimeError("boom")
-        gen_task_repo = StubGenTaskRepo(gen_task)
-
-        def fake_get_repos():
-            yield plan_repo, clip_repo, gen_task_repo, MagicMock()
-
-        bound_task = self._make_bound_task()
-
-        with patch(
-            "worker_app.tasks.edit_plan_generation._get_repos",
-            side_effect=fake_get_repos,
-        ):
-            with pytest.raises(RuntimeError):
-                render_edit_plan(bound_task, "plan-001")
-
-        assert gen_task.completed_at is not None
-
-    def test_no_generation_task_id_does_not_crash(self):
-        """generation_task_id 为空时，异常处理不崩溃"""
-        plan = StubEditPlan(status=_StubStatus("rendering"))
-        plan.config = {}  # 不设置 generation_task_id
-
-        plan_repo = StubPlanRepo(plan)
-        clip_repo = MagicMock()
-        clip_repo.list_by_plan.side_effect = RuntimeError("boom")
-        gen_task_repo = StubGenTaskRepo()  # 空 repo
-
-        def fake_get_repos():
-            yield plan_repo, clip_repo, gen_task_repo, MagicMock()
-
-        bound_task = self._make_bound_task()
-
-        with patch(
-            "worker_app.tasks.edit_plan_generation._get_repos",
-            side_effect=fake_get_repos,
-        ):
-            with pytest.raises(RuntimeError, match="retry called"):
-                render_edit_plan(bound_task, "plan-001")
-
-        # 计划仍被标记为 failed
-        assert plan.status.value == "failed"
-
-    def test_gen_task_update_failure_does_not_block_retry(self):
-        """更新 GenerationTask 失败时，不影响 retry 流程"""
-        plan = StubEditPlan(status=_StubStatus("rendering"))
-        plan.config["generation_task_id"] = "gen-task-001"
-        gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
-
-        plan_repo = StubPlanRepo(plan)
-        clip_repo = MagicMock()
-        clip_repo.list_by_plan.side_effect = RuntimeError("原始错误")
-        # gen_task_repo.update 也抛异常
-        gen_task_repo = MagicMock()
-        gen_task_repo.get.return_value = gen_task
-        gen_task_repo.update.side_effect = RuntimeError("DB 写入失败")
-
-        def fake_get_repos():
-            yield plan_repo, clip_repo, gen_task_repo, MagicMock()
-
-        bound_task = self._make_bound_task()
-
-        with patch(
-            "worker_app.tasks.edit_plan_generation._get_repos",
-            side_effect=fake_get_repos,
-        ):
-            with pytest.raises(RuntimeError, match="retry called"):
-                render_edit_plan(bound_task, "plan-001")
-
-        # retry 被调用说明流程正确
-        bound_task.retry.assert_called_once()
-
-    def test_already_failed_gen_task_not_overwritten(self):
-        """已经 failed 的 GenerationTask 不会被重复更新"""
-        plan = StubEditPlan(status=_StubStatus("rendering"))
-        plan.config["generation_task_id"] = "gen-task-001"
-        gen_task = StubGenerationTask(
-            id="gen-task-001",
-            status=_StubStatus("failed"),  # 已经是 failed
-            error_message="之前的错误",
-        )
-
-        plan_repo = StubPlanRepo(plan)
-        clip_repo = MagicMock()
-        clip_repo.list_by_plan.side_effect = RuntimeError("新错误")
-        gen_task_repo = StubGenTaskRepo(gen_task)
-
-        def fake_get_repos():
-            yield plan_repo, clip_repo, gen_task_repo, MagicMock()
-
-        bound_task = self._make_bound_task()
-
-        with patch(
-            "worker_app.tasks.edit_plan_generation._get_repos",
-            side_effect=fake_get_repos,
-        ):
-            with pytest.raises(RuntimeError, match="retry called"):
-                render_edit_plan(bound_task, "plan-001")
-
-        # error_message 应保持原值，不被覆盖
-        assert gen_task.error_message == "之前的错误"

tests/unit/test_fallback_user_assets.py

@@ -1,331 +0,0 @@
-"""Unit tests for PR #1338: 确认生成兜底增强 — user_id 查找素材.
-
-覆盖:
-- SQLAlchemyAssetRepository.find_ready_videos_by_user
-- _auto_fallback_auto_material_mode 策略2 (user_id 兜底)
-"""
-
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
-sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "packages"))
-
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-
-from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
-from packages.adapters.sqlalchemy_impl.models import Base
-from packages.domain import Asset, AssetStatus
-
-
-def _make_repo():
-    engine = create_engine("sqlite:///:memory:")
-    Base.metadata.create_all(engine)
-    session = sessionmaker(bind=engine)()
-    return SQLAlchemyAssetRepository(session)
-
-
-class TestFindReadyVideosByUser:
-    """SQLAlchemyAssetRepository.find_ready_videos_by_user 测试."""
-
-    def test_returns_ready_videos_for_user(self):
-        repo = _make_repo()
-        user_id = "user-abc-123"
-        v1 = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="video1.mp4",
-            storage_key="v/v1.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.READY,
-            uploaded_by_user_id=user_id,
-        )
-        v2 = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="video2.mp4",
-            storage_key="v/v2.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.READY,
-            uploaded_by_user_id=user_id,
-        )
-        repo.create(v1)
-        repo.create(v2)
-        result = repo.find_ready_videos_by_user(user_id)
-        assert len(result) == 2
-        assert {a.id for a in result} == {v1.id, v2.id}
-
-    def test_excludes_non_video_assets(self):
-        repo = _make_repo()
-        user_id = "user-abc-123"
-        video = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="clip.mp4",
-            storage_key="v/clip.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.READY,
-            uploaded_by_user_id=user_id,
-        )
-        image = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="photo.jpg",
-            storage_key="v/photo.jpg",
-            mime_type="image/jpeg",
-            status=AssetStatus.READY,
-            uploaded_by_user_id=user_id,
-        )
-        repo.create(video)
-        repo.create(image)
-        result = repo.find_ready_videos_by_user(user_id)
-        assert len(result) == 1
-        assert result[0].id == video.id
-
-    def test_excludes_non_ready_assets(self):
-        repo = _make_repo()
-        user_id = "user-abc-123"
-        ready = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="ready.mp4",
-            storage_key="v/ready.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.READY,
-            uploaded_by_user_id=user_id,
-        )
-        uploading = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="uploading.mp4",
-            storage_key="v/uploading.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.UPLOADING,
-            uploaded_by_user_id=user_id,
-        )
-        repo.create(ready)
-        repo.create(uploading)
-        result = repo.find_ready_videos_by_user(user_id)
-        assert len(result) == 1
-        assert result[0].id == ready.id
-
-    def test_excludes_other_users_assets(self):
-        repo = _make_repo()
-        my_video = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="mine.mp4",
-            storage_key="v/mine.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.READY,
-            uploaded_by_user_id="user-A",
-        )
-        other_video = Asset.create(
-            project_id="",
-            library_id="lib-1",
-            name="other.mp4",
-            storage_key="v/other.mp4",
-            mime_type="video/mp4",
-            status=AssetStatus.READY,
-            uploaded_by_user_id="user-B",
-        )
-        repo.create(my_video)
-        repo.create(other_video)
-        result = repo.find_ready_videos_by_user("user-A")
-        assert len(result) == 1
-        assert result[0].id == my_video.id
-
-    def test_empty_result_for_unknown_user(self):
-        repo = _make_repo()
-        result = repo.find_ready_videos_by_user("nonexistent-user")
-        assert result == []
-
-    def test_respects_limit(self):
-        repo = _make_repo()
-        user_id = "user-abc-123"
-        for i in range(10):
-            asset = Asset.create(
-                project_id="",
-                library_id="lib-1",
-                name=f"video_{i}.mp4",
-                storage_key=f"v/v{i}.mp4",
-                mime_type="video/mp4",
-                status=AssetStatus.READY,
-                uploaded_by_user_id=user_id,
-            )
-            repo.create(asset)
-        result = repo.find_ready_videos_by_user(user_id, limit=3)
-        assert len(result) == 3
-
-
-class TestAutoFallbackAutoMaterialModeUserId:
-    """_auto_fallback_auto_material_mode user_id 兜底策略测试."""
-
-    def _make_plan_check(self, project_id="", template_id="tmpl-1"):
-        plan = MagicMock()
-        plan.project_id = project_id
-        plan.template_id = template_id
-        plan.config = {}
-        return plan
-
-    def _make_clip(self, clip_id="clip-1"):
-        clip = MagicMock()
-        clip.id = clip_id
-        clip.asset_id = ""
-        return clip
-
-    def test_skips_when_no_clips_without_asset(self):
-        from app.api.routes.templates_editor._fallback import (
-            _auto_fallback_auto_material_mode,
-        )
-
-        svc = MagicMock()
-        plan_check = self._make_plan_check()
-        _auto_fallback_auto_material_mode(
-            svc,
-            "plan-1",
-            plan_check,
-            [],
-            MagicMock(),
-            MagicMock(),
-            user_id="user-1",
-        )
-        svc.assign_asset.assert_not_called()
-
-    def test_strategy2_user_id_fallback(self):
-        from app.api.routes.templates_editor._fallback import (
-            _auto_fallback_auto_material_mode,
-        )
-
-        svc = MagicMock()
-        plan_check = self._make_plan_check(project_id="")
-        clip = self._make_clip("clip-1")
-        mock_asset = MagicMock()
-        mock_asset.id = "asset-from-user"
-        mock_asset.status = AssetStatus.READY
-        mock_asset.mime_type = "video/mp4"
-        asset_repo = MagicMock()
-        asset_repo.find_ready_videos_by_user.return_value = [mock_asset]
-        asset_library_repo = MagicMock()
-        _auto_fallback_auto_material_mode(
-            svc,
-            "plan-1",
-            plan_check,
-            [clip],
-            asset_library_repo,
-            asset_repo,
-            user_id="user-123",
-        )
-        asset_repo.find_ready_videos_by_user.assert_called_once_with("user-123")
-        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-user")
-
-    def test_strategy1_takes_priority_over_strategy2(self):
-        from app.api.routes.templates_editor._fallback import (
-            _auto_fallback_auto_material_mode,
-        )
-
-        svc = MagicMock()
-        plan_check = self._make_plan_check(project_id="proj-1")
-        clip = self._make_clip("clip-1")
-        mock_lib = MagicMock()
-        mock_lib.id = "lib-video"
-        mock_lib.kind = MagicMock()
-        mock_lib.kind.value = "video"
-        mock_asset = MagicMock()
-        mock_asset.id = "asset-from-project"
-        mock_asset.status = "ready"
-        mock_asset.mime_type = "video/mp4"
-        asset_library_repo = MagicMock()
-        asset_library_repo.find_by_project.return_value = [mock_lib]
-        asset_repo = MagicMock()
-        asset_repo.find_by_library.return_value = [mock_asset]
-        _auto_fallback_auto_material_mode(
-            svc,
-            "plan-1",
-            plan_check,
-            [clip],
-            asset_library_repo,
-            asset_repo,
-            user_id="user-123",
-        )
-        asset_library_repo.find_by_project.assert_called_once_with("proj-1")
-        asset_repo.find_ready_videos_by_user.assert_not_called()
-        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-project")
-
-    def test_falls_back_when_project_has_no_videos(self):
-        from app.api.routes.templates_editor._fallback import (
-            _auto_fallback_auto_material_mode,
-        )
-
-        svc = MagicMock()
-        plan_check = self._make_plan_check(project_id="proj-1")
-        clip = self._make_clip("clip-1")
-        asset_library_repo = MagicMock()
-        asset_library_repo.find_by_project.return_value = []
-        mock_asset = MagicMock()
-        mock_asset.id = "asset-from-user"
-        mock_asset.status = AssetStatus.READY
-        mock_asset.mime_type = "video/mp4"
-        asset_repo = MagicMock()
-        asset_repo.find_ready_videos_by_user.return_value = [mock_asset]
-        _auto_fallback_auto_material_mode(
-            svc,
-            "plan-1",
-            plan_check,
-            [clip],
-            asset_library_repo,
-            asset_repo,
-            user_id="user-123",
-        )
-        asset_repo.find_ready_videos_by_user.assert_called_once_with("user-123")
-        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-user")
-
-    def test_no_assets_found_does_nothing(self):
-        from app.api.routes.templates_editor._fallback import (
-            _auto_fallback_auto_material_mode,
-        )
-
-        svc = MagicMock()
-        plan_check = self._make_plan_check(project_id="")
-        clip = self._make_clip("clip-1")
-        asset_repo = MagicMock()
-        asset_repo.find_ready_videos_by_user.return_value = []
-        asset_library_repo = MagicMock()
-        _auto_fallback_auto_material_mode(
-            svc,
-            "plan-1",
-            plan_check,
-            [clip],
-            asset_library_repo,
-            asset_repo,
-            user_id="user-123",
-        )
-        svc.assign_asset.assert_not_called()
-
-    def test_no_user_id_skips_strategy2(self):
-        from app.api.routes.templates_editor._fallback import (
-            _auto_fallback_auto_material_mode,
-        )
-
-        svc = MagicMock()
-        plan_check = self._make_plan_check(project_id="")
-        clip = self._make_clip("clip-1")
-        asset_repo = MagicMock()
-        asset_library_repo = MagicMock()
-        _auto_fallback_auto_material_mode(
-            svc,
-            "plan-1",
-            plan_check,
-            [clip],
-            asset_library_repo,
-            asset_repo,
-            user_id="",
-        )
-        asset_repo.find_ready_videos_by_user.assert_not_called()
-        svc.assign_asset.assert_not_called()

tests/unit/test_generate_title_and_cover.py

@@ -1,369 +0,0 @@
-"""Tests for /generate endpoint — custom_title and cover_url passing."""
-
-from __future__ import annotations
-
-import json
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-
-class TestGenerateEndpointTitleAndCover:
-    """测试 /generate 端点传递 custom_title 和 cover_url。"""
-
-    def test_generate_passes_cover_url_from_plan_config(self):
-        """从 plan.config.cover.image_url 读取封面 URL 传递给生成任务。"""
-        from app.api.routes.templates_editor.generation import generate_editor_draft
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        mock_plan = MagicMock()
-        mock_plan.id = "plan-123"
-        mock_plan.project_id = "project-1"
-        mock_plan.template_id = "template-1"
-        mock_plan.status = MagicMock(value="editing")
-        mock_plan.config = {
-            "clips": [{"id": "c1"}],
-            "asset_ids": ["a1"],
-            "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/cover.jpg"},
-        }
-        mock_plan.updated_at = None
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_plan_svc.can_generate.return_value = (True, "")
-        mock_plan_svc.mark_clips_ready.return_value = 1
-        mock_template_svc = MagicMock()
-
-        mock_gen_task = MagicMock()
-        mock_gen_task.id = "task-new"
-        mock_gen_task.project_id = "project-1"
-
-        mock_current_user = MagicMock()
-        mock_current_user.user.id = "user-1"
-
-        body = EditPlanGenerateRequest()  # No title_config
-
-        with (
-            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
-            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
-            patch("app.api.routes.templates_editor.generation.celery_app"),
-            patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-123"),
-        ):
-            mock_repo = MagicMock()
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = mock_gen_task
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_plan_svc.transition_status = MagicMock()
-            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-            result = generate_editor_draft(
-                template_id="template-1",
-                request=body,
-                plan_id="plan-123",
-                services=(mock_template_svc, mock_plan_svc),
-                db=MagicMock(),
-                current_user=mock_current_user,
-                asset_library_repo=MagicMock(),
-                asset_repo=MagicMock(),
-            )
-
-            # Verify cover_url was passed to CreateGenerationTaskCommand
-            call_args = mock_usecase.execute.call_args
-            command = call_args[0][0]
-            assert command.cover_url == "https://oss.example.com/uploaded/cover.jpg"
-            assert command.custom_title == ""
-
-    def test_generate_passes_custom_title_from_title_config(self):
-        """前端传 title_config 时，序列化为 JSON 存入 custom_title。"""
-        from app.api.routes.templates_editor.generation import generate_editor_draft
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        mock_plan = MagicMock()
-        mock_plan.id = "plan-456"
-        mock_plan.project_id = "project-1"
-        mock_plan.template_id = "template-1"
-        mock_plan.status = MagicMock(value="editing")
-        mock_plan.config = {
-            "clips": [{"id": "c1"}],
-            "asset_ids": ["a1"],
-            "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/cover.jpg"},
-        }
-        mock_plan.updated_at = None
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_plan_svc.can_generate.return_value = (True, "")
-        mock_plan_svc.mark_clips_ready.return_value = 1
-        mock_template_svc = MagicMock()
-
-        mock_gen_task = MagicMock()
-        mock_gen_task.id = "task-title"
-
-        mock_current_user = MagicMock()
-        mock_current_user.user.id = "user-1"
-
-        title_config = {
-            "text": "测试标题",
-            "font_size": 36,
-            "font_color": "#ffffff",
-            "position": "center",
-        }
-        body = EditPlanGenerateRequest(title_config=title_config)
-
-        with (
-            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
-            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
-            patch("app.api.routes.templates_editor.generation.celery_app"),
-        ):
-            mock_repo = MagicMock()
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = mock_gen_task
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_plan_svc.transition_status = MagicMock()
-            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-            result = generate_editor_draft(
-                template_id="template-1",
-                request=body,
-                plan_id="plan-456",
-                services=(mock_template_svc, mock_plan_svc),
-                db=MagicMock(),
-                current_user=mock_current_user,
-                asset_library_repo=MagicMock(),
-                asset_repo=MagicMock(),
-            )
-
-            # Verify custom_title was serialized to JSON
-            call_args = mock_usecase.execute.call_args
-            command = call_args[0][0]
-            parsed_title = json.loads(command.custom_title)
-            assert parsed_title["text"] == "测试标题"
-            assert parsed_title["font_size"] == 36
-            assert command.cover_url == "https://oss.example.com/cover.jpg"
-
-    def test_generate_empty_title_config_passes_empty_custom_title(self):
-        """title_config 为空时 custom_title 为空字符串。"""
-        from app.api.routes.templates_editor.generation import generate_editor_draft
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        mock_plan = MagicMock()
-        mock_plan.id = "plan-789"
-        mock_plan.project_id = "project-1"
-        mock_plan.template_id = "template-1"
-        mock_plan.status = MagicMock(value="editing")
-        mock_plan.config = {"clips": [{"id": "c1"}], "asset_ids": ["a1"]}
-        mock_plan.updated_at = None
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_plan_svc.can_generate.return_value = (True, "")
-        mock_plan_svc.mark_clips_ready.return_value = 1
-        mock_template_svc = MagicMock()
-
-        mock_gen_task = MagicMock()
-        mock_gen_task.id = "task-no-title"
-
-        body = EditPlanGenerateRequest()  # No title_config
-
-        with (
-            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
-            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
-            patch("app.api.routes.templates_editor.generation.celery_app"),
-        ):
-            mock_repo = MagicMock()
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = mock_gen_task
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_plan_svc.transition_status = MagicMock()
-            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-            result = generate_editor_draft(
-                template_id="template-1",
-                request=body,
-                plan_id="plan-789",
-                services=(mock_template_svc, mock_plan_svc),
-                db=MagicMock(),
-                current_user=MagicMock(),
-                asset_library_repo=MagicMock(),
-                asset_repo=MagicMock(),
-            )
-
-            call_args = mock_usecase.execute.call_args
-            command = call_args[0][0]
-            assert command.custom_title == ""
-
-
-class TestGenerateEndpointRequestSchema:
-    """测试 EditPlanGenerateRequest schema。"""
-
-    def test_schema_default_empty_title_config(self):
-        """默认 title_config 为空 dict。"""
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        req = EditPlanGenerateRequest()
-        assert req.title_config == {}
-
-    def test_schema_accepts_title_config(self):
-        """可以传入标题配置。"""
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        req = EditPlanGenerateRequest(title_config={"text": "我的标题", "font_size": 48})
-        assert req.title_config["text"] == "我的标题"
-        assert req.title_config["font_size"] == 48
-
-
-class TestGenerateTitleChangeSkipsReuse:
-    """测试标题变更时跳过预览产物复用。"""
-
-    def _make_mocks(self, custom_title=""):
-        mock_plan = MagicMock()
-        mock_plan.id = "plan-reuse"
-        mock_plan.project_id = "project-1"
-        mock_plan.template_id = "template-1"
-        mock_plan.status = MagicMock(value="editing")
-        mock_plan.config = {"clips": [{"id": "c1"}], "asset_ids": ["a1"]}
-        mock_plan.updated_at = None
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_plan_svc.can_generate.return_value = (True, "")
-        mock_plan_svc.mark_clips_ready.return_value = 1
-        mock_template_svc = MagicMock()
-
-        reusable_task = MagicMock()
-        reusable_task.id = "task-reusable"
-        reusable_task.is_completed = True
-        reusable_task.is_preview = True
-        reusable_task.custom_title = custom_title
-        reusable_task.project_id = "project-1"
-        reusable_task.source_edit_plan_id = "plan-reuse"
-
-        mock_new_task = MagicMock()
-        mock_new_task.id = "task-new"
-
-        return mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task
-
-    def test_title_removed_skips_reuse(self):
-        """原来有标题，现在移除了 → 跳过复用，创建新任务。"""
-        from app.api.routes.templates_editor.generation import generate_editor_draft
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task = self._make_mocks(
-            custom_title='{"text": "旧标题"}'
-        )
-
-        body = EditPlanGenerateRequest()  # No title_config → title removed
-
-        with (
-            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=reusable_task),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
-            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
-            patch("app.api.routes.templates_editor.generation.celery_app"),
-            patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-reuse"),
-        ):
-            mock_repo = MagicMock()
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = mock_new_task
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_plan_svc.transition_status = MagicMock()
-            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-            result = generate_editor_draft(
-                template_id="template-1",
-                request=body,
-                plan_id="plan-reuse",
-                services=(mock_template_svc, mock_plan_svc),
-                db=MagicMock(),
-                current_user=MagicMock(),
-                asset_library_repo=MagicMock(),
-                asset_repo=MagicMock(),
-            )
-
-            # 应该创建新任务而不是复用
-            mock_usecase.execute.assert_called_once()
-            # 不应该 mark_confirmed 在 reusable_task 上
-            reusable_task.mark_confirmed.assert_not_called()
-
-    def test_title_changed_skips_reuse(self):
-        """标题变更 → 跳过复用。"""
-        import json
-
-        from app.api.routes.templates_editor.generation import generate_editor_draft
-        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
-
-        mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task = self._make_mocks(
-            custom_title=json.dumps({"text": "旧标题", "font_size": 36}, ensure_ascii=False)
-        )
-
-        body = EditPlanGenerateRequest(title_config={"text": "新标题", "font_size": 48})
-
-        with (
-            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=reusable_task),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
-            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
-            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
-            patch("app.api.routes.templates_editor.generation.celery_app"),
-            patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-reuse"),
-        ):
-            mock_repo = MagicMock()
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = mock_new_task
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_plan_svc.transition_status = MagicMock()
-            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-            result = generate_editor_draft(
-                template_id="template-1",
-                request=body,
-                plan_id="plan-reuse",
-                services=(mock_template_svc, mock_plan_svc),
-                db=MagicMock(),
-                current_user=MagicMock(),
-                asset_library_repo=MagicMock(),
-                asset_repo=MagicMock(),
-            )
-
-            mock_usecase.execute.assert_called_once()
-            reusable_task.mark_confirmed.assert_not_called()

tests/unit/test_generation_cover.py

@@ -987,6 +987,159 @@ class TestUploadCoverType:
             # 标题文字必须透传给持久化函数（用于源素材帧叠加标题）
             assert mock_persist.call_args.kwargs.get("title_text") == "我的视频标题"
 
+    def test_e2_passes_full_title_style_to_persist(self):
+        """步骤E2：plan.config.title 包含完整样式时，color/position/font_size 都传给 _persist_cover_frame。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "title": {
+                "enabled": True,
+                "text": "样式标题",
+                "color": "#00ff00",
+                "position": "top",
+                "font_size": 42,
+            }
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+        mock_db = MagicMock()
+
+        mock_asset = MagicMock()
+        mock_asset.file_type = "video"
+        mock_asset.storage_key = "uploads/src.mp4"
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = mock_asset
+
+        mock_mk = MagicMock()
+        mock_mk.is_available = True
+        mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]
+
+        mock_storage = MagicMock()
+        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"
+
+        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                return_value=mock_asset_repo,
+            ),
+            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
+            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
+            patch(
+                "app.api.routes.generation_cover._persist_cover_frame",
+                return_value="https://oss.example.com/covers/styled.jpg",
+            ) as mock_persist,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/styled.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="tpl",
+                plan_id="plan-style",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            assert result.cover["image_url"] == "https://oss.example.com/covers/styled.jpg"
+            kwargs = mock_persist.call_args.kwargs
+            assert kwargs["title_text"] == "样式标题"
+            assert kwargs["title_color"] == "#00ff00"
+            assert kwargs["title_position"] == "top"
+            assert kwargs["title_font_size"] == 42
+
+    def test_e2_title_style_fallback_font_color(self):
+        """步骤E2：前端传 font_color 时能正确兼容读取。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "title": {
+                "enabled": True,
+                "text": "兼容标题",
+                "font_color": "#123456",
+                "position": "center",
+            }
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+        mock_db = MagicMock()
+
+        mock_asset = MagicMock()
+        mock_asset.file_type = "video"
+        mock_asset.storage_key = "uploads/src.mp4"
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = mock_asset
+
+        mock_mk = MagicMock()
+        mock_mk.is_available = True
+        mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]
+
+        mock_storage = MagicMock()
+        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"
+
+        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                return_value=mock_asset_repo,
+            ),
+            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
+            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
+            patch(
+                "app.api.routes.generation_cover._persist_cover_frame",
+                return_value="https://oss.example.com/covers/compat.jpg",
+            ) as mock_persist,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/compat.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            generate_cover(
+                body=body,
+                template_id="tpl",
+                plan_id="plan-compat",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            kwargs = mock_persist.call_args.kwargs
+            assert kwargs["title_color"] == "#123456"
+            assert kwargs["title_position"] == "center"
+            assert kwargs["title_font_size"] is None
+
     def test_step_e_skips_non_video_assets(self):
         """步骤E2：asset_ids 里只有图片素材时，不调用 MediaKit 并返回 400。"""
         from unittest.mock import MagicMock, patch

tests/unit/test_replace_all_clips_transactional.py

@@ -0,0 +1,147 @@
+"""Tests for EditPlanService.replace_all_clips_transactional."""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, PropertyMock, patch
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+
+
+class TestReplaceAllClipsTransactional:
+    """事务性替换片段方法测试。"""
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_success_commits_once(self, mock_clip_cls, mock_model_cls):
+        """成功时单次 commit，不 rollback。"""
+        from app.services.edit_plan_service import EditPlanService
+
+        db = MagicMock()
+        # Mock query chain for delete
+        query_mock = MagicMock()
+        query_mock.filter.return_value.delete.return_value = 3
+        db.query.return_value = query_mock
+
+        # Mock query chain for mark_ready (pending_with_asset)
+        # After the create loop, query returns empty list (no pending clips with asset)
+        ready_query = MagicMock()
+        ready_query.filter.return_value.filter.return_value.filter.return_value.all.return_value = []
+        db.query.side_effect = [query_mock, ready_query]
+
+        # Mock EditPlanClip.create to return a mock entity
+        mock_entity = MagicMock()
+        mock_entity.id = "clip-1"
+        mock_entity.plan_id = "plan-1"
+        mock_entity.clip_type = "main"
+        mock_entity.order = 0
+        mock_entity.asset_id = "asset-1"
+        mock_entity.text_content = ""
+        mock_entity.start_time = 0.0
+        mock_entity.duration = 3.0
+        mock_entity.transition_effect = "cut"
+        mock_entity.transition_duration = 0.0
+        mock_entity.playback_speed = 1.0
+        mock_entity.status.value = "pending"
+        mock_entity.config = {}
+        mock_clip_cls.create.return_value = mock_entity
+
+        # Mock the model constructor
+        mock_model_instance = MagicMock()
+        mock_model_cls.return_value = mock_model_instance
+
+        # Mock clip_repo
+        clip_repo = MagicMock()
+        clip_repo.session = db
+
+        svc = EditPlanService.__new__(EditPlanService)
+        svc._clip_repo = clip_repo
+
+        result = svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        assert result == 1
+        db.commit.assert_called_once()
+        db.rollback.assert_not_called()
+        db.add.assert_called_once_with(mock_model_instance)
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_failure_rolls_back(self, mock_clip_cls, mock_model_cls):
+        """异常时自动 rollback。"""
+        from app.services.edit_plan_service import EditPlanService
+
+        db = MagicMock()
+        query_mock = MagicMock()
+        query_mock.filter.return_value.delete.return_value = 0
+        db.query.return_value = query_mock
+
+        # Simulate failure during create
+        mock_clip_cls.create.side_effect = ValueError("模拟异常")
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+
+        svc = EditPlanService.__new__(EditPlanService)
+        svc._clip_repo = clip_repo
+
+        with pytest.raises(ValueError, match="模拟异常"):
+            svc.replace_all_clips_transactional(
+                "plan-1",
+                [{"asset_id": "bad", "start_time": 0.0, "duration": 1.0, "order": 0}],
+            )
+
+        db.rollback.assert_called_once()
+        db.commit.assert_not_called()
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_order_defaults_to_index(self, mock_clip_cls, mock_model_cls):
+        """order=0 时使用索引值作为 order。"""
+        from app.services.edit_plan_service import EditPlanService
+
+        db = MagicMock()
+        query_mock = MagicMock()
+        query_mock.filter.return_value.delete.return_value = 0
+        db.query.return_value = query_mock
+
+        ready_query = MagicMock()
+        ready_query.filter.return_value.filter.return_value.filter.return_value.all.return_value = []
+        db.query.side_effect = [query_mock, ready_query]
+
+        mock_entity = MagicMock()
+        mock_entity.id = "clip-1"
+        mock_entity.plan_id = "plan-1"
+        mock_entity.clip_type = "main"
+        mock_entity.order = 0  # order=0 → 使用 i=0
+        mock_entity.asset_id = "a1"
+        mock_entity.text_content = ""
+        mock_entity.start_time = 0.0
+        mock_entity.duration = 1.0
+        mock_entity.transition_effect = "cut"
+        mock_entity.transition_duration = 0.0
+        mock_entity.playback_speed = 1.0
+        mock_entity.status.value = "pending"
+        mock_entity.config = {}
+        mock_clip_cls.create.return_value = mock_entity
+
+        mock_model_cls.return_value = MagicMock()
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+
+        svc = EditPlanService.__new__(EditPlanService)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "a1", "start_time": 0.0, "duration": 1.0, "order": 0}],
+        )
+
+        # order=0 → falsy → use index i=0
+        create_call = mock_clip_cls.create.call_args
+        assert create_call.kwargs["order"] == 0

tests/unit/test_title_overlay.py

@@ -56,3 +56,34 @@ def test_wrap_title_text_respects_explicit_newline():
     font = ImageFont.load_default()
     lines = wrap_title_text("第一行\n第二行", font, max_width=10000)
     assert lines == ["第一行", "第二行"]
+
+
+def test_apply_title_to_image_custom_color(sample_image):
+    """自定义颜色参数能正常生成图片。"""
+    result = apply_title_to_image(sample_image, "彩色标题", color="#ff0000")
+    assert result == sample_image
+    assert Path(sample_image).stat().st_size > 0
+
+
+def test_apply_title_to_image_short_hex_color(sample_image):
+    """3 位缩写 hex 颜色也能正常解析。"""
+    result = apply_title_to_image(sample_image, "短色", color="#f00")
+    assert result == sample_image
+
+
+def test_apply_title_to_image_invalid_color_fallback(sample_image):
+    """无效颜色字符串 fallback 到白色，不报错。"""
+    result = apply_title_to_image(sample_image, "异常色", color="not-a-color")
+    assert result == sample_image
+
+
+def test_parse_hex_color():
+    from packages.shared.title_overlay import _parse_hex_color
+
+    assert _parse_hex_color("#ffffff") == (255, 255, 255)
+    assert _parse_hex_color("#000000") == (0, 0, 0)
+    assert _parse_hex_color("#ff0000") == (255, 0, 0)
+    assert _parse_hex_color("#f00") == (255, 0, 0)
+    assert _parse_hex_color("") == (255, 255, 255)
+    assert _parse_hex_color("invalid") == (255, 255, 255)
+    assert _parse_hex_color("#gggggg") == (255, 255, 255)

tests/unit/test_unified_render_reuse.py

@@ -1,287 +0,0 @@
-"""统一渲染路径 — 编辑器预览产物复用逻辑单元测试。
-
-覆盖：
-- _find_reusable_preview_task: 查找可复用的预览任务
-- _get_task_output_url: 获取任务输出 URL
-- 编辑器 generate 接口复用预览产物路径
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from dataclasses import dataclass, field
-from datetime import datetime, timedelta, timezone
-from typing import Any, Optional
-from unittest.mock import MagicMock, patch
-
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
-
-# ── Stub Repository ──────────────────────────────────────────────────────────
-
-
-class StubGenTaskRepo:
-    def __init__(self):
-        self._store = {}
-
-    def create(self, task):
-        self._store[task.id] = task
-        return task
-
-    def get(self, task_id):
-        return self._store.get(task_id)
-
-    def update(self, task):
-        self._store[task.id] = task
-        return task
-
-    def list_by_source_edit_plan(self, plan_id):
-        return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id]
-
-
-def _make_task(**kwargs):
-    defaults = dict(
-        id="task-001",
-        project_id="proj-1",
-        asset_library_id="lib-1",
-        strategy_id="one_take",
-        voice_library_id="",
-        template_id="tmpl-1",
-        asset_ids=["a1"],
-        title_ids=[],
-        voice_ids=[],
-        status=GenerationTaskStatus.COMPLETED,
-        progress=100.0,
-        result_count=1,
-        error_message="",
-        created_by_user_id="user-1",
-        source_edit_plan_id="plan-1",
-        asset_select_mode="all",
-        is_preview=True,
-        source_task_id="",
-        output_width=1920,
-        output_height=1080,
-        cover_url="",
-        custom_title="",
-        video_title="",
-        resolution="",
-        bgm_config={},
-        completed_at=datetime.now(timezone.utc),
-    )
-    defaults.update(kwargs)
-    return GenerationTask(**defaults)
-
-
-def _make_plan(updated_at=None):
-    plan = MagicMock()
-    plan.id = "plan-1"
-    plan.updated_at = updated_at or datetime.now(timezone.utc)
-    plan.status = MagicMock()
-    plan.status.value = "editing"
-    plan.config = {"clips": [{"id": "c1"}, {"id": "c2"}]}
-    return plan
-
-
-# ── _find_reusable_preview_task ──────────────────────────────────────────────
-
-
-class TestFindReusablePreviewTask:
-    def test_returns_completed_preview_task(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = StubGenTaskRepo()
-        now = datetime.now(timezone.utc)
-        task = _make_task(
-            source_edit_plan_id="plan-1",
-            is_preview=True,
-            status=GenerationTaskStatus.COMPLETED,
-            completed_at=now - timedelta(minutes=5),
-        )
-        repo.create(task)
-
-        plan = _make_plan(updated_at=now - timedelta(minutes=10))
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-
-        assert result is not None
-        assert result.id == "task-001"
-
-    def test_returns_none_when_no_tasks(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = StubGenTaskRepo()
-        plan = _make_plan()
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-        assert result is None
-
-    def test_returns_none_when_preview_not_completed(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = StubGenTaskRepo()
-        task = _make_task(
-            source_edit_plan_id="plan-1",
-            is_preview=True,
-            status=GenerationTaskStatus.RUNNING,
-            completed_at=None,
-        )
-        repo.create(task)
-
-        plan = _make_plan()
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-        assert result is None
-
-    def test_returns_none_when_plan_modified_after_preview(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = StubGenTaskRepo()
-        now = datetime.now(timezone.utc)
-        task = _make_task(
-            source_edit_plan_id="plan-1",
-            is_preview=True,
-            status=GenerationTaskStatus.COMPLETED,
-            completed_at=now - timedelta(minutes=10),
-        )
-        repo.create(task)
-
-        # Plan was updated AFTER preview completed
-        plan = _make_plan(updated_at=now)
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-        assert result is None
-
-    def test_skips_non_preview_tasks(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = StubGenTaskRepo()
-        task = _make_task(
-            source_edit_plan_id="plan-1",
-            is_preview=False,  # not a preview task
-            status=GenerationTaskStatus.COMPLETED,
-        )
-        repo.create(task)
-
-        plan = _make_plan()
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-        assert result is None
-
-    def test_handles_repo_exception(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = MagicMock()
-        repo.list_by_source_edit_plan.side_effect = Exception("db error")
-        plan = _make_plan()
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-        assert result is None
-
-
-# ── _get_task_output_url ─────────────────────────────────────────────────────
-
-
-class TestGetTaskOutputUrl:
-    def test_returns_video_url(self):
-        from app.api.routes.templates_editor.generation import _get_task_output_url
-
-        task = _make_task()
-        repo = MagicMock()
-        db = MagicMock()
-
-        mock_video = MagicMock()
-        mock_video.file_url = "https://oss.example.com/video.mp4"
-
-        mock_use_case = MagicMock()
-        mock_use_case.execute.return_value = [mock_video]
-
-        with patch(
-            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
-            return_value=mock_use_case,
-        ):
-            result = _get_task_output_url(task, repo, db)
-
-        assert result == "https://oss.example.com/video.mp4"
-
-    def test_returns_empty_when_no_videos(self):
-        from app.api.routes.templates_editor.generation import _get_task_output_url
-
-        task = _make_task()
-        repo = MagicMock()
-        db = MagicMock()
-
-        mock_use_case = MagicMock()
-        mock_use_case.execute.return_value = []
-
-        with patch(
-            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
-            return_value=mock_use_case,
-        ):
-            result = _get_task_output_url(task, repo, db)
-
-        assert result == ""
-
-    def test_returns_empty_on_exception(self):
-        from app.api.routes.templates_editor.generation import _get_task_output_url
-
-        task = _make_task()
-        repo = MagicMock()
-        db = MagicMock()
-
-        with patch(
-            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
-            side_effect=Exception("db error"),
-        ):
-            result = _get_task_output_url(task, repo, db)
-
-        assert result == ""
-
-
-# ── mark_confirmed ───────────────────────────────────────────────────────────
-
-
-class TestMarkConfirmed:
-    def test_sets_is_preview_false(self):
-        task = _make_task(is_preview=True)
-        task.mark_confirmed()
-        assert task.is_preview is False
-
-    def test_sets_cover_url(self):
-        task = _make_task()
-        task.mark_confirmed(cover_url="https://example.com/cover.jpg")
-        assert task.cover_url == "https://example.com/cover.jpg"
-
-    def test_sets_custom_title(self):
-        task = _make_task()
-        task.mark_confirmed(custom_title="My Video")
-        assert task.custom_title == "My Video"
-
-    def test_sets_output_dimensions(self):
-        task = _make_task()
-        task.mark_confirmed(output_width=1080, output_height=1920)
-        assert task.output_width == 1080
-        assert task.output_height == 1920
-
-    def test_zero_dimensions_not_applied(self):
-        task = _make_task(output_width=1920, output_height=1080)
-        task.mark_confirmed(output_width=0, output_height=0)
-        assert task.output_width == 1920
-        assert task.output_height == 1080
-
-    def test_skips_when_plan_updated_at_is_none(self):
-        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
-
-        repo = StubGenTaskRepo()
-        now = datetime.now(timezone.utc)
-        task = _make_task(
-            source_edit_plan_id="plan-1",
-            is_preview=True,
-            status=GenerationTaskStatus.COMPLETED,
-            completed_at=now - timedelta(minutes=5),
-        )
-        repo.create(task)
-
-        plan = _make_plan(updated_at=None)
-        result = _find_reusable_preview_task(repo, "plan-1", plan)
-        assert result is None

tests/unit/test_worker_cover_meta_and_status.py

@@ -0,0 +1,70 @@
+"""防回归测试：P1 修复
+- Bug 1: gen_task 过期内存对象 _repo.update() 覆盖 DB status 为 pending
+- Bug 2: 封面模型误用 .metadata（SQLAlchemy 保留属性），应为 .extra_meta
+"""
+
+import ast
+from pathlib import Path
+
+import pytest
+
+GENERATION_FILE = Path(__file__).resolve().parents[2] / "apps" / "worker" / "worker_app" / "tasks" / "generation.py"
+
+
+def _read_source() -> str:
+    return GENERATION_FILE.read_text(encoding="utf-8")
+
+
+class TestCoverModelUsesExtraMeta:
+    """封面持久化必须使用 ORM 属性 extra_meta，而不是 SQLAlchemy 保留的 .metadata。"""
+
+    def test_no_metadata_attribute_access_on_cover_model(self):
+        source = _read_source()
+        # 禁止对 _cover_model.metadata 进行读或写
+        assert "_cover_model.metadata" not in source, (
+            "_cover_model.metadata is the SQLAlchemy reserved MetaData object, "
+            "not the JSON column. Use _cover_model.extra_meta instead."
+        )
+
+    def test_extra_meta_used_for_cover_candidates(self):
+        source = _read_source()
+        assert "_cover_model.extra_meta" in source
+        assert 'meta["cover_candidates"]' in source
+
+
+class TestAssetAnalysesDoesNotOverwriteStatus:
+    """保存 asset_analyses 时不能用过期的 gen_task 内存对象整体 _repo.update，
+    否则会把已被 _update_task_status 改为 running 的 status 覆盖回 pending。"""
+
+    def test_no_stale_repo_update_with_gen_task(self):
+        source = _read_source()
+        # 旧代码：gen_task.extra_meta = {...}; _repo.update(gen_task)
+        # 这行会把内存中的 pending status 写回 DB
+        assert "_repo.update(gen_task)" not in source, (
+            "_repo.update(gen_task) writes a stale in-memory object back to DB, "
+            "overwriting status set by _update_task_status. "
+            "Use an independent session to update only extra_meta."
+        )
+
+    def test_asset_analyses_uses_independent_session(self):
+        """asset_analyses 持久化必须用独立 session 查询最新模型再提交。"""
+        source = _read_source()
+        assert "_meta_session" in source
+        assert "GenerationTaskModel" in source
+        # 必须只更新 extra_meta 字段
+        assert 'existing["asset_analyses"]' in source
+
+
+class TestGenerationTaskModelOrmAttribute:
+    """确认 ORM 属性映射：Python 属性 extra_meta -> DB 列 metadata。"""
+
+    def test_orm_attribute_is_extra_meta(self):
+        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+        # ORM 属性必须存在
+        assert hasattr(GenerationTaskModel, "extra_meta")
+        # .metadata 是 SQLAlchemy 声明基类保留的 MetaData，不是列描述符
+        # 它不应该是我们的 JSON 字段
+        from sqlalchemy import MetaData
+
+        assert isinstance(GenerationTaskModel.metadata, MetaData)

tests/unit/test_worker_generate_video_task_binding.py

@@ -0,0 +1,39 @@
+"""Regression test: ensure worker.generate_video Celery task is bound to the
+real generate_video function, not a helper introduced above it.
+
+Context (P0 incident 2026-08-23): a refactor inserted helper function
+_sync_task_config_to_plan directly under the @celery_app.task decorator,
+so Celery registered the helper as "worker.generate_video". Calling the
+task with a single task_id raised TypeError and every generation job
+failed immediately. This test pins the decorator target.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+
+def test_generate_video_task_registered_under_expected_name():
+    from worker_app.tasks.generation import generate_video
+
+    # Celery task object exposes its registered name
+    assert generate_video.name == "worker.generate_video"
+
+
+def test_generate_video_task_signature_has_task_id():
+    from worker_app.tasks.generation import generate_video
+
+    # For bind=True tasks Celery binds self at call time, so run() signature
+    # starts directly with task_id (verified on Celery 5.x).
+    sig = inspect.signature(generate_video.run)
+    params = list(sig.parameters)
+    assert params[0] == "task_id", f"expected task_id as first param, got {params}"
+
+
+def test_sync_task_config_to_plan_is_plain_function():
+    """Helper must NOT be registered as a Celery task."""
+    from worker_app.tasks.generation import _sync_task_config_to_plan
+
+    assert not hasattr(
+        _sync_task_config_to_plan, "run"
+    ), "_sync_task_config_to_plan must be a plain function, not a Celery task"