feat: 滤镜调色 - 预设库 + 全局滤镜配置 + FFmpeg参数生成 #397
@@ -264,8 +264,10 @@ def _to_response(p: EditPlan) -> EditPlanResponse:
|
||||
# 注意:含静态路径的子路由需放在 CRUD 路由之前,避免被 /{plan_id} 抢先匹配
|
||||
|
||||
from .edit_plans_export import router as export_router
|
||||
from .edit_plans_filter import router as filter_router
|
||||
|
||||
router.include_router(export_router)
|
||||
router.include_router(filter_router)
|
||||
|
||||
|
||||
# ── CRUD Routes ───────────────────────────────────────────────────────────────
|
||||
|
||||
Executable
+198
@@ -0,0 +1,198 @@
|
||||
"""滤镜调色 API.
|
||||
|
||||
- GET /filter-presets 滤镜预设列表
|
||||
- GET /{plan_id}/filter 获取全局滤镜配置
|
||||
- PUT /{plan_id}/filter 更新全局滤镜配置
|
||||
"""
|
||||
|
||||
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.config_schemas import normalize_plan_config
|
||||
from packages.domain.filter_presets import (
|
||||
FilterPreset,
|
||||
build_ffmpeg_filter,
|
||||
get_filter_preset,
|
||||
list_filter_presets,
|
||||
)
|
||||
|
||||
from ._helpers import check_project_access
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# ── Schemas ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class FilterPresetResponse(BaseModel):
|
||||
"""滤镜预设响应"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
category: str
|
||||
description: str
|
||||
tags: List[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class FilterConfigResponse(BaseModel):
|
||||
"""滤镜配置响应"""
|
||||
|
||||
enabled: bool
|
||||
preset_id: str
|
||||
intensity: int
|
||||
brightness: float
|
||||
contrast: float
|
||||
saturation: float
|
||||
warmth: float
|
||||
|
||||
|
||||
class FilterUpdateRequest(BaseModel):
|
||||
"""更新滤镜配置请求"""
|
||||
|
||||
enabled: Optional[bool] = None
|
||||
preset_id: Optional[str] = None
|
||||
intensity: Optional[int] = Field(default=None, ge=0, le=100)
|
||||
brightness: Optional[float] = Field(default=None, ge=-1.0, le=1.0)
|
||||
contrast: Optional[float] = Field(default=None, ge=0.0, le=2.0)
|
||||
saturation: Optional[float] = Field(default=None, ge=0.0, le=3.0)
|
||||
warmth: Optional[float] = Field(default=None, ge=-1.0, le=1.0)
|
||||
|
||||
|
||||
class FilterPresetListResponse(BaseModel):
|
||||
"""滤镜预设列表响应"""
|
||||
|
||||
items: List[FilterPresetResponse]
|
||||
total: int
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _preset_to_response(p: FilterPreset) -> FilterPresetResponse:
|
||||
return FilterPresetResponse(
|
||||
id=p.id,
|
||||
name=p.name,
|
||||
category=p.category,
|
||||
description=p.description,
|
||||
tags=list(p.tags),
|
||||
)
|
||||
|
||||
|
||||
def _get_filter_config(plan_config: dict) -> dict:
|
||||
"""从 plan.config 中提取滤镜配置"""
|
||||
f = plan_config.get("filter", {})
|
||||
if not isinstance(f, dict):
|
||||
f = {}
|
||||
return {
|
||||
"enabled": f.get("enabled", False),
|
||||
"preset_id": f.get("preset_id", "filter_none"),
|
||||
"intensity": f.get("intensity", 100),
|
||||
"brightness": f.get("brightness", 0.0),
|
||||
"contrast": f.get("contrast", 1.0),
|
||||
"saturation": f.get("saturation", 1.0),
|
||||
"warmth": f.get("warmth", 0.0),
|
||||
}
|
||||
|
||||
|
||||
# ── Routes ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("/filter-presets", response_model=FilterPresetListResponse)
|
||||
def list_presets(
|
||||
category: Optional[str] = Query(default=None, description="按分类筛选"),
|
||||
keyword: Optional[str] = Query(default=None, description="关键词搜索"),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> FilterPresetListResponse:
|
||||
"""获取滤镜预设列表"""
|
||||
presets = list_filter_presets(category=category, keyword=keyword)
|
||||
items = [_preset_to_response(p) for p in presets]
|
||||
return FilterPresetListResponse(items=items, total=len(items))
|
||||
|
||||
|
||||
@router.get("/{plan_id}/filter", response_model=FilterConfigResponse)
|
||||
def get_filter(
|
||||
plan_id: str,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
project_repository: Any = Depends(get_project_repository),
|
||||
) -> FilterConfigResponse:
|
||||
"""获取剪辑计划的全局滤镜配置"""
|
||||
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)
|
||||
|
||||
config = _get_filter_config(plan.config or {})
|
||||
return FilterConfigResponse(**config)
|
||||
|
||||
|
||||
@router.put("/{plan_id}/filter", response_model=FilterConfigResponse)
|
||||
def update_filter(
|
||||
plan_id: str,
|
||||
body: FilterUpdateRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
project_repository: Any = Depends(get_project_repository),
|
||||
) -> FilterConfigResponse:
|
||||
"""更新全局滤镜配置"""
|
||||
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)
|
||||
|
||||
# 验证 preset_id
|
||||
updates = body.model_dump(exclude_none=True)
|
||||
if "preset_id" in updates:
|
||||
preset = get_filter_preset(updates["preset_id"])
|
||||
if preset is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"无效的滤镜预设: {updates['preset_id']}",
|
||||
)
|
||||
|
||||
# 合并更新
|
||||
current = _get_filter_config(plan.config or {})
|
||||
new_filter = {**current, **updates}
|
||||
|
||||
# 如果设为原图 preset,自动关闭
|
||||
if new_filter["preset_id"] == "filter_none":
|
||||
new_filter["enabled"] = False
|
||||
|
||||
# 更新到 plan.config
|
||||
current_config = dict(plan.config or {})
|
||||
current_config["filter"] = new_filter
|
||||
normalized = normalize_plan_config(current_config)
|
||||
updated_plan = svc.update_plan_config(plan_id, {"filter": normalized["filter"]})
|
||||
|
||||
result = _get_filter_config(updated_plan.config or {})
|
||||
logger.info(
|
||||
"更新滤镜配置: plan_id=%s preset=%s intensity=%d by user=%s",
|
||||
plan_id,
|
||||
result["preset_id"],
|
||||
result["intensity"],
|
||||
current_user.user.id,
|
||||
)
|
||||
return FilterConfigResponse(**result)
|
||||
@@ -159,6 +159,23 @@ class ExportConfig(BaseModel):
|
||||
watermark_text: str = Field(default="", description="水印文字")
|
||||
|
||||
|
||||
class FilterConfig(BaseModel):
|
||||
"""滤镜调色配置
|
||||
|
||||
支持全局滤镜和按片段覆盖。
|
||||
强度 0-100,0 表示不应用,100 表示全量应用预设。
|
||||
"""
|
||||
|
||||
enabled: bool = Field(default=False, description="是否启用滤镜")
|
||||
preset_id: str = Field(default="filter_none", description="滤镜预设 ID")
|
||||
intensity: int = Field(default=100, ge=0, le=100, description="滤镜强度 0-100")
|
||||
# 自定义微调参数(在预设基础上叠加调整)
|
||||
brightness: float = Field(default=0.0, ge=-1.0, le=1.0, description="亮度微调")
|
||||
contrast: float = Field(default=1.0, ge=0.0, le=2.0, description="对比度微调(倍率)")
|
||||
saturation: float = Field(default=1.0, ge=0.0, le=3.0, description="饱和度微调(倍率)")
|
||||
warmth: float = Field(default=0.0, ge=-1.0, le=1.0, description="色温微调(正=暖,负=冷)")
|
||||
|
||||
|
||||
# ── 完整 config 模型 ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -175,6 +192,7 @@ class EditPlanConfigSchema(BaseModel):
|
||||
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig, description="字幕配置")
|
||||
bgm: BGMConfig = Field(default_factory=BGMConfig, description="BGM 配置")
|
||||
export: ExportConfig = Field(default_factory=ExportConfig, description="导出配置")
|
||||
filter: FilterConfig = Field(default_factory=FilterConfig, description="滤镜调色配置")
|
||||
editing_mode: str = Field(default="one_take", description="剪辑模式")
|
||||
|
||||
|
||||
@@ -190,6 +208,7 @@ class EditTemplateConfigSchema(BaseModel):
|
||||
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig, description="字幕默认配置")
|
||||
bgm: BGMConfig = Field(default_factory=BGMConfig, description="BGM 默认配置")
|
||||
export: ExportConfig = Field(default_factory=ExportConfig, description="导出默认配置")
|
||||
filter: FilterConfig = Field(default_factory=FilterConfig, description="滤镜默认配置")
|
||||
editing_mode: str = Field(default="one_take", description="剪辑模式")
|
||||
transition_enabled: bool = Field(default=True, description="是否启用转场")
|
||||
|
||||
@@ -249,6 +268,15 @@ DEFAULT_EDIT_PLAN_CONFIG: dict = {
|
||||
"watermark_enabled": False,
|
||||
"watermark_text": "",
|
||||
},
|
||||
"filter": {
|
||||
"enabled": False,
|
||||
"preset_id": "filter_none",
|
||||
"intensity": 100,
|
||||
"brightness": 0.0,
|
||||
"contrast": 1.0,
|
||||
"saturation": 1.0,
|
||||
"warmth": 0.0,
|
||||
},
|
||||
"editing_mode": "one_take",
|
||||
}
|
||||
|
||||
|
||||
Executable
+357
@@ -0,0 +1,357 @@
|
||||
"""滤镜预设库 — 视频调色滤镜预设清单.
|
||||
|
||||
每个滤镜预设对应一组 FFmpeg 滤镜参数,用于视频调色。
|
||||
所有参数均可调整强度(0-100),0表示原图,100表示全量应用。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FilterPreset:
|
||||
"""滤镜预设条目"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
category: str # 分类:basic / cinematic / vintage / bw / style
|
||||
description: str = ""
|
||||
tags: List[str] = field(default_factory=list)
|
||||
# FFmpeg eq 滤镜参数(基准值,实际应用时乘以强度系数)
|
||||
brightness: float = 0.0 # -1.0 ~ 1.0
|
||||
contrast: float = 1.0 # 0.0 ~ 2.0,1.0为原值
|
||||
saturation: float = 1.0 # 0.0 ~ 3.0,1.0为原值
|
||||
gamma: float = 1.0 # 0.1 ~ 10.0,1.0为原值
|
||||
gamma_r: float = 1.0 # 红通道伽马
|
||||
gamma_g: float = 1.0 # 绿通道伽马
|
||||
gamma_b: float = 1.0 # 蓝通道伽马
|
||||
hue: float = 0.0 # 色相偏移 -180 ~ 180度
|
||||
# 可选的颜色查找表 LUT(后续扩展)
|
||||
lut_url: str = ""
|
||||
|
||||
|
||||
# ── 预设库清单 ────────────────────────────────────────────────────────────────
|
||||
|
||||
FILTER_PRESET_LIBRARY: List[FilterPreset] = [
|
||||
# ── 基础 basic ─────────────────────────────────────────────────────
|
||||
FilterPreset(
|
||||
id="filter_none",
|
||||
name="原图",
|
||||
category="basic",
|
||||
description="不应用任何滤镜,保持原始画面",
|
||||
tags=["原图", "无"],
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_brighten",
|
||||
name="明亮",
|
||||
category="basic",
|
||||
description="提升画面亮度,适合偏暗的素材",
|
||||
tags=["提亮", "基础"],
|
||||
brightness=0.12,
|
||||
contrast=1.05,
|
||||
saturation=1.05,
|
||||
gamma=1.1,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_warm",
|
||||
name="暖色",
|
||||
category="basic",
|
||||
description="暖色调,增加温暖感",
|
||||
tags=["暖色", "温馨"],
|
||||
gamma_r=1.1,
|
||||
gamma_g=1.02,
|
||||
gamma_b=0.9,
|
||||
saturation=1.05,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_cool",
|
||||
name="冷色",
|
||||
category="basic",
|
||||
description="冷色调,清凉干净",
|
||||
tags=["冷色", "清新"],
|
||||
gamma_r=0.9,
|
||||
gamma_g=1.0,
|
||||
gamma_b=1.1,
|
||||
saturation=1.02,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_contrast",
|
||||
name="高对比",
|
||||
category="basic",
|
||||
description="增强对比度,画面更通透",
|
||||
tags=["对比", "通透"],
|
||||
contrast=1.25,
|
||||
saturation=1.1,
|
||||
gamma=0.95,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_saturate",
|
||||
name="鲜艳",
|
||||
category="basic",
|
||||
description="提升饱和度,色彩更浓郁",
|
||||
tags=["鲜艳", "浓郁"],
|
||||
saturation=1.4,
|
||||
contrast=1.05,
|
||||
),
|
||||
# ── 电影感 cinematic ─────────────────────────────────────────────
|
||||
FilterPreset(
|
||||
id="filter_cinematic",
|
||||
name="电影感",
|
||||
category="cinematic",
|
||||
description="经典电影色调,青橙对比",
|
||||
tags=["电影", "青橙", "质感"],
|
||||
contrast=1.2,
|
||||
saturation=0.9,
|
||||
gamma_r=1.15,
|
||||
gamma_g=0.95,
|
||||
gamma_b=0.85,
|
||||
brightness=-0.03,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_teal_orange",
|
||||
name="青橙色调",
|
||||
category="cinematic",
|
||||
description="好莱坞经典青橙对比色",
|
||||
tags=["青橙", "好莱坞", "对比"],
|
||||
contrast=1.15,
|
||||
saturation=1.1,
|
||||
gamma_r=1.2,
|
||||
gamma_g=0.9,
|
||||
gamma_b=0.8,
|
||||
),
|
||||
# ── 复古 vintage ─────────────────────────────────────────────────
|
||||
FilterPreset(
|
||||
id="filter_vintage",
|
||||
name="复古",
|
||||
category="vintage",
|
||||
description="复古胶片色调,怀旧感",
|
||||
tags=["复古", "怀旧", "胶片"],
|
||||
saturation=0.8,
|
||||
contrast=0.9,
|
||||
gamma_r=1.1,
|
||||
gamma_g=1.0,
|
||||
gamma_b=0.85,
|
||||
brightness=-0.02,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_retro",
|
||||
name="怀旧",
|
||||
category="vintage",
|
||||
description="80年代复古感",
|
||||
tags=["怀旧", "80年代"],
|
||||
saturation=0.75,
|
||||
contrast=0.95,
|
||||
gamma_r=1.2,
|
||||
gamma_g=1.05,
|
||||
gamma_b=0.9,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_sepia",
|
||||
name="棕褐色",
|
||||
category="vintage",
|
||||
description="老照片棕褐色调",
|
||||
tags=["棕褐", "老照片", "复古"],
|
||||
saturation=0.3,
|
||||
gamma_r=1.3,
|
||||
gamma_g=1.1,
|
||||
gamma_b=0.8,
|
||||
contrast=0.95,
|
||||
),
|
||||
# ── 黑白 bw ──────────────────────────────────────────────────────
|
||||
FilterPreset(
|
||||
id="filter_bw",
|
||||
name="黑白",
|
||||
category="bw",
|
||||
description="经典黑白",
|
||||
tags=["黑白", "经典"],
|
||||
saturation=0.0,
|
||||
contrast=1.1,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_bw_high",
|
||||
name="高对比黑白",
|
||||
category="bw",
|
||||
description="高对比度黑白,戏剧感强",
|
||||
tags=["黑白", "高对比", "戏剧"],
|
||||
saturation=0.0,
|
||||
contrast=1.4,
|
||||
gamma=0.9,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_bw_soft",
|
||||
name="柔和黑白",
|
||||
category="bw",
|
||||
description="柔和灰度过渡,细腻质感",
|
||||
tags=["黑白", "柔和", "细腻"],
|
||||
saturation=0.0,
|
||||
contrast=0.9,
|
||||
gamma=1.1,
|
||||
),
|
||||
# ── 风格化 style ────────────────────────────────────────────────
|
||||
FilterPreset(
|
||||
id="filter_japanese",
|
||||
name="日系",
|
||||
category="style",
|
||||
description="日系清新,低对比高明度",
|
||||
tags=["日系", "清新", "干净"],
|
||||
contrast=0.85,
|
||||
brightness=0.08,
|
||||
saturation=0.85,
|
||||
gamma_r=0.98,
|
||||
gamma_g=1.02,
|
||||
gamma_b=1.08,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_hk",
|
||||
name="港风",
|
||||
category="style",
|
||||
description="90年代港风,暖黄+高饱和",
|
||||
tags=["港风", "复古", "浓郁"],
|
||||
saturation=1.25,
|
||||
contrast=1.1,
|
||||
gamma_r=1.2,
|
||||
gamma_g=1.05,
|
||||
gamma_b=0.85,
|
||||
brightness=-0.02,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_cyberpunk",
|
||||
name="赛博朋克",
|
||||
category="style",
|
||||
description="赛博朋克风,青紫霓虹",
|
||||
tags=["赛博", "霓虹", "未来感"],
|
||||
contrast=1.2,
|
||||
saturation=1.3,
|
||||
gamma_r=1.3,
|
||||
gamma_g=0.7,
|
||||
gamma_b=1.2,
|
||||
brightness=-0.05,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_fresh",
|
||||
name="清新",
|
||||
category="style",
|
||||
description="清新自然,通透干净",
|
||||
tags=["清新", "自然", "通透"],
|
||||
brightness=0.05,
|
||||
saturation=1.05,
|
||||
contrast=1.05,
|
||||
gamma_g=1.03,
|
||||
gamma_b=1.05,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_dramatic",
|
||||
name="戏剧感",
|
||||
category="style",
|
||||
description="强对比暗角,戏剧化氛围",
|
||||
tags=["戏剧", "暗角", "氛围"],
|
||||
contrast=1.35,
|
||||
saturation=0.9,
|
||||
brightness=-0.08,
|
||||
gamma=0.85,
|
||||
),
|
||||
FilterPreset(
|
||||
id="filter_dreamy",
|
||||
name="梦幻",
|
||||
category="style",
|
||||
description="柔光梦幻感,低对比",
|
||||
tags=["梦幻", "柔光", "唯美"],
|
||||
contrast=0.8,
|
||||
brightness=0.1,
|
||||
saturation=1.1,
|
||||
gamma=1.15,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
# ── 工具函数 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_filter_preset(preset_id: str) -> Optional[FilterPreset]:
|
||||
"""根据 ID 获取滤镜预设"""
|
||||
for p in FILTER_PRESET_LIBRARY:
|
||||
if p.id == preset_id:
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def list_filter_presets(
|
||||
*,
|
||||
category: Optional[str] = None,
|
||||
keyword: Optional[str] = None,
|
||||
) -> List[FilterPreset]:
|
||||
"""筛选滤镜预设列表
|
||||
|
||||
Args:
|
||||
category: 按分类筛选
|
||||
keyword: 关键词搜索(名称/标签/描述)
|
||||
|
||||
Returns:
|
||||
筛选后的预设列表
|
||||
"""
|
||||
results = FILTER_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 build_ffmpeg_filter(preset_id: str, intensity: int = 100) -> str:
|
||||
"""根据预设和强度生成 FFmpeg eq 滤镜字符串.
|
||||
|
||||
Args:
|
||||
preset_id: 滤镜预设 ID
|
||||
intensity: 强度 0-100,0=原图,100=全量
|
||||
|
||||
Returns:
|
||||
FFmpeg eq 滤镜参数字符串
|
||||
"""
|
||||
preset = get_filter_preset(preset_id)
|
||||
if preset is None or intensity <= 0:
|
||||
return ""
|
||||
|
||||
if intensity >= 100:
|
||||
intensity = 100
|
||||
|
||||
factor = intensity / 100.0
|
||||
|
||||
# 计算插值后的参数(向原值插值)
|
||||
brightness = preset.brightness * factor
|
||||
contrast = 1.0 + (preset.contrast - 1.0) * factor
|
||||
saturation = 1.0 + (preset.saturation - 1.0) * factor
|
||||
gamma = 1.0 + (preset.gamma - 1.0) * factor
|
||||
gamma_r = 1.0 + (preset.gamma_r - 1.0) * factor
|
||||
gamma_g = 1.0 + (preset.gamma_g - 1.0) * factor
|
||||
gamma_b = 1.0 + (preset.gamma_b - 1.0) * factor
|
||||
|
||||
parts = []
|
||||
if abs(brightness) > 0.001:
|
||||
parts.append(f"brightness={brightness:.3f}")
|
||||
if abs(contrast - 1.0) > 0.001:
|
||||
parts.append(f"contrast={contrast:.3f}")
|
||||
if abs(saturation - 1.0) > 0.001:
|
||||
parts.append(f"saturation={saturation:.3f}")
|
||||
if abs(gamma - 1.0) > 0.001:
|
||||
parts.append(f"gamma={gamma:.3f}")
|
||||
if abs(gamma_r - 1.0) > 0.001:
|
||||
parts.append(f"gamma_r={gamma_r:.3f}")
|
||||
if abs(gamma_g - 1.0) > 0.001:
|
||||
parts.append(f"gamma_g={gamma_g:.3f}")
|
||||
if abs(gamma_b - 1.0) > 0.001:
|
||||
parts.append(f"gamma_b={gamma_b:.3f}")
|
||||
|
||||
if not parts:
|
||||
return ""
|
||||
|
||||
return f"eq={':'.join(parts)}"
|
||||
@@ -0,0 +1,476 @@
|
||||
"""
|
||||
滤镜调色 API 单元测试
|
||||
|
||||
覆盖:
|
||||
- GET /filter-presets - 滤镜预设列表
|
||||
- GET /{plan_id}/filter - 获取滤镜配置
|
||||
- PUT /{plan_id}/filter - 更新滤镜配置
|
||||
"""
|
||||
|
||||
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.filter_presets import FILTER_PRESET_LIBRARY, build_ffmpeg_filter
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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 list_by_plan(self, plan_id, *, status=None, skip=0, limit=100):
|
||||
return []
|
||||
|
||||
def count(self, plan_id, *, status=None):
|
||||
return 0
|
||||
|
||||
def get(self, clip_id: str):
|
||||
return None
|
||||
|
||||
def create(self, clip):
|
||||
return clip
|
||||
|
||||
def update(self, clip):
|
||||
return clip
|
||||
|
||||
def delete(self, clip_id: str) -> bool:
|
||||
return False
|
||||
|
||||
def delete_by_plan(self, plan_id: str) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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 _create_test_app():
|
||||
import app.api.routes.edit_plans_filter as filter_module
|
||||
import app.services.edit_plan_service as service_module
|
||||
from app.api.routes.edit_plans import router
|
||||
|
||||
plan = _make_sample_plan()
|
||||
stub_plan_repo = StubEditPlanRepository({plan.id: plan})
|
||||
stub_clip_repo = StubEditPlanClipRepository()
|
||||
|
||||
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[filter_module.get_current_user] = _mock_auth
|
||||
app.dependency_overrides[filter_module.get_db_session] = lambda: MagicMock()
|
||||
app.dependency_overrides[filter_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, cleanup
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def filter_client():
|
||||
app, plan_repo, cleanup = _create_test_app()
|
||||
yield TestClient(app), plan_repo
|
||||
cleanup()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Filter Presets 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFilterPresets:
|
||||
def test_list_all_presets(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-presets")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == len(FILTER_PRESET_LIBRARY)
|
||||
assert data["total"] > 10
|
||||
assert len(data["items"]) == data["total"]
|
||||
# 验证字段
|
||||
first = data["items"][0]
|
||||
assert "id" in first
|
||||
assert "name" in first
|
||||
assert "category" in first
|
||||
assert "description" in first
|
||||
assert "tags" in first
|
||||
|
||||
def test_filter_by_category_basic(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-presets?category=basic")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] > 0
|
||||
for item in data["items"]:
|
||||
assert item["category"] == "basic"
|
||||
|
||||
def test_filter_by_category_bw(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-presets?category=bw")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] >= 3
|
||||
for item in data["items"]:
|
||||
assert item["category"] == "bw"
|
||||
|
||||
def test_filter_by_keyword(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-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_by_keyword_japanese(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-presets?keyword=日系")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] >= 1
|
||||
assert data["items"][0]["name"] == "日系"
|
||||
|
||||
def test_filter_empty_result(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-presets?keyword=不存在的滤镜")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 0
|
||||
assert data["items"] == []
|
||||
|
||||
def test_filter_invalid_category(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/filter-presets?category=nonexistent")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /{plan_id}/filter 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetFilter:
|
||||
def test_get_default_filter(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/plan-001/filter")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["enabled"] is False
|
||||
assert data["preset_id"] == "filter_none"
|
||||
assert data["intensity"] == 100
|
||||
assert data["brightness"] == 0.0
|
||||
assert data["contrast"] == 1.0
|
||||
assert data["saturation"] == 1.0
|
||||
assert data["warmth"] == 0.0
|
||||
|
||||
def test_get_filter_not_found(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.get("/api/v1/edit-plans/plan-nonexist/filter")
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_get_filter_with_custom_config(self, filter_client):
|
||||
c, plan_repo = filter_client
|
||||
plan = plan_repo.get("plan-001")
|
||||
new_config = dict(plan.config)
|
||||
new_config["filter"] = {
|
||||
"enabled": True,
|
||||
"preset_id": "filter_cinematic",
|
||||
"intensity": 80,
|
||||
"brightness": 0.1,
|
||||
"contrast": 1.2,
|
||||
"saturation": 0.9,
|
||||
"warmth": 0.3,
|
||||
}
|
||||
plan.config = new_config
|
||||
plan_repo.update(plan)
|
||||
|
||||
resp = c.get("/api/v1/edit-plans/plan-001/filter")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["enabled"] is True
|
||||
assert data["preset_id"] == "filter_cinematic"
|
||||
assert data["intensity"] == 80
|
||||
assert data["brightness"] == 0.1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PUT /{plan_id}/filter 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUpdateFilter:
|
||||
def test_enable_filter(self, filter_client):
|
||||
c, plan_repo = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"enabled": True, "preset_id": "filter_cinematic"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["enabled"] is True
|
||||
assert data["preset_id"] == "filter_cinematic"
|
||||
|
||||
plan = plan_repo.get("plan-001")
|
||||
assert plan.config["filter"]["enabled"] is True
|
||||
assert plan.config["filter"]["preset_id"] == "filter_cinematic"
|
||||
|
||||
def test_adjust_intensity(self, filter_client):
|
||||
c, plan_repo = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"enabled": True, "preset_id": "filter_cinematic", "intensity": 50},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["intensity"] == 50
|
||||
|
||||
plan = plan_repo.get("plan-001")
|
||||
assert plan.config["filter"]["intensity"] == 50
|
||||
|
||||
def test_invalid_intensity_returns_422(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"intensity": 150},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_invalid_preset_returns_400(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"preset_id": "nonexistent_filter"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "无效的滤镜预设" in resp.json()["detail"]
|
||||
|
||||
def test_filter_not_found(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-nonexist/filter",
|
||||
json={"enabled": True},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_set_none_preset_disables_filter(self, filter_client):
|
||||
c, plan_repo = filter_client
|
||||
# 先启用一个滤镜
|
||||
c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"enabled": True, "preset_id": "filter_cinematic"},
|
||||
)
|
||||
|
||||
# 再设为原图
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"preset_id": "filter_none"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["preset_id"] == "filter_none"
|
||||
assert data["enabled"] is False # 原图自动关闭
|
||||
|
||||
def test_partial_update(self, filter_client):
|
||||
c, plan_repo = filter_client
|
||||
# 先设置完整配置
|
||||
c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={
|
||||
"enabled": True,
|
||||
"preset_id": "filter_warm",
|
||||
"intensity": 70,
|
||||
"brightness": 0.05,
|
||||
},
|
||||
)
|
||||
|
||||
# 只修改强度,其他保持不变
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"intensity": 90},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["intensity"] == 90
|
||||
assert data["preset_id"] == "filter_warm" # 保持不变
|
||||
assert data["enabled"] is True # 保持不变
|
||||
assert data["brightness"] == 0.05 # 保持不变
|
||||
|
||||
def test_custom_adjustments(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={
|
||||
"enabled": True,
|
||||
"preset_id": "filter_cinematic",
|
||||
"brightness": 0.1,
|
||||
"contrast": 1.3,
|
||||
"saturation": 1.2,
|
||||
"warmth": 0.2,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["brightness"] == 0.1
|
||||
assert data["contrast"] == 1.3
|
||||
assert data["saturation"] == 1.2
|
||||
assert data["warmth"] == 0.2
|
||||
|
||||
def test_invalid_brightness_returns_422(self, filter_client):
|
||||
c, _ = filter_client
|
||||
resp = c.put(
|
||||
"/api/v1/edit-plans/plan-001/filter",
|
||||
json={"brightness": 2.0},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FFmpeg 滤镜生成测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildFFmpegFilter:
|
||||
def test_no_filter(self):
|
||||
assert build_ffmpeg_filter("filter_none", 100) == ""
|
||||
|
||||
def test_zero_intensity(self):
|
||||
assert build_ffmpeg_filter("filter_cinematic", 0) == ""
|
||||
|
||||
def test_invalid_preset(self):
|
||||
assert build_ffmpeg_filter("nonexistent", 100) == ""
|
||||
|
||||
def test_cinematic_full(self):
|
||||
result = build_ffmpeg_filter("filter_cinematic", 100)
|
||||
assert result.startswith("eq=")
|
||||
assert "contrast=" in result
|
||||
assert "saturation=" in result
|
||||
assert "gamma_r=" in result
|
||||
|
||||
def test_cinematic_half(self):
|
||||
full = build_ffmpeg_filter("filter_cinematic", 100)
|
||||
half = build_ffmpeg_filter("filter_cinematic", 50)
|
||||
assert full != half
|
||||
# 50% 强度的参数应该更接近原值
|
||||
assert "eq=" in half
|
||||
|
||||
def test_bw_filter(self):
|
||||
result = build_ffmpeg_filter("filter_bw", 100)
|
||||
assert "saturation=0" in result
|
||||
|
||||
def test_warm_filter(self):
|
||||
result = build_ffmpeg_filter("filter_warm", 100)
|
||||
assert "gamma_r=" in result
|
||||
assert "gamma_b=" in result
|
||||
Reference in New Issue
Block a user