diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index db1d86cc2..0d1ed6f96 100755 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -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 ─────────────────────────────────────────────────────────────── diff --git a/apps/api/app/api/routes/edit_plans_filter.py b/apps/api/app/api/routes/edit_plans_filter.py new file mode 100755 index 000000000..5f9356a3f --- /dev/null +++ b/apps/api/app/api/routes/edit_plans_filter.py @@ -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) diff --git a/packages/domain/config_schemas.py b/packages/domain/config_schemas.py index cee9f9433..a717f1c18 100755 --- a/packages/domain/config_schemas.py +++ b/packages/domain/config_schemas.py @@ -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", } diff --git a/packages/domain/filter_presets.py b/packages/domain/filter_presets.py new file mode 100755 index 000000000..3a347dacd --- /dev/null +++ b/packages/domain/filter_presets.py @@ -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)}" diff --git a/tests/unit/test_edit_plan_filter_api.py b/tests/unit/test_edit_plan_filter_api.py new file mode 100644 index 000000000..9eadc59c4 --- /dev/null +++ b/tests/unit/test_edit_plan_filter_api.py @@ -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