diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 0d1ed6f96..ca6ae1d13 100755 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -605,6 +605,7 @@ def save_plan_as_template( from .edit_plans_ai import router as ai_router from .edit_plans_clips import router as clips_router from .edit_plans_clips_batch import router as clips_batch_router +from .edit_plans_cover import router as cover_router from .edit_plans_generation import router as generation_router from .edit_plans_timeline import router as timeline_router @@ -613,3 +614,4 @@ router.include_router(ai_router) router.include_router(timeline_router) router.include_router(clips_router, prefix="/{plan_id}/clips", tags=["EditPlan Clips"]) router.include_router(clips_batch_router, prefix="/{plan_id}/clips", tags=["EditPlan Clips"]) +router.include_router(cover_router) diff --git a/apps/api/app/api/routes/edit_plans_cover.py b/apps/api/app/api/routes/edit_plans_cover.py new file mode 100755 index 000000000..13982ac6a --- /dev/null +++ b/apps/api/app/api/routes/edit_plans_cover.py @@ -0,0 +1,315 @@ +"""封面管理 API. + +- GET /{plan_id}/cover 获取封面配置 +- PUT /{plan_id}/cover 更新封面配置 +- POST /{plan_id}/cover/extract 从指定片段抽帧生成封面 +- POST /{plan_id}/cover/smart 智能选帧生成封面 +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, Optional + +from app.auth import AuthenticatedUser, get_current_user +from app.core.storage import get_storage_service +from app.dependencies import ( + get_asset_repository, + get_db_session, + get_project_repository, +) +from app.services import EditPlanService +from app.services.cover_service import CoverService +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from packages.domain.config_schemas import normalize_plan_config + +from ._helpers import check_project_access + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +# ── Schemas ────────────────────────────────────────────────────────────────── + + +class CoverConfigResponse(BaseModel): + """封面配置响应""" + + type: str = Field(..., description="封面类型: ai_frame / manual / upload") + image_url: str = Field(default="", description="封面图片 URL") + frame_time: Optional[float] = Field(default=None, description="抽帧时间点（秒）") + + +class CoverUpdateRequest(BaseModel): + """更新封面配置请求""" + + type: Optional[str] = Field(default=None, description="封面类型") + image_url: Optional[str] = Field(default=None, description="封面图片 URL") + frame_time: Optional[float] = Field(default=None, ge=0.0, description="抽帧时间点（秒）") + + +class CoverExtractRequest(BaseModel): + """从片段抽帧生成封面请求""" + + clip_id: str = Field(..., description="片段 ID") + frame_time: float = Field(1.0, ge=0.0, description="抽帧时间点（秒）") + + +class CoverSmartRequest(BaseModel): + """智能选帧请求""" + + clip_id: Optional[str] = Field(default=None, description="指定片段 ID（不传则用第一个视频片段）") + + +class CoverGenerateResponse(BaseModel): + """封面生成响应""" + + type: str = Field(..., description="封面类型") + image_url: str = Field(..., description="封面图片 URL") + frame_time: Optional[float] = Field(default=None, description="抽帧时间点（秒）") + + +# ── Routes ─────────────────────────────────────────────────────────────────── + + +@router.get("/{plan_id}/cover", response_model=CoverConfigResponse) +def get_cover( + plan_id: str, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> CoverConfigResponse: + """获取封面配置""" + svc = EditPlanService(db) + plan = svc.get_plan(plan_id) + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"剪辑计划不存在: {plan_id}", + ) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + cover = CoverService.get_cover_config(plan.config or {}) + return CoverConfigResponse(**cover) + + +@router.put("/{plan_id}/cover", response_model=CoverConfigResponse) +def update_cover( + plan_id: str, + body: CoverUpdateRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> CoverConfigResponse: + """更新封面配置 + + 用于：设置上传的封面图片 URL、切换封面类型、调整时间点等。 + """ + svc = EditPlanService(db) + plan = svc.get_plan(plan_id) + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"剪辑计划不存在: {plan_id}", + ) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + # 合并更新 + current_cover = CoverService.get_cover_config(plan.config or {}) + updates = body.model_dump(exclude_none=True) + new_cover = {**current_cover, **updates} + + # 验证 type 值 + valid_types = {"ai_frame", "manual", "upload", "ai_regenerate"} + if "type" in updates and updates["type"] not in valid_types: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"无效的封面类型: {updates['type']}，有效值: {valid_types}", + ) + + # 更新到 plan.config + current_config = dict(plan.config or {}) + current_config["cover"] = new_cover + normalized = normalize_plan_config(current_config) + updated_plan = svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + result = CoverService.get_cover_config(updated_plan.config or {}) + logger.info("更新封面配置: plan_id=%s type=%s by user=%s", plan_id, result["type"], current_user.user.id) + return CoverConfigResponse(**result) + + +@router.post("/{plan_id}/cover/extract", response_model=CoverGenerateResponse) +def extract_cover( + plan_id: str, + body: CoverExtractRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), + storage_service: Any = Depends(get_storage_service), + asset_repository: Any = Depends(get_asset_repository), +) -> CoverGenerateResponse: + """从指定片段的指定时间点抽帧生成封面""" + svc = EditPlanService(db) + plan = svc.get_plan(plan_id) + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"剪辑计划不存在: {plan_id}", + ) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + # 获取片段对应的素材 + clip = svc.get_clip(body.clip_id) + if not clip: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"片段不存在: {body.clip_id}", + ) + + if clip.plan_id != plan_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="片段不属于该剪辑计划", + ) + + if not clip.asset_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="片段没有关联素材，无法抽帧", + ) + + # 抽帧生成封面 + cover_svc = CoverService(storage_service, asset_repository) + try: + cover_data = cover_svc.extract_cover_from_clip( + plan_id=plan_id, + asset_id=clip.asset_id, + frame_time=body.frame_time, + ) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e + except RuntimeError as e: + logger.error("封面抽帧失败: plan_id=%s clip_id=%s error=%s", plan_id, body.clip_id, e) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"封面抽帧失败: {e}", + ) from e + + # 更新到 plan.config.cover + current_config = dict(plan.config or {}) + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + logger.info( + "封面抽帧完成: plan_id=%s clip_id=%s time=%.2fs by user=%s", + plan_id, + body.clip_id, + body.frame_time, + current_user.user.id, + ) + return CoverGenerateResponse(**cover_data) + + +@router.post("/{plan_id}/cover/smart", response_model=CoverGenerateResponse) +def smart_cover( + plan_id: str, + body: CoverSmartRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), + storage_service: Any = Depends(get_storage_service), + asset_repository: Any = Depends(get_asset_repository), +) -> CoverGenerateResponse: + """智能选帧生成封面 + + 从指定片段（或第一个视频片段）中智能选取一帧作为封面。 + 当前实现：取片段第3秒帧（后续可优化为多帧选最清晰）。 + """ + svc = EditPlanService(db) + plan = svc.get_plan(plan_id) + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"剪辑计划不存在: {plan_id}", + ) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + # 确定使用哪个片段 + clip_id = body.clip_id + asset_id = "" + + if clip_id: + clip = svc.get_clip(clip_id) + if not clip: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"片段不存在: {clip_id}", + ) + if clip.plan_id != plan_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="片段不属于该剪辑计划", + ) + if not clip.asset_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="片段没有关联素材", + ) + asset_id = clip.asset_id + else: + # 找第一个有素材的视频片段 + clips = svc.list_clips(plan_id, limit=50, skip=0) + for c in clips: + if c.asset_id and c.clip_type == "video": + asset_id = c.asset_id + clip_id = c.id + break + + if not asset_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="没有找到可用的视频片段", + ) + + # 智能选帧 + cover_svc = CoverService(storage_service, asset_repository) + try: + cover_data = cover_svc.generate_smart_cover( + plan_id=plan_id, + asset_id=asset_id, + ) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e + except RuntimeError as e: + logger.error("智能封面生成失败: plan_id=%s error=%s", plan_id, e) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"智能封面生成失败: {e}", + ) from e + + # 更新到 plan.config.cover + current_config = dict(plan.config or {}) + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + logger.info( + "智能封面生成完成: plan_id=%s clip_id=%s by user=%s", + plan_id, + clip_id, + current_user.user.id, + ) + return CoverGenerateResponse(**cover_data) diff --git a/apps/api/app/services/cover_service.py b/apps/api/app/services/cover_service.py new file mode 100644 index 000000000..d7bc14248 --- /dev/null +++ b/apps/api/app/services/cover_service.py @@ -0,0 +1,276 @@ +"""封面管理服务. + +提供封面配置管理和从视频抽帧生成封面的能力。 +抽帧使用 FFmpeg，上传使用共享存储服务。 +""" + +from __future__ import annotations + +import logging +import tempfile +from pathlib import Path +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +DEFAULT_COVER_WIDTH = 1080 +DEFAULT_COVER_HEIGHT = 1920 +DEFAULT_COVER_QUALITY = 5 # JPEG quality (1-31, 越小越好) +COVER_STORAGE_PREFIX = "covers" + + +class CoverService: + """封面管理服务.""" + + def __init__(self, storage_service: Any, asset_repository: Any) -> None: + self._storage = storage_service + self._asset_repo = asset_repository + + # ── 配置读写 ────────────────────────────────────────────────────────── + + @staticmethod + def get_cover_config(plan_config: Dict[str, Any]) -> Dict[str, Any]: + """从 plan.config 中提取封面配置. + + Args: + plan_config: 剪辑计划的 config 字段 + + Returns: + 封面配置 dict + """ + cover = plan_config.get("cover", {}) + if not isinstance(cover, dict): + cover = {} + # 确保默认字段存在 + return { + "type": cover.get("type", "ai_frame"), + "image_url": cover.get("image_url", ""), + "frame_time": cover.get("frame_time"), + } + + # ── 抽帧生成封面 ────────────────────────────────────────────────────── + + def extract_cover_from_clip( + self, + plan_id: str, + asset_id: str, + frame_time: float = 1.0, + *, + width: int = DEFAULT_COVER_WIDTH, + height: int = DEFAULT_COVER_HEIGHT, + quality: int = DEFAULT_COVER_QUALITY, + ) -> Dict[str, Any]: + """从指定素材的指定时间点抽取一帧作为封面. + + Args: + plan_id: 剪辑计划 ID（用于生成存储路径） + asset_id: 素材 ID + frame_time: 抽帧时间点（秒） + width: 输出宽度 + height: 输出高度 + quality: JPEG 质量 + + Returns: + 封面数据 dict，包含 type / image_url / frame_time + + Raises: + ValueError: 素材不存在或不是视频 + RuntimeError: 抽帧或上传失败 + """ + # 1. 获取素材 + asset = self._asset_repo.get(asset_id) if self._asset_repo else None + if not asset: + raise ValueError(f"素材不存在: {asset_id}") + + storage_key = getattr(asset, "storage_key", "") + if not storage_key: + raise ValueError(f"素材没有文件: {asset_id}") + + mime_type = getattr(asset, "mime_type", "") + if mime_type and not mime_type.startswith("video"): + raise ValueError(f"素材不是视频类型: {mime_type}") + + # 2. 下载视频到临时目录 + with tempfile.TemporaryDirectory(prefix="cover_extract_") as tmp_dir: + tmp_path = Path(tmp_dir) + video_path = tmp_path / f"source_{asset_id[:8]}" + + logger.info("下载素材用于封面抽帧: asset_id=%s", asset_id) + try: + self._storage.download_file(storage_key, str(video_path)) + except Exception as e: + raise RuntimeError(f"下载素材失败: {e}") from e + + if not video_path.exists() or video_path.stat().st_size == 0: + raise RuntimeError("下载的素材文件为空") + + # 3. FFmpeg 抽帧 + output_path = tmp_path / "cover.jpg" + self._extract_frame( + video_path=video_path, + output_path=output_path, + time_sec=frame_time, + width=width, + height=height, + quality=quality, + ) + + if not output_path.exists() or output_path.stat().st_size == 0: + raise RuntimeError("封面抽帧失败") + + # 4. 上传到 OSS + cover_key = f"{COVER_STORAGE_PREFIX}/{plan_id}/cover_{int(frame_time * 1000)}.jpg" + logger.info("上传封面到存储: key=%s", cover_key) + + try: + self._storage.upload_file( + file_or_path=str(output_path), + storage_key=cover_key, + content_type="image/jpeg", + ) + except Exception as e: + raise RuntimeError(f"上传封面失败: {e}") from e + + # 5. 获取访问 URL + try: + image_url = self._storage.get_url(cover_key) + except Exception: + image_url = cover_key # 降级为 storage_key + + logger.info( + "封面抽帧完成: plan_id=%s asset_id=%s time=%.2fs size=%d", + plan_id, + asset_id, + frame_time, + output_path.stat().st_size if output_path.exists() else 0, + ) + + return { + "type": "manual", + "image_url": image_url, + "frame_time": frame_time, + } + + def generate_smart_cover( + self, + plan_id: str, + asset_id: str, + *, + width: int = DEFAULT_COVER_WIDTH, + height: int = DEFAULT_COVER_HEIGHT, + quality: int = DEFAULT_COVER_QUALITY, + ) -> Dict[str, Any]: + """智能选帧：从视频中选取多帧，选最清晰的一帧. + + Args: + plan_id: 剪辑计划 ID + asset_id: 素材 ID + width: 输出宽度 + height: 输出高度 + quality: JPEG 质量 + + Returns: + 封面数据 dict + """ + # 简单实现：取视频 1/3 处的帧作为智能封面 + # 更复杂的多帧选清晰帧可以后续优化 + frame_time = 3.0 # 默认第3秒，后续可以根据视频时长动态计算 + + result = self.extract_cover_from_clip( + plan_id=plan_id, + asset_id=asset_id, + frame_time=frame_time, + width=width, + height=height, + quality=quality, + ) + result["type"] = "ai_frame" + return result + + # ── 内部方法 ────────────────────────────────────────────────────────── + + @staticmethod + def _extract_frame( + video_path: Path, + output_path: Path, + *, + time_sec: float, + width: int, + height: int, + quality: int, + ) -> None: + """使用 FFmpeg 从视频中抽取一帧. + + Args: + video_path: 视频文件路径 + output_path: 输出图片路径 + time_sec: 抽帧时间点（秒） + width: 输出宽度 + height: 输出高度 + quality: JPEG 质量 + """ + import subprocess + + # scale + crop 实现 cover 裁剪 + vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}" + + command = [ + "ffmpeg", + "-y", + "-ss", + f"{time_sec:.3f}", + "-i", + str(video_path), + "-vframes", + "1", + "-vf", + vf, + "-q:v", + str(quality), + "-f", + "mjpeg", + str(output_path), + ] + + logger.debug("FFmpeg 抽帧命令: %s", " ".join(command)) + + try: + result = subprocess.run( + command, + capture_output=True, + text=True, + timeout=60, + ) + if result.returncode != 0: + logger.warning("FFmpeg 抽帧返回非零: %s\nstderr: %s", result.returncode, result.stderr[-500:]) + # 尝试不使用 scale+crop 的简化命令 + simple_command = [ + "ffmpeg", + "-y", + "-ss", + f"{time_sec:.3f}", + "-i", + str(video_path), + "-vframes", + "1", + "-q:v", + str(quality), + "-f", + "mjpeg", + str(output_path), + ] + result2 = subprocess.run( + simple_command, + capture_output=True, + text=True, + timeout=60, + ) + if result2.returncode != 0: + raise RuntimeError(f"FFmpeg 抽帧失败: {result2.stderr[-300:]}") + except subprocess.TimeoutExpired as e: + raise RuntimeError("FFmpeg 抽帧超时") from e + except FileNotFoundError as e: + raise RuntimeError("FFmpeg 不可用") from e diff --git a/tests/unit/test_edit_plan_cover_api.py b/tests/unit/test_edit_plan_cover_api.py new file mode 100644 index 000000000..a38d1de01 --- /dev/null +++ b/tests/unit/test_edit_plan_cover_api.py @@ -0,0 +1,559 @@ +""" +封面管理 API 单元测试 + +覆盖： +- GET /{plan_id}/cover - 获取封面配置 +- PUT /{plan_id}/cover - 更新封面配置 +- POST /{plan_id}/cover/extract - 从片段抽帧 +- POST /{plan_id}/cover/smart - 智能选帧 +""" + +from __future__ import annotations + +import os +import sys +from datetime import datetime +from pathlib import Path +from typing import 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, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from packages.domain.config_schemas import normalize_plan_config +from packages.domain.edit_plan import EditPlan, EditPlanStatus +from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus + +# --------------------------------------------------------------------------- +# Stub Repository +# --------------------------------------------------------------------------- + + +class StubEditPlanRepository: + def __init__(self, plans: dict[str, EditPlan] | None = None): + self._plans = plans or {} + self._counter = 100 + + def _next_id(self) -> str: + self._counter += 1 + return f"plan-{self._counter:03d}" + + def list_all(self, *, status=None, skip=0, limit=50): + items = list(self._plans.values()) + if status is not None: + 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, *, status=None, skip=0, limit=50): + items = [p for p in self._plans.values() if p.template_id == template_id] + if status is not None: + 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._plans.get(plan_id) + + def create(self, plan: EditPlan) -> EditPlan: + if not plan.id: + plan.id = self._next_id() + self._plans[plan.id] = plan + return plan + + def update(self, plan: EditPlan) -> EditPlan: + self._plans[plan.id] = plan + return plan + + def delete(self, plan_id: str) -> bool: + if plan_id in self._plans: + del self._plans[plan_id] + return True + return False + + def count(self, *, status=None, template_id=None): + items = list(self._plans.values()) + if status is not None: + items = [p for p in items if p.status == status] + if template_id is not None: + items = [p for p in items if p.template_id == template_id] + return len(items) + + +class StubEditPlanClipRepository: + def __init__(self, clips: dict[str, EditPlanClip] | None = None): + self._clips = clips or {} + self._counter = 200 + + def _next_id(self) -> str: + self._counter += 1 + return f"clip-{self._counter:03d}" + + def list_by_plan(self, plan_id, *, status=None, skip=0, limit=100): + items = [c for c in self._clips.values() if c.plan_id == plan_id] + if status is not None: + items = [c for c in items if c.status == status] + items.sort(key=lambda c: c.order) + return items[skip : skip + limit] + + def count(self, plan_id, *, status=None): + items = [c for c in self._clips.values() if c.plan_id == plan_id] + if status is not None: + items = [c for c in items if c.status == status] + return len(items) + + def get(self, clip_id: str) -> Optional[EditPlanClip]: + return self._clips.get(clip_id) + + def create(self, clip: EditPlanClip) -> EditPlanClip: + if not clip.id: + clip.id = self._next_id() + self._clips[clip.id] = clip + return clip + + def update(self, clip: EditPlanClip) -> EditPlanClip: + self._clips[clip.id] = clip + return clip + + def delete(self, clip_id: str) -> bool: + if clip_id in self._clips: + del self._clips[clip_id] + return True + return False + + def delete_by_plan(self, plan_id: str) -> int: + to_delete = [cid for cid, c in self._clips.items() if c.plan_id == plan_id] + for cid in to_delete: + del self._clips[cid] + return len(to_delete) + + +class StubAssetRepository: + def __init__(self, assets: dict | None = None): + self._assets = assets or {} + + def get(self, asset_id: str): + return self._assets.get(asset_id) + + +class StubStorageService: + def __init__(self): + self.uploaded = {} + self.downloaded = {} + + def upload_file(self, file_or_path, storage_key, content_type="application/octet-stream"): + self.uploaded[storage_key] = file_or_path + return f"https://oss.example.com/{storage_key}" + + def get_url(self, storage_key: str) -> str: + return f"https://oss.example.com/{storage_key}" + + def download_file(self, storage_key: str, local_path: str): + self.downloaded[storage_key] = local_path + # 创建一个假文件（空文件也可以，因为抽帧会被 mock 掉） + Path(local_path).parent.mkdir(parents=True, exist_ok=True) + with open(local_path, "wb") as f: + f.write(b"fake video data for testing") + + +# --------------------------------------------------------------------------- +# Test Fixtures +# --------------------------------------------------------------------------- + + +def _make_sample_plan(plan_id="plan-001", config=None): + if config is None: + config = normalize_plan_config({}) + return EditPlan( + id=plan_id, + template_id="tpl-001", + name="测试计划", + status=EditPlanStatus.EDITING, + total_duration=30.0, + config=config, + project_id="", + created_by_user_id="user-001", + created_at=datetime(2026, 7, 16, 10, 0, 0), + updated_at=datetime(2026, 7, 16, 10, 0, 0), + ) + + +def _make_sample_clip(clip_id="clip-001", plan_id="plan-001", asset_id="asset-001", clip_type="video"): + return EditPlanClip( + id=clip_id, + plan_id=plan_id, + clip_type=clip_type, + order=0, + asset_id=asset_id, + text_content="", + start_time=0.0, + duration=10.0, + transition_effect="none", + transition_duration=0.0, + playback_speed=1.0, + status=EditPlanClipStatus.READY, + config={}, + created_at=datetime(2026, 7, 16, 10, 0, 0), + updated_at=datetime(2026, 7, 16, 10, 0, 0), + ) + + +def _create_test_app(): + import app.api.routes.edit_plans_cover as cover_module + import app.services.edit_plan_service as service_module + from app.api.routes.edit_plans import router + + # 创建 stub + plan = _make_sample_plan() + clip = _make_sample_clip() + stub_plan_repo = StubEditPlanRepository({plan.id: plan}) + stub_clip_repo = StubEditPlanClipRepository({clip.id: clip}) + + # 替换服务模块中的 Repository 类 + original_plan_repo = service_module.SQLAlchemyEditPlanRepository + original_clip_repo = service_module.SQLAlchemyEditPlanClipRepository + original_gen_repo = service_module.SQLAlchemyGenerationTaskRepository + service_module.SQLAlchemyEditPlanRepository = lambda db: stub_plan_repo + service_module.SQLAlchemyEditPlanClipRepository = lambda db: stub_clip_repo + service_module.SQLAlchemyGenerationTaskRepository = lambda db: MagicMock() + + app = FastAPI() + app.include_router(router, prefix="/api/v1/edit-plans") + + # Mock 认证 + def _mock_auth(): + mock = MagicMock() + mock.user.id = "user-001" + return mock + + # Mock 项目访问检查 + import app.api.routes._helpers as helpers_module + + original_check = helpers_module.check_project_access + helpers_module.check_project_access = lambda *a, **kw: None + + # 覆盖依赖 + app.dependency_overrides[cover_module.get_current_user] = _mock_auth + app.dependency_overrides[cover_module.get_db_session] = lambda: MagicMock() + app.dependency_overrides[cover_module.get_project_repository] = lambda: MagicMock() + + # Mock storage 和 asset repo + stub_storage = StubStorageService() + stub_asset_repo = StubAssetRepository( + { + "asset-001": MagicMock( + storage_key="videos/test.mp4", + mime_type="video/mp4", + ), + "asset-img": MagicMock( + storage_key="images/test.jpg", + mime_type="image/jpeg", + ), + } + ) + + app.dependency_overrides[cover_module.get_storage_service] = lambda: stub_storage + app.dependency_overrides[cover_module.get_asset_repository] = lambda: stub_asset_repo + + # 也需要覆盖 edit_plans 主模块的 auth（用于其他路由） + from app.api.routes import edit_plans as main_module + + app.dependency_overrides[main_module.get_current_user] = _mock_auth + app.dependency_overrides[main_module.get_db_session] = lambda: MagicMock() + app.dependency_overrides[main_module.get_project_repository] = lambda: MagicMock() + + def cleanup(): + service_module.SQLAlchemyEditPlanRepository = original_plan_repo + service_module.SQLAlchemyEditPlanClipRepository = original_clip_repo + service_module.SQLAlchemyGenerationTaskRepository = original_gen_repo + helpers_module.check_project_access = original_check + + return app, stub_plan_repo, stub_clip_repo, stub_storage, cleanup + + +@pytest.fixture +def cover_client(): + app, plan_repo, clip_repo, storage, cleanup = _create_test_app() + yield TestClient(app), plan_repo, clip_repo, storage + cleanup() + + +# --------------------------------------------------------------------------- +# GET /{plan_id}/cover 测试 +# --------------------------------------------------------------------------- + + +class TestGetCover: + def test_get_default_cover(self, cover_client): + c, _, _, _ = cover_client + resp = c.get("/api/v1/edit-plans/plan-001/cover") + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "ai_frame" + assert data["image_url"] == "" + assert data["frame_time"] is None + + def test_get_cover_not_found(self, cover_client): + c, _, _, _ = cover_client + resp = c.get("/api/v1/edit-plans/plan-nonexist/cover") + assert resp.status_code == 404 + + def test_get_cover_with_custom_config(self, cover_client): + c, plan_repo, _, _ = cover_client + # 更新 plan 的 cover 配置 + plan = plan_repo.get("plan-001") + new_config = dict(plan.config) + new_config["cover"] = {"type": "manual", "image_url": "https://example.com/cover.jpg", "frame_time": 5.5} + plan.config = new_config + plan_repo.update(plan) + + resp = c.get("/api/v1/edit-plans/plan-001/cover") + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "manual" + assert data["image_url"] == "https://example.com/cover.jpg" + assert data["frame_time"] == 5.5 + + +# --------------------------------------------------------------------------- +# PUT /{plan_id}/cover 测试 +# --------------------------------------------------------------------------- + + +class TestUpdateCover: + def test_update_cover_type_and_url(self, cover_client): + c, plan_repo, _, _ = cover_client + resp = c.put( + "/api/v1/edit-plans/plan-001/cover", + json={"type": "upload", "image_url": "https://example.com/uploaded.jpg"}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "upload" + assert data["image_url"] == "https://example.com/uploaded.jpg" + + # 验证存储 + plan = plan_repo.get("plan-001") + assert plan.config["cover"]["type"] == "upload" + assert plan.config["cover"]["image_url"] == "https://example.com/uploaded.jpg" + + def test_update_cover_frame_time(self, cover_client): + c, plan_repo, _, _ = cover_client + resp = c.put( + "/api/v1/edit-plans/plan-001/cover", + json={"type": "manual", "frame_time": 3.14}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "manual" + assert data["frame_time"] == 3.14 + + plan = plan_repo.get("plan-001") + assert plan.config["cover"]["frame_time"] == 3.14 + + def test_update_cover_invalid_type(self, cover_client): + c, _, _, _ = cover_client + resp = c.put( + "/api/v1/edit-plans/plan-001/cover", + json={"type": "invalid_type"}, + ) + assert resp.status_code == 400 + + def test_update_cover_not_found(self, cover_client): + c, _, _, _ = cover_client + resp = c.put( + "/api/v1/edit-plans/plan-nonexist/cover", + json={"type": "upload", "image_url": "test.jpg"}, + ) + assert resp.status_code == 404 + + def test_update_cover_partial(self, cover_client): + """只更新 image_url，type 保持不变""" + c, plan_repo, _, _ = cover_client + # 先设置一个类型 + c.put("/api/v1/edit-plans/plan-001/cover", json={"type": "manual", "frame_time": 2.0}) + + # 只更新 image_url + resp = c.put( + "/api/v1/edit-plans/plan-001/cover", + json={"image_url": "https://example.com/new.jpg"}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "manual" # 保持不变 + assert data["image_url"] == "https://example.com/new.jpg" + assert data["frame_time"] == 2.0 # 保持不变 + + +# --------------------------------------------------------------------------- +# POST /{plan_id}/cover/extract 测试 +# --------------------------------------------------------------------------- + + +class TestExtractCover: + def test_extract_success(self, cover_client): + c, plan_repo, _, _ = cover_client + with patch("app.services.cover_service.CoverService._extract_frame") as mock_extract: + # mock ffmpeg 抽帧，直接创建输出文件 + def fake_extract(video_path, output_path, **kwargs): + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake jpeg data") + + mock_extract.side_effect = fake_extract + + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/extract", + json={"clip_id": "clip-001", "frame_time": 2.5}, + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "manual" + assert data["frame_time"] == 2.5 + assert data["image_url"].startswith("https://oss.example.com/covers/") + + # 验证 plan.config 已更新 + plan = plan_repo.get("plan-001") + assert plan.config["cover"]["type"] == "manual" + assert plan.config["cover"]["frame_time"] == 2.5 + + def test_extract_clip_not_found(self, cover_client): + c, _, _, _ = cover_client + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/extract", + json={"clip_id": "clip-nonexist", "frame_time": 1.0}, + ) + assert resp.status_code == 404 + + def test_extract_plan_not_found(self, cover_client): + c, _, _, _ = cover_client + resp = c.post( + "/api/v1/edit-plans/plan-nonexist/cover/extract", + json={"clip_id": "clip-001", "frame_time": 1.0}, + ) + assert resp.status_code == 404 + + def test_extract_clip_no_asset(self, cover_client): + c, _, clip_repo, _ = cover_client + # 创建一个没有 asset 的片段 + empty_clip = _make_sample_clip(clip_id="clip-empty", asset_id="") + clip_repo.create(empty_clip) + + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/extract", + json={"clip_id": "clip-empty", "frame_time": 1.0}, + ) + assert resp.status_code == 400 + assert "没有关联素材" in resp.json()["detail"] + + def test_extract_clip_not_in_plan(self, cover_client): + c, _, clip_repo, _ = cover_client + # 创建属于另一个 plan 的片段 + other_clip = _make_sample_clip(clip_id="clip-other", plan_id="plan-other") + clip_repo.create(other_clip) + + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/extract", + json={"clip_id": "clip-other", "frame_time": 1.0}, + ) + assert resp.status_code == 400 + assert "不属于该剪辑计划" in resp.json()["detail"] + + def test_extract_negative_frame_time(self, cover_client): + c, _, _, _ = cover_client + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/extract", + json={"clip_id": "clip-001", "frame_time": -1.0}, + ) + assert resp.status_code == 422 # pydantic 校验失败 + + +# --------------------------------------------------------------------------- +# POST /{plan_id}/cover/smart 测试 +# --------------------------------------------------------------------------- + + +class TestSmartCover: + def test_smart_cover_with_clip_id(self, cover_client): + c, plan_repo, _, _ = cover_client + with patch("app.services.cover_service.CoverService._extract_frame") as mock_extract: + + def fake_extract(video_path, output_path, **kwargs): + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake jpeg data") + + mock_extract.side_effect = fake_extract + + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/smart", + json={"clip_id": "clip-001"}, + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "ai_frame" + assert data["image_url"].startswith("https://oss.example.com/covers/") + + plan = plan_repo.get("plan-001") + assert plan.config["cover"]["type"] == "ai_frame" + + def test_smart_cover_auto_pick_first_video(self, cover_client): + c, plan_repo, clip_repo, _ = cover_client + # 添加多个片段，第一个视频应该被选中 + clip2 = _make_sample_clip(clip_id="clip-002", clip_type="audio", asset_id="asset-audio") + clip2.order = 1 + clip_repo.create(clip2) + + with patch("app.services.cover_service.CoverService._extract_frame") as mock_extract: + + def fake_extract(video_path, output_path, **kwargs): + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake jpeg data") + + mock_extract.side_effect = fake_extract + + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/smart", + json={}, + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["type"] == "ai_frame" + + def test_smart_cover_clip_not_found(self, cover_client): + c, _, _, _ = cover_client + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/smart", + json={"clip_id": "clip-nonexist"}, + ) + assert resp.status_code == 404 + + def test_smart_cover_no_video_clips(self, cover_client): + c, _, clip_repo, _ = cover_client + # 删除原有片段，添加纯音频片段 + clip_repo.delete("clip-001") + audio_clip = _make_sample_clip(clip_id="clip-audio", clip_type="audio", asset_id="asset-001") + clip_repo.create(audio_clip) + + resp = c.post( + "/api/v1/edit-plans/plan-001/cover/smart", + json={}, + ) + assert resp.status_code == 400 + assert "没有找到可用的视频片段" in resp.json()["detail"] + + def test_smart_cover_plan_not_found(self, cover_client): + c, _, _, _ = cover_client + resp = c.post( + "/api/v1/edit-plans/plan-nonexist/cover/smart", + json={}, + ) + assert resp.status_code == 404