feat: 滤镜调色 - 预设库 + 全局滤镜配置 + FFmpeg参数生成 #397

Merged
xiaoxia merged 1 commits from feat/filter-preset into develop 2026-07-17 09:18:24 +08:00
5 changed files with 1061 additions and 0 deletions
+2
View File
@@ -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 ───────────────────────────────────────────────────────────────
+198
View File
@@ -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)
+28
View File
@@ -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",
}
+357
View File
@@ -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.01.0为原值
saturation: float = 1.0 # 0.0 ~ 3.01.0为原值
gamma: float = 1.0 # 0.1 ~ 10.01.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-1000=原图,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)}"
+476
View File
@@ -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