xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 11 Pull Requests 2 Actions 1 Packages Projects Releases 7 Wiki Activity

feat: 导出设置 - 分辨率/帧率/码率/格式/水印 + 5套预设 #402

Merged
xiaoxia merged 5 commits from feat/export-settings into develop 2026-07-17 00:56:52 +08:00
Conversation 1 Commits 5 Files Changed 4
+668 -1
4 changed files with 668 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/api/routes/edit_plans.py
+13 -1
View File
@@ -67,7 +67,9 @@ class EditPlanUpdateRequest(BaseModel):
class CopyPlanRequest(BaseModel):
"""复制剪辑计划请求体"""
name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="新计划名称，不传则为「原名 - 副本」")
name: Optional[str] = Field(
default=None, min_length=1, max_length=200, description="新计划名称，不传则为「原名 - 副本」"
)
project_id: Optional[str] = Field(default=None, description="目标项目 ID，不传则复用源计划的项目")
@@ -258,6 +260,14 @@ def _to_response(p: EditPlan) -> EditPlanResponse:
)
# ── Include sub-routers (拆分模块) ────────────────────────────────────────────
# 注意：含静态路径的子路由需放在 CRUD 路由之前，避免被 /{plan_id} 抢先匹配
from .edit_plans_export import router as export_router
router.include_router(export_router)
# ── CRUD Routes ───────────────────────────────────────────────────────────────
@@ -506,6 +516,8 @@ def copy_plan(
current_user.user.id,
)
return _to_response(new_plan)
# ── 保存为模板 ────────────────────────────────────────────────────────────────
apps/api/app/api/routes/edit_plans_export.py
Executable
+274
View File
@@ -0,0 +1,274 @@
"""导出设置 API.
- GET /{plan_id}/export 获取导出配置
- PUT /{plan_id}/export 更新导出配置
- GET /export-presets 导出预设列表
"""
from __future__ import annotations
import logging
import re
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, status
from pydantic import BaseModel, Field, validator
from sqlalchemy.orm import Session
from packages.domain.config_schemas import normalize_plan_config
from ._helpers import check_project_access
logger = logging.getLogger(__name__)
router = APIRouter()
# ── 导出预设 ──────────────────────────────────────────────────────────────────
EXPORT_PRESETS = [
{
"id": "export_1080p_30",
"name": "1080P 高清",
"resolution": "1080x1920",
"fps": 30,
"video_bitrate": 8000,
"audio_bitrate": 128,
"format": "mp4",
"quality_preset": "balanced",
"description": "竖屏高清，适合短视频平台",
"size_hint": "约 10MB/分钟",
},
{
"id": "export_1080p_60",
"name": "1080P 高帧率",
"resolution": "1080x1920",
"fps": 60,
"video_bitrate": 12000,
"audio_bitrate": 128,
"format": "mp4",
"quality_preset": "high",
"description": "60帧高帧率，流畅运动画面",
"size_hint": "约 18MB/分钟",
},
{
"id": "export_720p_30",
"name": "720P 流畅",
"resolution": "720x1280",
"fps": 30,
"video_bitrate": 4000,
"audio_bitrate": 128,
"format": "mp4",
"quality_preset": "fast",
"description": "快速导出，文件较小",
"size_hint": "约 5MB/分钟",
},
{
"id": "export_4k_30",
"name": "4K 超清",
"resolution": "2160x3840",
"fps": 30,
"video_bitrate": 20000,
"audio_bitrate": 192,
"format": "mp4",
"quality_preset": "best",
"description": "4K超清画质，专业品质",
"size_hint": "约 30MB/分钟",
},
{
"id": "export_1080p_30_mov",
"name": "1080P ProRes",
"resolution": "1080x1920",
"fps": 30,
"video_bitrate": 15000,
"audio_bitrate": 256,
"format": "mov",
"quality_preset": "high",
"description": "MOV格式，适合后期剪辑",
"size_hint": "约 25MB/分钟",
},
]
VALID_QUALITY_PRESETS = {"ultra_fast", "fast", "balanced", "high", "best"}
VALID_FORMATS = {"mp4", "mov"}
RESOLUTION_PATTERN = re.compile(r"^\d+x\d+$")
# ── Schemas ──────────────────────────────────────────────────────────────────
class ExportConfigResponse(BaseModel):
"""导出配置响应"""
resolution: str
fps: int
video_bitrate: int
audio_bitrate: int
format: str
quality_preset: str
watermark_enabled: bool
watermark_text: str
class ExportUpdateRequest(BaseModel):
"""更新导出配置请求"""
resolution: Optional[str] = None
fps: Optional[int] = Field(default=None, ge=15, le=60)
video_bitrate: Optional[int] = Field(default=None, ge=1000, le=20000)
audio_bitrate: Optional[int] = Field(default=None, ge=64, le=320)
format: Optional[str] = None
quality_preset: Optional[str] = None
watermark_enabled: Optional[bool] = None
watermark_text: Optional[str] = None
@validator("resolution")
def validate_resolution(cls, v):
if v is None:
return v
if not RESOLUTION_PATTERN.match(v):
raise ValueError("分辨率格式错误，应为 宽x高，如 1080x1920")
w, h = v.split("x")
if int(w) < 100 or int(h) < 100:
raise ValueError("分辨率数值过小")
if int(w) > 4096 or int(h) > 4096:
raise ValueError("分辨率数值过大，最大 4096x4096")
return v
@validator("format")
def validate_format(cls, v):
if v is None:
return v
if v not in VALID_FORMATS:
raise ValueError(f"无效格式: {v}，支持: {VALID_FORMATS}")
return v
@validator("quality_preset")
def validate_quality_preset(cls, v):
if v is None:
return v
if v not in VALID_QUALITY_PRESETS:
raise ValueError(f"无效质量预设: {v}，支持: {VALID_QUALITY_PRESETS}")
return v
class ExportPresetItem(BaseModel):
"""导出预设条目"""
id: str
name: str
resolution: str
fps: int
video_bitrate: int
audio_bitrate: int
format: str
quality_preset: str
description: str
size_hint: str
class ExportPresetListResponse(BaseModel):
"""导出预设列表响应"""
items: List[ExportPresetItem]
total: int
# ── Helpers ──────────────────────────────────────────────────────────────────
def _get_export_config(plan_config: dict) -> dict:
e = plan_config.get("export", {})
if not isinstance(e, dict):
e = {}
return {
"resolution": e.get("resolution", "1080x1920"),
"fps": e.get("fps", 30),
"video_bitrate": e.get("video_bitrate", 8000),
"audio_bitrate": e.get("audio_bitrate", 128),
"format": e.get("format", "mp4"),
"quality_preset": e.get("quality_preset", "balanced"),
"watermark_enabled": e.get("watermark_enabled", False),
"watermark_text": e.get("watermark_text", ""),
}
# ── Routes ───────────────────────────────────────────────────────────────────
@router.get("/export-presets", response_model=ExportPresetListResponse)
def list_export_presets(
current_user: AuthenticatedUser = Depends(get_current_user),
) -> ExportPresetListResponse:
"""获取导出预设列表"""
items = [ExportPresetItem(**p) for p in EXPORT_PRESETS]
return ExportPresetListResponse(items=items, total=len(items))
@router.get("/{plan_id}/export", response_model=ExportConfigResponse)
def get_export_config(
plan_id: str,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
project_repository: Any = Depends(get_project_repository),
) -> ExportConfigResponse:
"""获取导出配置"""
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_export_config(plan.config or {})
return ExportConfigResponse(**config)
@router.put("/{plan_id}/export", response_model=ExportConfigResponse)
def update_export_config(
plan_id: str,
body: ExportUpdateRequest,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
project_repository: Any = Depends(get_project_repository),
) -> ExportConfigResponse:
"""更新导出配置"""
svc = EditPlanService(db)
plan = svc.get_plan(plan_id)
if not plan:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"剪辑计划不存在: {plan_id}",
)
if plan.project_id:
check_project_access(plan.project_id, current_user.user.id, project_repository)
# 合并更新
current = _get_export_config(plan.config or {})
updates = body.model_dump(exclude_none=True)
new_export = {**current, **updates}
# 更新到 plan.config
current_config = dict(plan.config or {})
current_config["export"] = new_export
normalized = normalize_plan_config(current_config)
updated_plan = svc.update_plan_config(plan_id, {"export": normalized["export"]})
result = _get_export_config(updated_plan.config or {})
logger.info(
"更新导出配置: plan_id=%s resolution=%s fps=%d by user=%s",
plan_id,
result["resolution"],
result["fps"],
current_user.user.id,
)
return ExportConfigResponse(**result)
packages/domain/config_schemas.py
+31
View File
@@ -140,6 +140,25 @@ class BGMConfig(BaseModel):
sidechain_threshold: float = Field(default=-25.0, ge=-60.0, le=0.0, description="人声闪避触发阈值（dB）")
class ExportConfig(BaseModel):
"""导出配置
视频输出参数设置。
"""
resolution: str = Field(default="1080x1920", description="输出分辨率，如 1080x1920 / 720x1280 / 2160x3840")
fps: int = Field(default=30, ge=15, le=60, description="输出帧率 15~60")
video_bitrate: int = Field(default=8000, ge=1000, le=20000, description="视频码率（kbps）")
audio_bitrate: int = Field(default=128, ge=64, le=320, description="音频码率（kbps）")
format: str = Field(default="mp4", description="输出格式：mp4 / mov")
quality_preset: str = Field(
default="balanced",
description="质量预设：ultra_fast / fast / balanced / high / best",
)
watermark_enabled: bool = Field(default=False, description="是否启用水印")
watermark_text: str = Field(default="", description="水印文字")
# ── 完整 config 模型 ─────────────────────────────────────────────────────────
@@ -155,6 +174,7 @@ class EditPlanConfigSchema(BaseModel):
title: TitleConfig = Field(default_factory=TitleConfig, description="标题配置")
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig, description="字幕配置")
bgm: BGMConfig = Field(default_factory=BGMConfig, description="BGM 配置")
export: ExportConfig = Field(default_factory=ExportConfig, description="导出配置")
editing_mode: str = Field(default="one_take", description="剪辑模式")
@@ -169,6 +189,7 @@ class EditTemplateConfigSchema(BaseModel):
title: TitleConfig = Field(default_factory=TitleConfig, description="标题默认配置")
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig, description="字幕默认配置")
bgm: BGMConfig = Field(default_factory=BGMConfig, description="BGM 默认配置")
export: ExportConfig = Field(default_factory=ExportConfig, description="导出默认配置")
editing_mode: str = Field(default="one_take", description="剪辑模式")
transition_enabled: bool = Field(default=True, description="是否启用转场")
@@ -218,6 +239,16 @@ DEFAULT_EDIT_PLAN_CONFIG: dict = {
"sidechain_release": 0.5,
"sidechain_threshold": -25.0,
},
"export": {
"resolution": "1080x1920",
"fps": 30,
"video_bitrate": 8000,
"audio_bitrate": 128,
"format": "mp4",
"quality_preset": "balanced",
"watermark_enabled": False,
"watermark_text": "",
},
"editing_mode": "one_take",
}
tests/unit/test_edit_plan_export_api.py
+350
View File
@@ -0,0 +1,350 @@
"""
导出设置 API 单元测试
覆盖：
- GET /export-presets - 导出预设列表
- GET /{plan_id}/export - 获取导出配置
- PUT /{plan_id}/export - 更新导出配置
"""
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
# ---------------------------------------------------------------------------
# Stub Repository
# ---------------------------------------------------------------------------
class StubEditPlanRepository:
def __init__(self, plans: dict[str, EditPlan] | None = None):
self._plans = plans or {}
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 list_all(self, *, status=None, skip=0, limit=50):
return list(self._plans.values())[skip : skip + limit]
def list_by_template(self, template_id, *, status=None, skip=0, limit=50):
return [p for p in self._plans.values() if p.template_id == template_id][skip : skip + limit]
def delete(self, plan_id: str) -> bool:
return self._plans.pop(plan_id, None) is not None
def count(self, *, status=None, template_id=None):
return len(self._plans)
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"):
return EditPlan(
id=plan_id,
template_id="tpl-001",
name="测试计划",
status=EditPlanStatus.EDITING,
total_duration=30.0,
config=normalize_plan_config({}),
project_id="",
created_by_user_id="user-001",
created_at=datetime(2026, 7, 16, 10, 0, 0),
updated_at=datetime(2026, 7, 16, 10, 0, 0),
)
def _create_test_app():
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")
def _mock_auth():
mock = MagicMock()
mock.user.id = "user-001"
return 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()
import app.api.routes.edit_plans_export as export_module
app.dependency_overrides[export_module.get_current_user] = _mock_auth
app.dependency_overrides[export_module.get_db_session] = lambda: MagicMock()
app.dependency_overrides[export_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 export_client():
app, plan_repo, cleanup = _create_test_app()
yield TestClient(app), plan_repo
cleanup()
# ---------------------------------------------------------------------------
# Export Presets 测试
# ---------------------------------------------------------------------------
class TestExportPresets:
def test_list_all_presets(self, export_client):
c, _ = export_client
resp = c.get("/api/v1/edit-plans/export-presets")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 5
assert len(data["items"]) == data["total"]
first = data["items"][0]
assert "id" in first
assert "name" in first
assert "resolution" in first
assert "fps" in first
assert "video_bitrate" in first
assert "format" in first
assert "description" in first
assert "size_hint" in first
def test_preset_has_valid_resolution(self, export_client):
c, _ = export_client
resp = c.get("/api/v1/edit-plans/export-presets")
data = resp.json()
for item in data["items"]:
assert "x" in item["resolution"]
assert item["fps"] >= 15
assert item["fps"] <= 60
assert item["format"] in ("mp4", "mov")
# ---------------------------------------------------------------------------
# GET /{plan_id}/export 测试
# ---------------------------------------------------------------------------
class TestGetExportConfig:
def test_default_export_config(self, export_client):
c, _ = export_client
resp = c.get("/api/v1/edit-plans/plan-001/export")
assert resp.status_code == 200
data = resp.json()
assert data["resolution"] == "1080x1920"
assert data["fps"] == 30
assert data["video_bitrate"] == 8000
assert data["audio_bitrate"] == 128
assert data["format"] == "mp4"
assert data["quality_preset"] == "balanced"
assert data["watermark_enabled"] is False
assert data["watermark_text"] == ""
def test_export_not_found(self, export_client):
c, _ = export_client
resp = c.get("/api/v1/edit-plans/plan-nonexist/export")
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# PUT /{plan_id}/export 测试
# ---------------------------------------------------------------------------
class TestUpdateExportConfig:
def test_update_resolution_and_fps(self, export_client):
c, plan_repo = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"resolution": "720x1280", "fps": 60},
)
assert resp.status_code == 200
data = resp.json()
assert data["resolution"] == "720x1280"
assert data["fps"] == 60
plan = plan_repo.get("plan-001")
assert plan.config["export"]["resolution"] == "720x1280"
assert plan.config["export"]["fps"] == 60
def test_update_bitrate(self, export_client):
c, plan_repo = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"video_bitrate": 12000, "audio_bitrate": 192},
)
assert resp.status_code == 200
data = resp.json()
assert data["video_bitrate"] == 12000
assert data["audio_bitrate"] == 192
def test_update_format(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"format": "mov"},
)
assert resp.status_code == 200
assert resp.json()["format"] == "mov"
def test_invalid_format(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"format": "avi"},
)
assert resp.status_code == 422
def test_update_quality_preset(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"quality_preset": "best"},
)
assert resp.status_code == 200
assert resp.json()["quality_preset"] == "best"
def test_invalid_quality_preset(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"quality_preset": "ultimate"},
)
assert resp.status_code == 422
def test_update_watermark(self, export_client):
c, plan_repo = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"watermark_enabled": True, "watermark_text": "我的视频"},
)
assert resp.status_code == 200
data = resp.json()
assert data["watermark_enabled"] is True
assert data["watermark_text"] == "我的视频"
plan = plan_repo.get("plan-001")
assert plan.config["export"]["watermark_enabled"] is True
assert plan.config["export"]["watermark_text"] == "我的视频"
def test_invalid_resolution_format(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"resolution": "1080*1920"},
)
assert resp.status_code == 422
def test_resolution_too_large(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"resolution": "8000x8000"},
)
assert resp.status_code == 422
def test_fps_out_of_range(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-001/export",
json={"fps": 120},
)
assert resp.status_code == 422
def test_export_not_found(self, export_client):
c, _ = export_client
resp = c.put(
"/api/v1/edit-plans/plan-nonexist/export",
json={"fps": 30},
)
assert resp.status_code == 404
def test_partial_update_preserves_other_fields(self, export_client):
c, _ = export_client
# 先修改一个
c.put("/api/v1/edit-plans/plan-001/export", json={"resolution": "720x1280"})
# 再修改另一个
resp = c.put("/api/v1/edit-plans/plan-001/export", json={"fps": 60})
data = resp.json()
# 分辨率应该保持
assert data["resolution"] == "720x1280"
# fps 更新了
assert data["fps"] == 60
# 其他默认值不变
assert data["format"] == "mp4"
assert data["video_bitrate"] == 8000