feat: 转场特效 - 预设库 + 单片段设置 + 批量应用 #398
@@ -265,9 +265,11 @@ def _to_response(p: EditPlan) -> EditPlanResponse:
|
||||
|
||||
from .edit_plans_export import router as export_router
|
||||
from .edit_plans_filter import router as filter_router
|
||||
from .edit_plans_transitions import router as transitions_router
|
||||
|
||||
router.include_router(export_router)
|
||||
router.include_router(filter_router)
|
||||
router.include_router(transitions_router)
|
||||
|
||||
|
||||
# ── CRUD Routes ───────────────────────────────────────────────────────────────
|
||||
|
||||
+272
@@ -0,0 +1,272 @@
|
||||
"""转场特效 API.
|
||||
|
||||
- GET /transition-presets 转场预设列表
|
||||
- PUT /clips/{clip_id}/transition 设置单个片段转场
|
||||
- POST /{plan_id}/transitions/batch 批量设置转场（所有片段）
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, List, Optional
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
from app.dependencies import get_db_session, get_project_repository
|
||||
from app.services import EditPlanService
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from packages.domain.transition_presets import (
|
||||
TransitionPreset,
|
||||
get_transition_preset,
|
||||
list_transition_presets,
|
||||
)
|
||||
|
||||
from ._helpers import check_project_access
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# ── Schemas ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TransitionPresetResponse(BaseModel):
|
||||
"""转场预设响应"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
category: str
|
||||
description: str
|
||||
tags: List[str] = Field(default_factory=list)
|
||||
default_duration: float
|
||||
min_duration: float
|
||||
max_duration: float
|
||||
|
||||
|
||||
class TransitionUpdateRequest(BaseModel):
|
||||
"""更新转场请求"""
|
||||
|
||||
effect: str = Field(..., description="转场效果 ID")
|
||||
duration: Optional[float] = Field(default=None, ge=0.0, description="转场时长（秒）")
|
||||
|
||||
|
||||
class BatchTransitionRequest(BaseModel):
|
||||
"""批量设置转场请求"""
|
||||
|
||||
effect: str = Field(..., description="转场效果 ID")
|
||||
duration: Optional[float] = Field(default=None, ge=0.0, description="转场时长（秒）")
|
||||
apply_to: str = Field(
|
||||
default="all",
|
||||
description="应用范围: all=所有片段, except_first=除第一个外, except_last=除最后一个, middle=中间片段",
|
||||
)
|
||||
|
||||
|
||||
class ClipTransitionResponse(BaseModel):
|
||||
"""片段转场信息响应"""
|
||||
|
||||
clip_id: str
|
||||
effect: str
|
||||
duration: float
|
||||
|
||||
|
||||
class BatchTransitionResponse(BaseModel):
|
||||
"""批量转场响应"""
|
||||
|
||||
updated_count: int
|
||||
plan_id: str
|
||||
|
||||
|
||||
class TransitionPresetListResponse(BaseModel):
|
||||
"""转场预设列表响应"""
|
||||
|
||||
items: List[TransitionPresetResponse]
|
||||
total: int
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _preset_to_response(p: TransitionPreset) -> TransitionPresetResponse:
|
||||
return TransitionPresetResponse(
|
||||
id=p.id,
|
||||
name=p.name,
|
||||
category=p.category,
|
||||
description=p.description,
|
||||
tags=list(p.tags),
|
||||
default_duration=p.default_duration,
|
||||
min_duration=p.min_duration,
|
||||
max_duration=p.max_duration,
|
||||
)
|
||||
|
||||
|
||||
def _validate_transition(effect: str, duration: Optional[float] = None) -> tuple[str, float]:
|
||||
"""验证转场效果和时长，返回 (effect, duration)"""
|
||||
preset = get_transition_preset(effect)
|
||||
if preset is None:
|
||||
raise ValueError(f"无效的转场效果: {effect}")
|
||||
|
||||
# 硬切特殊处理，时长强制为0
|
||||
if effect == "transition_none" or preset.transition == "none":
|
||||
return "cut", 0.0
|
||||
|
||||
final_duration = duration if duration is not None else preset.default_duration
|
||||
if final_duration < preset.min_duration:
|
||||
final_duration = preset.min_duration
|
||||
if final_duration > preset.max_duration:
|
||||
final_duration = preset.max_duration
|
||||
|
||||
return preset.transition, round(final_duration, 3)
|
||||
|
||||
|
||||
# ── Routes ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("/transition-presets", response_model=TransitionPresetListResponse)
|
||||
def list_presets(
|
||||
category: Optional[str] = Query(default=None, description="按分类筛选"),
|
||||
keyword: Optional[str] = Query(default=None, description="关键词搜索"),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> TransitionPresetListResponse:
|
||||
"""获取转场预设列表"""
|
||||
presets = list_transition_presets(category=category, keyword=keyword)
|
||||
items = [_preset_to_response(p) for p in presets]
|
||||
return TransitionPresetListResponse(items=items, total=len(items))
|
||||
|
||||
|
||||
@router.put("/clips/{clip_id}/transition", response_model=ClipTransitionResponse)
|
||||
def update_clip_transition(
|
||||
clip_id: str,
|
||||
body: TransitionUpdateRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
project_repository: Any = Depends(get_project_repository),
|
||||
) -> ClipTransitionResponse:
|
||||
"""设置单个片段的转场效果"""
|
||||
svc = EditPlanService(db)
|
||||
clip = svc.get_clip(clip_id)
|
||||
if not clip:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"片段不存在: {clip_id}",
|
||||
)
|
||||
|
||||
plan = svc.get_plan(clip.plan_id)
|
||||
if plan and plan.project_id:
|
||||
check_project_access(plan.project_id, current_user.user.id, project_repository)
|
||||
|
||||
# 验证转场参数
|
||||
try:
|
||||
effect, duration = _validate_transition(body.effect, body.duration)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
|
||||
|
||||
# 更新片段
|
||||
updated_clip = svc.update_clip(
|
||||
clip_id,
|
||||
transition_effect=effect,
|
||||
transition_duration=duration,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"更新片段转场: clip_id=%s effect=%s duration=%.3f by user=%s",
|
||||
clip_id,
|
||||
effect,
|
||||
duration,
|
||||
current_user.user.id,
|
||||
)
|
||||
return ClipTransitionResponse(
|
||||
clip_id=clip_id,
|
||||
effect=updated_clip.transition_effect,
|
||||
duration=updated_clip.transition_duration,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{plan_id}/transitions/batch", response_model=BatchTransitionResponse)
|
||||
def batch_update_transitions(
|
||||
plan_id: str,
|
||||
body: BatchTransitionRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
project_repository: Any = Depends(get_project_repository),
|
||||
) -> BatchTransitionResponse:
|
||||
"""批量设置计划内所有片段的转场效果
|
||||
|
||||
apply_to 说明：
|
||||
- all: 所有片段
|
||||
- except_first: 除第一个片段外（第一个片段不需要前转场）
|
||||
- except_last: 除最后一个片段外
|
||||
- middle: 只设置中间片段（除首尾）
|
||||
"""
|
||||
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)
|
||||
|
||||
# 验证转场参数
|
||||
try:
|
||||
effect, duration = _validate_transition(body.effect, body.duration)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
|
||||
|
||||
# 获取所有片段
|
||||
clips = svc.list_clips(plan_id, limit=500, skip=0)
|
||||
if not clips:
|
||||
return BatchTransitionResponse(updated_count=0, plan_id=plan_id)
|
||||
|
||||
# 确定应用范围
|
||||
total = len(clips)
|
||||
if total <= 1:
|
||||
# 只有一个片段时，只有 all 模式才应用
|
||||
if body.apply_to != "all":
|
||||
return BatchTransitionResponse(updated_count=0, plan_id=plan_id)
|
||||
|
||||
# 按 order 排序
|
||||
clips_sorted = sorted(clips, key=lambda c: c.order)
|
||||
indices_to_update = []
|
||||
|
||||
if body.apply_to == "all":
|
||||
indices_to_update = list(range(total))
|
||||
elif body.apply_to == "except_first":
|
||||
indices_to_update = list(range(1, total))
|
||||
elif body.apply_to == "except_last":
|
||||
indices_to_update = list(range(total - 1))
|
||||
elif body.apply_to == "middle":
|
||||
if total <= 2:
|
||||
indices_to_update = []
|
||||
else:
|
||||
indices_to_update = list(range(1, total - 1))
|
||||
else:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"无效的 apply_to: {body.apply_to}",
|
||||
)
|
||||
|
||||
# 批量更新
|
||||
count = 0
|
||||
for idx in indices_to_update:
|
||||
clip = clips_sorted[idx]
|
||||
svc.update_clip(
|
||||
clip.id,
|
||||
transition_effect=effect,
|
||||
transition_duration=duration,
|
||||
)
|
||||
count += 1
|
||||
|
||||
logger.info(
|
||||
"批量更新转场: plan_id=%s count=%d effect=%s apply_to=%s by user=%s",
|
||||
plan_id,
|
||||
count,
|
||||
effect,
|
||||
body.apply_to,
|
||||
current_user.user.id,
|
||||
)
|
||||
return BatchTransitionResponse(updated_count=count, plan_id=plan_id)
|
||||
Executable
+306
@@ -0,0 +1,306 @@
|
||||
"""转场特效预设库.
|
||||
|
||||
视频片段之间的转场效果，基于 FFmpeg xfade 滤镜实现。
|
||||
所有转场预设包含时长范围和默认参数。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TransitionPreset:
|
||||
"""转场特效预设"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
category: str # 分类：basic / fade / slide / zoom / warp / special
|
||||
description: str = ""
|
||||
tags: List[str] = field(default_factory=list)
|
||||
# FFmpeg xfade transition 名称
|
||||
transition: str = "fade"
|
||||
# 默认时长（秒）
|
||||
default_duration: float = 0.5
|
||||
# 支持的时长范围
|
||||
min_duration: float = 0.1
|
||||
max_duration: float = 3.0
|
||||
# 是否需要额外参数
|
||||
has_custom_params: bool = False
|
||||
|
||||
|
||||
# ── 预设库清单 ────────────────────────────────────────────────────────────────
|
||||
|
||||
TRANSITION_PRESET_LIBRARY: List[TransitionPreset] = [
|
||||
# ── 基础 basic ────────────────────────────────────────────────────
|
||||
TransitionPreset(
|
||||
id="transition_none",
|
||||
name="无转场",
|
||||
category="basic",
|
||||
description="硬切，无过渡效果",
|
||||
tags=["硬切", "无"],
|
||||
transition="none",
|
||||
default_duration=0.0,
|
||||
min_duration=0.0,
|
||||
max_duration=0.0,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_random",
|
||||
name="随机",
|
||||
category="basic",
|
||||
description="随机选择一个转场效果",
|
||||
tags=["随机", "惊喜"],
|
||||
transition="random",
|
||||
default_duration=0.5,
|
||||
),
|
||||
# ── 淡入淡出 fade ────────────────────────────────────────────────
|
||||
TransitionPreset(
|
||||
id="transition_fade",
|
||||
name="淡入淡出",
|
||||
category="fade",
|
||||
description="经典交叉淡入淡出",
|
||||
tags=["经典", "柔和"],
|
||||
transition="fade",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=2.0,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_fadeblack",
|
||||
name="黑场过渡",
|
||||
category="fade",
|
||||
description="经过黑色画面过渡",
|
||||
tags=["黑场", "电影感"],
|
||||
transition="fadeblack",
|
||||
default_duration=0.6,
|
||||
min_duration=0.2,
|
||||
max_duration=2.0,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_fadewhite",
|
||||
name="白场过渡",
|
||||
category="fade",
|
||||
description="经过白色画面过渡",
|
||||
tags=["白场", "梦幻"],
|
||||
transition="fadewhite",
|
||||
default_duration=0.6,
|
||||
min_duration=0.2,
|
||||
max_duration=2.0,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_fadegrays",
|
||||
name="灰度过渡",
|
||||
category="fade",
|
||||
description="经过灰度画面过渡",
|
||||
tags=["灰度", "文艺"],
|
||||
transition="fadegrays",
|
||||
default_duration=0.6,
|
||||
min_duration=0.2,
|
||||
max_duration=2.0,
|
||||
),
|
||||
# ── 滑动 slide ──────────────────────────────────────────────────
|
||||
TransitionPreset(
|
||||
id="transition_slideleft",
|
||||
name="左滑",
|
||||
category="slide",
|
||||
description="画面向左滑动",
|
||||
tags=["滑动", "左"],
|
||||
transition="slideleft",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_slideright",
|
||||
name="右滑",
|
||||
category="slide",
|
||||
description="画面向右滑动",
|
||||
tags=["滑动", "右"],
|
||||
transition="slideright",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_slideup",
|
||||
name="上滑",
|
||||
category="slide",
|
||||
description="画面向上滑动",
|
||||
tags=["滑动", "上"],
|
||||
transition="slideup",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_slidedown",
|
||||
name="下滑",
|
||||
category="slide",
|
||||
description="画面向下滑动",
|
||||
tags=["滑动", "下"],
|
||||
transition="slidedown",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
# ── 缩放 zoom ───────────────────────────────────────────────────
|
||||
TransitionPreset(
|
||||
id="transition_zoomin",
|
||||
name="放大进入",
|
||||
category="zoom",
|
||||
description="下一段画面从中心放大出现",
|
||||
tags=["放大", "冲击"],
|
||||
transition="zoomin",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_zoomout",
|
||||
name="缩小退出",
|
||||
category="zoom",
|
||||
description="当前画面缩小退出",
|
||||
tags=["缩小", "拉远"],
|
||||
transition="zoomout",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
# ── 擦除 warp ───────────────────────────────────────────────────
|
||||
TransitionPreset(
|
||||
id="transition_dissolve",
|
||||
name="溶解",
|
||||
category="warp",
|
||||
description="像素溶解效果",
|
||||
tags=["溶解", "像素"],
|
||||
transition="dissolve",
|
||||
default_duration=0.8,
|
||||
min_duration=0.3,
|
||||
max_duration=2.0,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_wipeleft",
|
||||
name="左擦除",
|
||||
category="warp",
|
||||
description="从右向左擦除",
|
||||
tags=["擦除", "左"],
|
||||
transition="wipeleft",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_wiperight",
|
||||
name="右擦除",
|
||||
category="warp",
|
||||
description="从左向右擦除",
|
||||
tags=["擦除", "右"],
|
||||
transition="wiperight",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_wipeup",
|
||||
name="上擦除",
|
||||
category="warp",
|
||||
description="从下向上擦除",
|
||||
tags=["擦除", "上"],
|
||||
transition="wipeup",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_wipedown",
|
||||
name="下擦除",
|
||||
category="warp",
|
||||
description="从上向下擦除",
|
||||
tags=["擦除", "下"],
|
||||
transition="wipedown",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_circlecrop",
|
||||
name="圆形展开",
|
||||
category="warp",
|
||||
description="圆形从中心展开",
|
||||
tags=["圆形", "展开"],
|
||||
transition="circlecrop",
|
||||
default_duration=0.6,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
# ── 特效 special ───────────────────────────────────────────────
|
||||
TransitionPreset(
|
||||
id="transition_hblur",
|
||||
name="水平模糊",
|
||||
category="special",
|
||||
description="水平方向模糊过渡",
|
||||
tags=["模糊", "水平"],
|
||||
transition="hblur",
|
||||
default_duration=0.5,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
TransitionPreset(
|
||||
id="transition_wipeblur",
|
||||
name="模糊擦除",
|
||||
category="special",
|
||||
description="带模糊效果的擦除",
|
||||
tags=["模糊", "擦除"],
|
||||
transition="wipeblur",
|
||||
default_duration=0.6,
|
||||
min_duration=0.2,
|
||||
max_duration=1.5,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
# ── 工具函数 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_transition_preset(preset_id: str) -> Optional[TransitionPreset]:
|
||||
"""根据 ID 获取转场预设"""
|
||||
for p in TRANSITION_PRESET_LIBRARY:
|
||||
if p.id == preset_id:
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def list_transition_presets(
|
||||
*,
|
||||
category: Optional[str] = None,
|
||||
keyword: Optional[str] = None,
|
||||
) -> List[TransitionPreset]:
|
||||
"""筛选转场预设列表
|
||||
|
||||
Args:
|
||||
category: 按分类筛选
|
||||
keyword: 关键词搜索
|
||||
|
||||
Returns:
|
||||
筛选后的预设列表
|
||||
"""
|
||||
results = TRANSITION_PRESET_LIBRARY
|
||||
|
||||
if category:
|
||||
results = [p for p in results if p.category == category]
|
||||
|
||||
if keyword:
|
||||
kw = keyword.lower()
|
||||
results = [
|
||||
p
|
||||
for p in results
|
||||
if kw in p.name.lower() or kw in p.description.lower() or any(kw in t.lower() for t in p.tags)
|
||||
]
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def get_default_transition() -> TransitionPreset:
|
||||
"""获取默认转场（无转场/硬切）"""
|
||||
return get_transition_preset("transition_none") # type: ignore
|
||||
@@ -0,0 +1,494 @@
|
||||
"""
|
||||
转场特效 API 单元测试
|
||||
|
||||
覆盖：
|
||||
- GET /transition-presets - 转场预设列表
|
||||
- PUT /clips/{clip_id}/transition - 设置单个片段转场
|
||||
- POST /{plan_id}/transitions/batch - 批量设置转场
|
||||
"""
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
from packages.domain.transition_presets import TRANSITION_PRESET_LIBRARY
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stub Repository
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class StubEditPlanRepository:
|
||||
def __init__(self, plans: dict[str, EditPlan] | None = None):
|
||||
self._plans = plans or {}
|
||||
|
||||
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:
|
||||
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 = 0
|
||||
|
||||
def _next_id(self) -> str:
|
||||
self._counter += 1
|
||||
return f"clip-new{self._counter}"
|
||||
|
||||
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)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_sample_plan(plan_id="plan-001"):
|
||||
return EditPlan(
|
||||
id=plan_id,
|
||||
template_id="tpl-001",
|
||||
name="测试计划",
|
||||
status=EditPlanStatus.EDITING,
|
||||
total_duration=30.0,
|
||||
config=normalize_plan_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_clip(clip_id, plan_id="plan-001", order=0, transition_effect="cut", transition_duration=0.0):
|
||||
return EditPlanClip(
|
||||
id=clip_id,
|
||||
plan_id=plan_id,
|
||||
clip_type="video",
|
||||
order=order,
|
||||
asset_id="asset-001",
|
||||
text_content="",
|
||||
start_time=0.0,
|
||||
duration=10.0,
|
||||
transition_effect=transition_effect,
|
||||
transition_duration=transition_duration,
|
||||
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_transitions as transitions_module
|
||||
import app.services.edit_plan_service as service_module
|
||||
from app.api.routes.edit_plans import router
|
||||
|
||||
plan = _make_sample_plan()
|
||||
clips = {
|
||||
"clip-001": _make_clip("clip-001", order=0),
|
||||
"clip-002": _make_clip("clip-002", order=1),
|
||||
"clip-003": _make_clip("clip-003", order=2),
|
||||
"clip-004": _make_clip("clip-004", order=3),
|
||||
}
|
||||
stub_plan_repo = StubEditPlanRepository({plan.id: plan})
|
||||
stub_clip_repo = StubEditPlanClipRepository(clips)
|
||||
|
||||
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
|
||||
|
||||
# 覆盖依赖
|
||||
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()
|
||||
|
||||
app.dependency_overrides[transitions_module.get_current_user] = _mock_auth
|
||||
app.dependency_overrides[transitions_module.get_db_session] = lambda: MagicMock()
|
||||
app.dependency_overrides[transitions_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, cleanup
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def transition_client():
|
||||
app, plan_repo, clip_repo, cleanup = _create_test_app()
|
||||
yield TestClient(app), plan_repo, clip_repo
|
||||
cleanup()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Transition Presets 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTransitionPresets:
|
||||
def test_list_all_presets(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.get("/api/v1/edit-plans/transition-presets")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == len(TRANSITION_PRESET_LIBRARY)
|
||||
assert data["total"] > 10
|
||||
first = data["items"][0]
|
||||
assert "id" in first
|
||||
assert "name" in first
|
||||
assert "category" in first
|
||||
assert "default_duration" in first
|
||||
assert "min_duration" in first
|
||||
assert "max_duration" in first
|
||||
|
||||
def test_filter_by_category_fade(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.get("/api/v1/edit-plans/transition-presets?category=fade")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] >= 3
|
||||
for item in data["items"]:
|
||||
assert item["category"] == "fade"
|
||||
|
||||
def test_filter_by_category_slide(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.get("/api/v1/edit-plans/transition-presets?category=slide")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] >= 4
|
||||
for item in data["items"]:
|
||||
assert item["category"] == "slide"
|
||||
|
||||
def test_filter_by_keyword(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.get("/api/v1/edit-plans/transition-presets?keyword=模糊")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] > 0
|
||||
names = [item["name"] for item in data["items"]]
|
||||
assert any("模糊" in n for n in names)
|
||||
|
||||
def test_filter_empty_result(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.get("/api/v1/edit-plans/transition-presets?keyword=不存在的转场")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 0
|
||||
|
||||
def test_contains_none_transition(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.get("/api/v1/edit-plans/transition-presets?category=basic")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
ids = [item["id"] for item in data["items"]]
|
||||
assert "transition_none" in ids
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PUT /clips/{clip_id}/transition 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUpdateClipTransition:
|
||||
def test_set_fade_transition(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-001/transition",
|
||||
json={"effect": "transition_fade", "duration": 0.8},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["clip_id"] == "clip-001"
|
||||
assert data["effect"] == "fade"
|
||||
assert data["duration"] == 0.8
|
||||
|
||||
clip = clip_repo.get("clip-001")
|
||||
assert clip.transition_effect == "fade"
|
||||
assert clip.transition_duration == 0.8
|
||||
|
||||
def test_set_none_transition(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-001/transition",
|
||||
json={"effect": "transition_none"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["effect"] == "cut"
|
||||
assert data["duration"] == 0.0
|
||||
|
||||
def test_use_default_duration(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
# 不传 duration，使用预设默认值
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-001/transition",
|
||||
json={"effect": "transition_fade"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["effect"] == "fade"
|
||||
assert data["duration"] > 0 # 使用默认值
|
||||
|
||||
def test_invalid_effect(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-001/transition",
|
||||
json={"effect": "invalid_effect"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "无效的转场效果" in resp.json()["detail"]
|
||||
|
||||
def test_clip_not_found(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-nonexist/transition",
|
||||
json={"effect": "transition_fade"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_negative_duration_422(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-001/transition",
|
||||
json={"effect": "transition_fade", "duration": -0.5},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_duration_clamped_to_max(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
# 传一个超过最大值的时长，应该被钳制
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/clips/clip-001/transition",
|
||||
json={"effect": "transition_fade", "duration": 10.0},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
# fade 最大 2.0s
|
||||
assert data["duration"] <= 2.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /{plan_id}/transitions/batch 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBatchUpdateTransitions:
|
||||
def test_batch_all(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_fade", "duration": 0.5, "apply_to": "all"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["updated_count"] == 4 # 4个片段
|
||||
|
||||
for cid in ["clip-001", "clip-002", "clip-003", "clip-004"]:
|
||||
clip = clip_repo.get(cid)
|
||||
assert clip.transition_effect == "fade"
|
||||
assert clip.transition_duration == 0.5
|
||||
|
||||
def test_batch_except_first(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_fade", "apply_to": "except_first"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["updated_count"] == 3
|
||||
|
||||
# 第一个不变
|
||||
assert clip_repo.get("clip-001").transition_effect == "cut"
|
||||
# 其余三个被更新
|
||||
for cid in ["clip-002", "clip-003", "clip-004"]:
|
||||
assert clip_repo.get(cid).transition_effect == "fade"
|
||||
|
||||
def test_batch_except_last(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_slideleft", "apply_to": "except_last"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["updated_count"] == 3
|
||||
|
||||
# 最后一个不变
|
||||
assert clip_repo.get("clip-004").transition_effect == "cut"
|
||||
# 前三个被更新
|
||||
for cid in ["clip-001", "clip-002", "clip-003"]:
|
||||
assert clip_repo.get(cid).transition_effect == "slideleft"
|
||||
|
||||
def test_batch_middle(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_dissolve", "apply_to": "middle"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["updated_count"] == 2 # 4个片段，中间2个
|
||||
|
||||
# 首尾不变
|
||||
assert clip_repo.get("clip-001").transition_effect == "cut"
|
||||
assert clip_repo.get("clip-004").transition_effect == "cut"
|
||||
# 中间被更新
|
||||
assert clip_repo.get("clip-002").transition_effect == "dissolve"
|
||||
assert clip_repo.get("clip-003").transition_effect == "dissolve"
|
||||
|
||||
def test_batch_invalid_effect(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "invalid"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_batch_plan_not_found(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-nonexist/transitions/batch",
|
||||
json={"effect": "transition_fade"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_batch_invalid_apply_to(self, transition_client):
|
||||
c, _, _ = transition_client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_fade", "apply_to": "invalid"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_batch_none_transition(self, transition_client):
|
||||
c, _, clip_repo = transition_client
|
||||
# 先设一个转场
|
||||
c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_fade", "apply_to": "all"},
|
||||
)
|
||||
# 再全部设为无
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans/plan-001/transitions/batch",
|
||||
json={"effect": "transition_none", "apply_to": "all"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["updated_count"] == 4
|
||||
|
||||
for cid in ["clip-001", "clip-002", "clip-003", "clip-004"]:
|
||||
clip = clip_repo.get(cid)
|
||||
assert clip.transition_effect == "cut"
|
||||
assert clip.transition_duration == 0.0
|
||||
Reference in New Issue
Block a user