diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 069b58e4f..c3cc7692d 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/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, + ) diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 5dd182a0e..fabaec025 100644 --- a/apps/worker/worker_app/celery_app.py +++ b/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", ) diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py new file mode 100644 index 000000000..1d2797887 --- /dev/null +++ b/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": "数据库连接失败"} diff --git a/tests/unit/test_edit_plan_generation_api.py b/tests/unit/test_edit_plan_generation_api.py new file mode 100644 index 000000000..99d48eb59 --- /dev/null +++ b/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