apps/api/app/api/routes/edit_plans.py

@@ -6,6 +6,8 @@ RESTful CRUD for EditPlan:
 - POST   /api/v1/edit-plans            创建
 - PUT    /api/v1/edit-plans/{id}       更新（含状态机流转）
 - DELETE /api/v1/edit-plans/{id}       删除
+- POST   /api/v1/edit-plans/{id}/generate          触发剪辑渲染生成（任务 2.05）
+- GET    /api/v1/edit-plans/{id}/generation-status  查询生成进度（任务 2.05）
 """
 
 from __future__ import annotations
@@ -15,13 +17,25 @@ from datetime import datetime
 from typing import Any, List, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.core.celery_app import celery_app
 from app.dependencies import get_db_session
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
 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,
+)
+from packages.application.generation_tasks import (
+    CreateGenerationTaskCommand,
+    CreateGenerationTaskUseCase,
+)
 from packages.domain.edit_plan import EditPlan, EditPlanStatus
+from packages.domain.edit_plan_clip import EditPlanClipStatus
 
 logger = logging.getLogger(__name__)
 
@@ -76,6 +90,36 @@ class EditPlanListResponse(BaseModel):
     page_size: int
 
 
+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
+    clips: List[ClipStatusItem]
+
+
+class EditPlanGenerateResponse(BaseModel):
+    """剪辑计划触发生成响应体"""
+
+    plan_id: str
+    plan_status: str
+    generation_task_id: str
+    clip_count: int
+
+
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
 
@@ -302,3 +346,140 @@ def delete_plan(
         plan_id,
         current_user.user.id,
     )
+
+
+# ── 生成相关端点（任务 2.05） ─────────────────────────────────────────────────
+
+
+@router.post("/{plan_id}/generate", response_model=EditPlanGenerateResponse)
+def generate_plan(
+    plan_id: str,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+) -> EditPlanGenerateResponse:
+    """触发剪辑计划渲染生成
+
+    前置条件：计划状态必须为 editing，且至少有一个片段。
+    流程：
+    1. 验证计划状态为 editing
+    2. 将 pending 片段标记为 ready
+    3. 创建 GenerationTask
+    4. 调度 Celery 任务 worker.render_edit_plan
+    5. 将计划状态流转为 rendering
+    """
+    plan_repo = SQLAlchemyEditPlanRepository(db)
+    clip_repo = SQLAlchemyEditPlanClipRepository(db)
+    gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+
+    plan = plan_repo.get(plan_id)
+    if plan is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"剪辑计划不存在: {plan_id}",
+        )
+
+    # 验证状态必须为 editing
+    if plan.status != EditPlanStatus.EDITING:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=(
+                f"只有 editing 状态的计划才能触发生成，当前状态: {plan.status.value}。"
+                f"请先将计划状态流转为 editing"
+            ),
+        )
+
+    # 检查片段
+    clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
+    if not clips:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="剪辑计划没有片段，请先添加片段再触发生成",
+        )
+
+    # 将 pending 片段标记为 ready
+    pending_clips = [c for c in clips if c.status == EditPlanClipStatus.PENDING]
+    for clip in pending_clips:
+        clip.mark_ready()
+        clip_repo.update(clip)
+
+    # 创建 GenerationTask
+    gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
+    gen_task = gen_task_use_case.execute(
+        CreateGenerationTaskCommand(
+            project_id="",
+            template_id=plan.template_id,
+            created_by_user_id=current_user.user.id,
+        )
+    )
+
+    # 将 generation_task_id 存入 plan config
+    plan.config["generation_task_id"] = gen_task.id
+
+    # 流转状态为 rendering
+    plan.start_rendering()
+    plan_repo.update(plan)
+
+    # 调度 Celery 任务
+    celery_app.send_task("worker.render_edit_plan", args=[plan_id])
+
+    logger.info(
+        "触发剪辑计划生成: plan_id=%s gen_task_id=%s clips=%d by user=%s",
+        plan_id,
+        gen_task.id,
+        len(clips),
+        current_user.user.id,
+    )
+
+    return EditPlanGenerateResponse(
+        plan_id=plan_id,
+        plan_status=plan.status.value if hasattr(plan.status, "value") else plan.status,
+        generation_task_id=gen_task.id,
+        clip_count=len(clips),
+    )
+
+
+@router.get(
+    "/{plan_id}/generation-status",
+    response_model=EditPlanGenerationStatusResponse,
+)
+def get_generation_status(
+    plan_id: str,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+) -> EditPlanGenerationStatusResponse:
+    """查询剪辑计划生成进度
+
+    返回计划状态、关联的 GenerationTask ID、以及每个片段的状态。
+    """
+    plan_repo = SQLAlchemyEditPlanRepository(db)
+    clip_repo = SQLAlchemyEditPlanClipRepository(db)
+
+    plan = plan_repo.get(plan_id)
+    if plan is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"剪辑计划不存在: {plan_id}",
+        )
+
+    clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
+    generation_task_id = plan.config.get("generation_task_id")
+
+    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
+    ]
+
+    return EditPlanGenerationStatusResponse(
+        plan_id=plan_id,
+        plan_status=plan.status.value if hasattr(plan.status, "value") else plan.status,
+        generation_task_id=generation_task_id,
+        clips=clip_items,
+    )

apps/worker/worker_app/celery_app.py

@@ -12,5 +12,6 @@ celery_app.conf.imports = (
     "worker_app.tasks.classification",
     "worker_app.tasks.generation",
     "worker_app.tasks.voice_extraction",
+    "worker_app.tasks.edit_plan_generation",
     "apps.worker.video_processing.dedup",
 )

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -0,0 +1,351 @@
+"""剪辑计划渲染任务 — Phase 8 任务 2.05.
+
+Celery 任务 worker.render_edit_plan:
+  1. 加载 EditPlan + EditPlanClips
+  2. 下载各片段素材
+  3. 按 order 顺序拼接片段
+  4. 上传渲染结果到 OSS
+  5. 更新 EditPlan / EditPlanClip 状态
+  6. 更新 GenerationTask 进度
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import subprocess  # nosec B404
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+from urllib.parse import urlparse
+
+import oss2
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+FFMPEG_BIN = shutil.which("ffmpeg") or "ffmpeg"
+FFPROBE_BIN = shutil.which("ffprobe") or "ffprobe"
+OUTPUT_WIDTH = 1280
+OUTPUT_HEIGHT = 720
+OUTPUT_FPS = 25.0
+
+
+# ── OSS helpers ───────────────────────────────────────────────────────────────
+
+
+def _oss_settings() -> tuple[str, str, str, str] | None:
+    """获取 OSS 配置"""
+    access_key_id = os.getenv("OSS_ACCESS_KEY_ID")
+    access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET")
+    endpoint = os.getenv("OSS_ENDPOINT")
+    bucket_name = os.getenv("OSS_BUCKET_NAME")
+    if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
+        return None
+    return access_key_id, access_key_secret, endpoint, bucket_name
+
+
+def _oss_bucket() -> oss2.Bucket | None:
+    """获取 OSS Bucket"""
+    settings = _oss_settings()
+    if settings is None:
+        return None
+    access_key_id, access_key_secret, endpoint, bucket_name = settings
+    return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
+
+
+def _normalize_storage_key(storage_key_or_url: str) -> str:
+    """标准化存储键"""
+    if storage_key_or_url.startswith(("http://", "https://")):
+        return urlparse(storage_key_or_url).path.lstrip("/")
+    return storage_key_or_url.lstrip("/")
+
+
+def _download_asset(asset_storage_key: str, local_path: Path) -> bool:
+    """下载素材文件到本地"""
+    bucket = _oss_bucket()
+    if bucket is None:
+        return False
+    try:
+        bucket.get_object_to_file(_normalize_storage_key(asset_storage_key), str(local_path))
+        return local_path.exists() and local_path.stat().st_size > 0
+    except Exception:
+        logger.exception("下载素材失败: %s", asset_storage_key)
+        return False
+
+
+def _upload_to_oss(local_path: Path, storage_key: str) -> str | None:
+    """上传文件到 OSS，返回公开 URL"""
+    bucket = _oss_bucket()
+    if bucket is None:
+        return None
+    try:
+        bucket.put_object_from_file(storage_key, str(local_path))
+        settings = _oss_settings()
+        if settings:
+            _, _, endpoint, bucket_name = settings
+            return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}"
+        return None
+    except Exception:
+        logger.exception("上传 OSS 失败: %s", storage_key)
+        return None
+
+
+# ── FFmpeg helpers ────────────────────────────────────────────────────────────
+
+
+def _run_ffmpeg(command: list[str]) -> None:
+    """执行 FFmpeg 命令"""
+    subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)  # nosec B603
+
+
+def _probe_duration(local_path: Path) -> float:
+    """获取视频/音频时长"""
+    try:
+        result = subprocess.run(  # nosec B603
+            [
+                FFPROBE_BIN,
+                "-v", "error",
+                "-show_entries", "format=duration",
+                "-of", "default=noprint_wrappers=1:nokey=1",
+                str(local_path),
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        return float(result.stdout.strip())
+    except Exception:
+        return 0.0
+
+
+def _concatenate_clips(
+    clip_paths: list[Path],
+    output_path: Path,
+    transition_effects: list[str] | None = None,
+) -> bool:
+    """将多个片段拼接为最终视频
+
+    使用 FFmpeg concat demuxer 实现。
+    """
+    if not clip_paths:
+        return False
+
+    if len(clip_paths) == 1:
+        # 单片段直接复制
+        try:
+            shutil.copy2(str(clip_paths[0]), str(output_path))
+            return True
+        except Exception:
+            return False
+
+    # 多片段：使用 concat demuxer
+    concat_file = output_path.parent / "concat_list.txt"
+    try:
+        with open(concat_file, "w") as f:
+            for p in clip_paths:
+                f.write(f"file '{p}'\n")
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-f", "concat",
+            "-safe", "0",
+            "-i", str(concat_file),
+            "-c", "copy",
+            str(output_path),
+        ]
+        _run_ffmpeg(command)
+        return output_path.exists() and output_path.stat().st_size > 0
+    except Exception:
+        logger.exception("拼接片段失败")
+        return False
+    finally:
+        if concat_file.exists():
+            concat_file.unlink()
+
+
+# ── 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 ───────────────────────────────────────────────────────────────
+
+
+@celery_app.task(name="worker.render_edit_plan", bind=True, max_retries=2)
+def render_edit_plan(self, plan_id: str) -> dict:
+    """渲染剪辑计划
+
+    流程：
+    1. 加载 EditPlan + EditPlanClips
+    2. 下载各片段素材到临时目录
+    3. 按 order 顺序拼接片段
+    4. 上传渲染结果到 OSS
+    5. 更新 EditPlan → completed, EditPlanClips → rendered
+    6. 更新 GenerationTask 进度
+    """
+    logger.info("开始渲染剪辑计划: plan_id=%s", plan_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}"}
+
+            # 2. 加载片段列表（按 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": "没有可渲染的片段"}
+
+            # 获取 generation_task_id
+            generation_task_id = plan.config.get("generation_task_id", "")
+
+            # 更新 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_repo.update(gen_task)
+
+            # 3. 下载素材并拼接
+            with tempfile.TemporaryDirectory(prefix="edit_plan_") as tmpdir:
+                tmpdir_path = Path(tmpdir)
+                clip_paths: list[Path] = []
+                rendered_clip_ids: list[str] = []
+                failed_clip_ids: list[str] = []
+
+                for clip in clips:
+                    if not clip.asset_id:
+                        # 没有素材的片段跳过，标记为失败
+                        clip.mark_failed()
+                        clip_repo.update(clip)
+                        failed_clip_ids.append(clip.id)
+                        continue
+
+                    # 下载素材
+                    ext = Path(clip.asset_id).suffix or ".mp4"
+                    local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}"
+                    if _download_asset(clip.asset_id, local_path):
+                        clip_paths.append(local_path)
+                        rendered_clip_ids.append(clip.id)
+                    else:
+                        clip.mark_failed()
+                        clip_repo.update(clip)
+                        failed_clip_ids.append(clip.id)
+
+                if not clip_paths:
+                    logger.error("所有片段素材下载失败: %s", plan_id)
+                    plan.mark_failed()
+                    plan_repo.update(plan)
+                    if generation_task_id:
+                        gen_task = gen_task_repo.get(generation_task_id)
+                        if gen_task:
+                            gen_task.status = "failed"
+                            gen_task.error_message = "所有片段素材下载失败"
+                            gen_task.completed_at = datetime.now(timezone.utc)
+                            gen_task_repo.update(gen_task)
+                    return {"status": "error", "message": "所有片段素材下载失败"}
+
+                # 4. 拼接片段
+                output_path = tmpdir_path / f"rendered_{plan_id}.mp4"
+                transition_effects = [c.transition_effect for c in clips if c.asset_id]
+                success = _concatenate_clips(clip_paths, output_path, transition_effects)
+
+                if not success:
+                    logger.error("片段拼接失败: %s", plan_id)
+                    plan.mark_failed()
+                    plan_repo.update(plan)
+                    if generation_task_id:
+                        gen_task = gen_task_repo.get(generation_task_id)
+                        if gen_task:
+                            gen_task.status = "failed"
+                            gen_task.error_message = "片段拼接失败"
+                            gen_task.completed_at = datetime.now(timezone.utc)
+                            gen_task_repo.update(gen_task)
+                    return {"status": "error", "message": "片段拼接失败"}
+
+                # 5. 上传到 OSS
+                storage_key = f"rendered/{plan_id}/output.mp4"
+                output_url = _upload_to_oss(output_path, storage_key)
+
+                # 6. 更新片段状态为 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)
+
+                # 7. 更新 EditPlan 状态为 completed
+                plan.config["rendered_url"] = output_url or ""
+                plan.config["rendered_storage_key"] = storage_key
+                plan.mark_completed()
+                plan_repo.update(plan)
+
+                # 8. 更新 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
+                        gen_task.result_count = len(rendered_clip_ids)
+                        gen_task.completed_at = datetime.now(timezone.utc)
+                        gen_task_repo.update(gen_task)
+
+                logger.info(
+                    "剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d",
+                    plan_id,
+                    len(rendered_clip_ids),
+                    len(failed_clip_ids),
+                )
+
+                return {
+                    "status": "completed",
+                    "plan_id": plan_id,
+                    "rendered_count": len(rendered_clip_ids),
+                    "failed_count": len(failed_clip_ids),
+                    "output_url": output_url,
+                }
+
+        except Exception as exc:
+            logger.exception("渲染剪辑计划异常: %s", plan_id)
+            # 尝试标记计划为失败
+            try:
+                plan = plan_repo.get(plan_id)
+                if plan and plan.status.value == "rendering":
+                    plan.mark_failed()
+                    plan_repo.update(plan)
+            except Exception:
+                pass
+            raise self.retry(exc=exc, countdown=60)
+
+    return {"status": "error", "message": "数据库连接失败"}

tests/unit/test_edit_plan_generation_api.py

@@ -0,0 +1,590 @@
+"""剪辑计划生成 API 单元测试 — Phase 8 任务 2.05.
+
+覆盖 2 个新端点：
+  POST /api/v1/edit-plans/{id}/generate          — 触发剪辑渲染生成
+  GET  /api/v1/edit-plans/{id}/generation-status  — 查询生成进度
+
+使用 FastAPI TestClient + Stub Repository + dependency_overrides.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, 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
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
+
+from packages.domain.edit_plan import EditPlan, EditPlanStatus
+from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
+
+
+# ── Stub Repositories ─────────────────────────────────────────────────────────
+
+
+class StubEditPlanRepository:
+    """内存中模拟 EditPlan 仓储"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, EditPlan] = {}
+
+    def list_all(
+        self,
+        *,
+        status: Optional[EditPlanStatus] = None,
+        skip: int = 0,
+        limit: int = 50,
+    ) -> list[EditPlan]:
+        items = list(self._store.values())
+        if status:
+            items = [p for p in items if p.status == status]
+        items.sort(key=lambda p: p.created_at, reverse=True)
+        return items[skip : skip + limit]
+
+    def list_by_template(
+        self,
+        template_id: str,
+        *,
+        status: Optional[EditPlanStatus] = None,
+        skip: int = 0,
+        limit: int = 50,
+    ) -> list[EditPlan]:
+        items = [p for p in self._store.values() if p.template_id == template_id]
+        if status:
+            items = [p for p in items if p.status == status]
+        items.sort(key=lambda p: p.created_at, reverse=True)
+        return items[skip : skip + limit]
+
+    def get(self, plan_id: str) -> Optional[EditPlan]:
+        return self._store.get(plan_id)
+
+    def create(self, plan: EditPlan) -> EditPlan:
+        self._store[plan.id] = plan
+        return plan
+
+    def update(self, plan: EditPlan) -> EditPlan:
+        if plan.id not in self._store:
+            raise ValueError(f"EditPlan {plan.id} not found")
+        self._store[plan.id] = plan
+        return plan
+
+    def delete(self, plan_id: str) -> bool:
+        if plan_id in self._store:
+            del self._store[plan_id]
+            return True
+        return False
+
+    def count(self, *, status: Optional[EditPlanStatus] = None) -> int:
+        items = list(self._store.values())
+        if status:
+            items = [p for p in items if p.status == status]
+        return len(items)
+
+
+class StubEditPlanClipRepository:
+    """内存中模拟 EditPlanClip 仓储"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, EditPlanClip] = {}
+
+    def list_by_plan(
+        self,
+        plan_id: str,
+        *,
+        status: Optional[EditPlanClipStatus] = None,
+        skip: int = 0,
+        limit: int = 100,
+    ) -> list[EditPlanClip]:
+        items = [c for c in self._store.values() if c.plan_id == plan_id]
+        if status:
+            items = [c for c in items if c.status == status]
+        items.sort(key=lambda c: c.order)
+        return items[skip : skip + limit]
+
+    def get(self, clip_id: str) -> Optional[EditPlanClip]:
+        return self._store.get(clip_id)
+
+    def create(self, clip: EditPlanClip) -> EditPlanClip:
+        self._store[clip.id] = clip
+        return clip
+
+    def update(self, clip: EditPlanClip) -> EditPlanClip:
+        if clip.id not in self._store:
+            raise ValueError(f"EditPlanClip {clip.id} not found")
+        self._store[clip.id] = clip
+        return clip
+
+    def delete(self, clip_id: str) -> bool:
+        if clip_id in self._store:
+            del self._store[clip_id]
+            return True
+        return False
+
+    def delete_by_plan(self, plan_id: str) -> int:
+        to_delete = [c.id for c in self._store.values() if c.plan_id == plan_id]
+        for cid in to_delete:
+            del self._store[cid]
+        return len(to_delete)
+
+    def count(
+        self,
+        plan_id: Optional[str] = None,
+        *,
+        status: Optional[EditPlanClipStatus] = None,
+    ) -> int:
+        items = list(self._store.values())
+        if plan_id:
+            items = [c for c in items if c.plan_id == plan_id]
+        if status:
+            items = [c for c in items if c.status == status]
+        return len(items)
+
+
+class StubGenerationTaskRepository:
+    """内存中模拟 GenerationTask 仓储"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, Any] = {}
+
+    def create(self, task: Any) -> Any:
+        self._store[task.id] = task
+        return task
+
+    def get(self, task_id: str) -> Optional[Any]:
+        return self._store.get(task_id)
+
+    def update(self, task: Any) -> Any:
+        if task.id not in self._store:
+            raise ValueError(f"GenerationTask {task.id} not found")
+        self._store[task.id] = task
+        return task
+
+    def list_by_project(self, project_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.project_id == project_id]
+
+    def list_by_user(self, user_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.created_by_user_id == user_id]
+
+    def count_by_user(self, user_id: str) -> int:
+        return len([t for t in self._store.values() if t.created_by_user_id == user_id])
+
+    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[Any]:
+        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
+        items.sort(key=lambda t: t.created_at, reverse=True)
+        return items[:limit]
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeUser:
+    id: str = "user-001"
+    email: str = "test@example.com"
+
+
+@dataclass
+class FakeAuthenticatedUser:
+    user: FakeUser = field(default_factory=FakeUser)
+    session_id: str | None = None
+    token_type: str | None = None
+
+
+@pytest.fixture
+def plan_repo() -> StubEditPlanRepository:
+    return StubEditPlanRepository()
+
+
+@pytest.fixture
+def clip_repo() -> StubEditPlanClipRepository:
+    return StubEditPlanClipRepository()
+
+
+@pytest.fixture
+def gen_task_repo() -> StubGenerationTaskRepository:
+    return StubGenerationTaskRepository()
+
+
+@pytest.fixture
+def app(
+    plan_repo: StubEditPlanRepository,
+    clip_repo: StubEditPlanClipRepository,
+    gen_task_repo: StubGenerationTaskRepository,
+) -> FastAPI:
+    """构建测试 FastAPI 应用，注入 Stub Repository"""
+    from app.api.routes.edit_plans import router
+    from app.auth import get_current_user
+    from app.dependencies import get_db_session
+    import app.api.routes.edit_plans as route_module
+
+    # 替换 Repository 类
+    original_plan_repo = route_module.SQLAlchemyEditPlanRepository
+    original_clip_repo = route_module.SQLAlchemyEditPlanClipRepository
+    original_gen_repo = route_module.SQLAlchemyGenerationTaskRepository
+
+    route_module.SQLAlchemyEditPlanRepository = lambda session: plan_repo
+    route_module.SQLAlchemyEditPlanClipRepository = lambda session: clip_repo
+    route_module.SQLAlchemyGenerationTaskRepository = lambda session: gen_task_repo
+
+    test_app = FastAPI()
+    test_app.include_router(router, prefix="/api/v1/edit-plans")
+
+    def override_get_db_session():
+        yield MagicMock()
+
+    def override_get_current_user():
+        return FakeAuthenticatedUser()
+
+    test_app.dependency_overrides[get_db_session] = override_get_db_session
+    test_app.dependency_overrides[get_current_user] = override_get_current_user
+
+    yield test_app
+
+    # 恢复
+    route_module.SQLAlchemyEditPlanRepository = original_plan_repo
+    route_module.SQLAlchemyEditPlanClipRepository = original_clip_repo
+    route_module.SQLAlchemyGenerationTaskRepository = original_gen_repo
+    test_app.dependency_overrides.clear()
+
+
+@pytest.fixture
+def client(app: FastAPI) -> TestClient:
+    return TestClient(app)
+
+
+def _make_plan(
+    name: str = "测试计划",
+    template_id: str = "tmpl-001",
+    status: EditPlanStatus = EditPlanStatus.DRAFT,
+    **kwargs: Any,
+) -> EditPlan:
+    plan = EditPlan.create(template_id=template_id, name=name, **kwargs)
+    plan.status = status
+    return plan
+
+
+def _make_clip(
+    plan_id: str,
+    clip_type: str = "MAIN",
+    order: int = 1,
+    asset_id: str = "assets/video.mp4",
+    status: EditPlanClipStatus = EditPlanClipStatus.PENDING,
+    **kwargs: Any,
+) -> EditPlanClip:
+    clip = EditPlanClip.create(
+        plan_id=plan_id,
+        clip_type=clip_type,
+        order=order,
+        asset_id=asset_id,
+        **kwargs,
+    )
+    clip.status = status
+    return clip
+
+
+# ── POST /api/v1/edit-plans/{id}/generate ─────────────────────────────────────
+
+
+class TestGeneratePlan:
+    def test_generate_success(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """editing 状态 + pending 片段 → 触发成功"""
+        plan = _make_plan(status=EditPlanStatus.EDITING)
+        plan_repo.create(plan)
+        clip = _make_clip(plan.id, order=1)
+        clip_repo.create(clip)
+
+        with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["plan_id"] == plan.id
+        assert data["plan_status"] == "rendering"
+        assert data["clip_count"] == 1
+        assert "generation_task_id" in data
+
+        # 验证计划状态已更新
+        updated = plan_repo.get(plan.id)
+        assert updated.status == EditPlanStatus.RENDERING
+
+        # 验证片段状态已更新为 ready
+        updated_clip = clip_repo.get(clip.id)
+        assert updated_clip.status == EditPlanClipStatus.READY
+
+    def test_generate_not_found(self, client: TestClient) -> None:
+        """计划不存在 → 404"""
+        resp = client.post("/api/v1/edit-plans/nonexistent/generate")
+        assert resp.status_code == 404
+        assert "不存在" in resp.json()["detail"]
+
+    def test_generate_wrong_status_draft(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+    ) -> None:
+        """draft 状态 → 400"""
+        plan = _make_plan(status=EditPlanStatus.DRAFT)
+        plan_repo.create(plan)
+        resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+        assert resp.status_code == 400
+        assert "editing" in resp.json()["detail"]
+
+    def test_generate_wrong_status_rendering(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+    ) -> None:
+        """rendering 状态 → 400"""
+        plan = _make_plan(status=EditPlanStatus.RENDERING)
+        plan_repo.create(plan)
+        resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+        assert resp.status_code == 400
+
+    def test_generate_wrong_status_completed(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+    ) -> None:
+        """completed 状态 → 400"""
+        plan = _make_plan(status=EditPlanStatus.COMPLETED)
+        plan_repo.create(plan)
+        resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+        assert resp.status_code == 400
+
+    def test_generate_no_clips(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+    ) -> None:
+        """editing 状态但没有片段 → 400"""
+        plan = _make_plan(status=EditPlanStatus.EDITING)
+        plan_repo.create(plan)
+        resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+        assert resp.status_code == 400
+        assert "片段" in resp.json()["detail"]
+
+    def test_generate_with_ready_clips(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """已有 ready 状态的片段也可以触发生成"""
+        plan = _make_plan(status=EditPlanStatus.EDITING)
+        plan_repo.create(plan)
+        clip = _make_clip(plan.id, order=1, status=EditPlanClipStatus.READY)
+        clip_repo.create(clip)
+
+        with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["plan_status"] == "rendering"
+
+    def test_generate_multiple_clips(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """多个片段全部从 pending 转为 ready"""
+        plan = _make_plan(status=EditPlanStatus.EDITING)
+        plan_repo.create(plan)
+        for i in range(3):
+            clip = _make_clip(plan.id, order=i + 1)
+            clip_repo.create(clip)
+
+        with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["clip_count"] == 3
+
+        # 验证所有片段都变为 ready
+        clips = clip_repo.list_by_plan(plan.id)
+        assert all(c.status == EditPlanClipStatus.READY for c in clips)
+
+
+# ── GET /api/v1/edit-plans/{id}/generation-status ─────────────────────────────
+
+
+class TestGetGenerationStatus:
+    def test_status_not_found(self, client: TestClient) -> None:
+        """计划不存在 → 404"""
+        resp = client.get("/api/v1/edit-plans/nonexistent/generation-status")
+        assert resp.status_code == 404
+
+    def test_status_draft_plan(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """draft 状态计划的生成状态"""
+        plan = _make_plan(status=EditPlanStatus.DRAFT)
+        plan_repo.create(plan)
+        clip = _make_clip(plan.id, order=1)
+        clip_repo.create(clip)
+
+        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["plan_id"] == plan.id
+        assert data["plan_status"] == "draft"
+        assert data["generation_task_id"] is None
+        assert len(data["clips"]) == 1
+        assert data["clips"][0]["status"] == "pending"
+
+    def test_status_rendering_plan(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """rendering 状态计划的生成状态"""
+        plan = _make_plan(status=EditPlanStatus.RENDERING)
+        plan.config["generation_task_id"] = "gen-task-001"
+        plan_repo.create(plan)
+
+        clip1 = _make_clip(plan.id, order=1, status=EditPlanClipStatus.RENDERED)
+        clip2 = _make_clip(plan.id, order=2, status=EditPlanClipStatus.READY)
+        clip_repo.create(clip1)
+        clip_repo.create(clip2)
+
+        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["plan_status"] == "rendering"
+        assert data["generation_task_id"] == "gen-task-001"
+        assert len(data["clips"]) == 2
+        statuses = {c["status"] for c in data["clips"]}
+        assert "rendered" in statuses
+        assert "ready" in statuses
+
+    def test_status_completed_plan(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """completed 状态计划的生成状态"""
+        plan = _make_plan(status=EditPlanStatus.COMPLETED)
+        plan.config["generation_task_id"] = "gen-task-002"
+        plan.config["rendered_url"] = "https://oss.example.com/rendered/output.mp4"
+        plan_repo.create(plan)
+
+        clip = _make_clip(plan.id, order=1, status=EditPlanClipStatus.RENDERED)
+        clip_repo.create(clip)
+
+        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["plan_status"] == "completed"
+        assert data["generation_task_id"] == "gen-task-002"
+        assert len(data["clips"]) == 1
+        assert data["clips"][0]["status"] == "rendered"
+
+    def test_status_clip_fields(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """验证片段状态返回的字段完整性"""
+        plan = _make_plan(status=EditPlanStatus.EDITING)
+        plan_repo.create(plan)
+        clip = _make_clip(
+            plan.id,
+            clip_type="INTRO",
+            order=1,
+            asset_id="assets/intro.mp4",
+            duration=5.0,
+        )
+        clip.text_content = "欢迎观看"
+        clip_repo.create(clip)
+
+        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
+        assert resp.status_code == 200
+        data = resp.json()
+        clip_data = data["clips"][0]
+        assert clip_data["clip_id"] == clip.id
+        assert clip_data["clip_type"] == "INTRO"
+        assert clip_data["order"] == 1
+        assert clip_data["asset_id"] == "assets/intro.mp4"
+        assert clip_data["text_content"] == "欢迎观看"
+        assert clip_data["duration"] == 5.0
+
+    def test_status_no_clips(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+    ) -> None:
+        """没有片段的计划也能查询状态"""
+        plan = _make_plan(status=EditPlanStatus.DRAFT)
+        plan_repo.create(plan)
+
+        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["clips"] == []
+
+
+# ── Response Schema 验证 ──────────────────────────────────────────────────────
+
+
+class TestResponseSchema:
+    def test_generate_response_structure(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+        clip_repo: StubEditPlanClipRepository,
+    ) -> None:
+        """验证 generate 端点响应结构"""
+        plan = _make_plan(status=EditPlanStatus.EDITING)
+        plan_repo.create(plan)
+        clip = _make_clip(plan.id, order=1)
+        clip_repo.create(clip)
+
+        with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        expected_keys = {"plan_id", "plan_status", "generation_task_id", "clip_count"}
+        assert set(data.keys()) == expected_keys
+
+    def test_generation_status_response_structure(
+        self,
+        client: TestClient,
+        plan_repo: StubEditPlanRepository,
+    ) -> None:
+        """验证 generation-status 端点响应结构"""
+        plan = _make_plan(status=EditPlanStatus.DRAFT)
+        plan_repo.create(plan)
+
+        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
+        assert resp.status_code == 200
+        data = resp.json()
+        expected_keys = {"plan_id", "plan_status", "generation_task_id", "clips"}
+        assert set(data.keys()) == expected_keys