feat: 封面管理 - 配置读写 + 片段抽帧 + 智能选帧 #396
@@ -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)
|
||||
|
||||
Executable
+315
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user