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

feat(phase2): 模板编辑器子路由迁移第一批（BGM/生成/时间线/转场/滤镜/导出）
AI Code Review / AI Code Review (pull_request) Successful in 2m3s
Details
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m5s
Details
CI/CD Pipeline / Unit Tests (pull_request) Failing after 5m12s
Details
PR Automation / Auto Approve on CI Green (pull_request) Successful in 6m39s
Details
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 6m39s
Details
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 1393h38m22s
Details
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1393h39m43s
Details
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1393h39m43s
Details
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1393h39m43s
Details
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1393h39m45s
Details
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1393h39m45s
Details
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1393h39m47s
Details
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1393h39m45s
Details
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1393h39m47s
Details
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1393h39m48s
Details
CI/CD Pipeline / Check if frontend-only change (pull_request) Has been skipped
Details
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Has been skipped
Details
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
Details
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
Details
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1394h11m51s
Details
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1394h11m49s
Details
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1394h11m53s
Details

Browse Source 
新增16个模板编辑器端点：
- BGM: GET/PUT /bgm + GET /bgm/presets（3个）
- 生成: POST /generate + GET /generation-status + GET /generations（3个，含4个自动兜底+队列限流）
- 时间线: GET /timeline（1个）
- 转场: GET /transition-presets + PUT /clips/{id}/transition + POST /transitions/batch（3个）
- 滤镜: GET /filter-presets + GET/PUT /filter（3个）
- 导出: GET /export-presets + GET/PUT /export（3个）

总计: 模板编辑器从7个端点 → 23个端点
全部复用 EditPlanService 业务逻辑，零代码重复
This commit is contained in:
CI Bot
2026-07-20 13:49:00 +08:00
parent 3bc7c3057a
commit 5d106319d0
1 changed files with 836 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/api/routes/templates_editor.py
+836 -1
View File
@@ -25,14 +25,61 @@ from __future__ import annotations
import logging
from typing import Any, List, Optional
from app.api.routes.edit_plans import (
BGMConfigUpdateRequest,
ClipStatusItem,
EditPlanGenerateResponse,
EditPlanGenerationsResponse,
EditPlanGenerationStatusResponse,
)
from app.api.routes.edit_plans_export import (
ExportConfigResponse,
ExportPresetListResponse,
ExportUpdateRequest,
)
from app.api.routes.edit_plans_filter import (
FilterConfigResponse,
FilterPresetListResponse,
FilterUpdateRequest,
)
from app.api.routes.edit_plans_transitions import (
BatchTransitionRequest,
BatchTransitionResponse,
ClipTransitionResponse,
TransitionPresetListResponse,
TransitionUpdateRequest,
)
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session
from app.core.celery_app import celery_app
from app.core.storage import OSSStorageService, get_storage_service
from app.core.task_enqueue import GLOBAL_PENDING_LIMIT, USER_PENDING_LIMIT
from app.dependencies import (
get_asset_library_repository,
get_asset_repository,
get_db_session,
)
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
SQLAlchemyGenerationTaskRepository,
)
from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
SQLAlchemyTemplateClipConfigRepository,
)
from packages.adapters.sqlalchemy_impl.template_repository import (
SQLAlchemyTemplateRepository,
)
from packages.application.generation_tasks import (
CreateGenerationTaskCommand,
CreateGenerationTaskUseCase,
)
from packages.domain.config_schemas import normalize_plan_config
from packages.domain.edit_plan import EditPlanStatus
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Template Editor"])
@@ -351,3 +398,791 @@ def delete_draft_clip(
if not success:
raise HTTPException(status_code=404, detail="片段不存在")
return None
# ── BGM 模块 ────────────────────────────────────────────────────────────────────
@router.get("/bgm", response_model=dict[str, Any])
def get_editor_bgm(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
):
"""获取草稿的 BGM 配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = plan.config or {}
return {
"plan_id": plan.id,
"bgm": config.get("bgm", {}),
}
@router.put("/bgm", response_model=dict[str, Any])
def update_editor_bgm(
template_id: str,
body: BGMConfigUpdateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
current_user: AuthenticatedUser = Depends(get_current_user),
):
"""更新草稿的 BGM 配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = dict(plan.config) if plan.config else {}
current_bgm = dict(config.get("bgm", {}))
update_data = body.model_dump(exclude_none=True)
current_bgm.update(update_data)
if current_bgm.get("enabled"):
has_source = any(current_bgm.get(key) for key in ("asset_id", "preset_id", "audio_url") if current_bgm.get(key))
if not has_source:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="启用 BGM 时需要指定素材来源（asset_id / preset_id / audio_url）",
)
config["bgm"] = current_bgm
updated_plan = plan_svc.update_plan_config(plan_id, config)
logger.info(
"模板编辑器更新BGM: template_id=%s plan_id=%s enabled=%s by user=%s",
template_id,
plan_id,
current_bgm.get("enabled", False),
current_user.user_id,
)
return {
"plan_id": updated_plan.id,
"bgm": current_bgm,
}
@router.get("/bgm/presets", response_model=dict[str, Any])
def list_editor_bgm_presets(
style: Optional[str] = Query(default=None, description="按风格筛选"),
keyword: Optional[str] = Query(default=None, description="关键词搜索"),
skip: int = Query(default=0, ge=0, description="分页偏移"),
limit: int = Query(default=50, ge=1, le=200, description="每页数量"),
_: AuthenticatedUser = Depends(get_current_user),
):
"""获取预设 BGM 列表"""
from packages.domain.preset_bgm import (
BGM_STYLES,
PRESET_BGM_LIBRARY,
list_preset_bgm_by_style,
search_preset_bgm,
)
bgm_list = PRESET_BGM_LIBRARY
if keyword:
bgm_list = search_preset_bgm(keyword)
elif style:
bgm_list = list_preset_bgm_by_style(style)
total = len(bgm_list)
paged = bgm_list[skip : skip + limit]
return {
"total": total,
"skip": skip,
"limit": limit,
"styles": BGM_STYLES,
"items": [
{
"id": bgm.id,
"name": bgm.name,
"style": bgm.style,
"style_label": BGM_STYLES.get(bgm.style, bgm.style),
"duration": bgm.duration,
"artist": bgm.artist,
"description": bgm.description,
"tags": bgm.tags,
"audio_url": bgm.audio_url,
}
for bgm in paged
],
}
# ── 生成模块 ────────────────────────────────────────────────────────────────────
def _auto_fallback_draft_to_editing(svc: EditPlanService, plan_id: str, plan_check) -> None:
"""自动兜底 1: draft → editing"""
if plan_check.status == EditPlanStatus.DRAFT:
logger.info("模板编辑器自动兜底: plan=%s draft→editing", plan_id)
svc.transition_status(plan_id, EditPlanStatus.EDITING)
def _auto_fallback_copy_template_clips(svc: EditPlanService, plan_id: str, plan_check, db: Session) -> None:
"""自动兜底 2: 无片段 + 有 template_id → 从模板复制片段配置"""
existing_clips = svc.count_clips(plan_id)
if existing_clips == 0 and plan_check.template_id:
logger.info(
"模板编辑器自动兜底: plan=%s 无片段，从模板 %s 复制片段配置",
plan_id,
plan_check.template_id,
)
clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db)
configs = clip_config_repo.list_by_template(plan_check.template_id)
if configs:
for cfg in configs:
svc.create_clip(
plan_id=plan_id,
clip_type=cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
order=cfg.order,
template_clip_config_id=cfg.id,
duration=cfg.default_duration,
transition_effect=(
cfg.transition_effect.value
if hasattr(cfg.transition_effect, "value")
else cfg.transition_effect
),
)
logger.info(
"模板编辑器自动兜底: plan=%s 从 template_clip_configs 复制了 %d 个片段",
plan_id,
len(configs),
)
else:
tpl_repo = SQLAlchemyTemplateRepository(db)
segments = tpl_repo.list_segments(plan_check.template_id)
for seg in segments:
avg_duration = (seg.duration_min + seg.duration_max) / 2
svc.create_clip(
plan_id=plan_id,
clip_type="main",
order=seg.segment_order,
duration=avg_duration,
config={
"material_type": seg.material_type or "",
"template_segment_id": seg.id,
},
)
logger.info(
"模板编辑器自动兜底: plan=%s 从旧模板 segments 复制了 %d 个片段",
plan_id,
len(segments),
)
def _auto_fallback_assign_assets(svc: EditPlanService, plan_id: str, plan_check) -> list:
"""自动兜底 3: 为没有素材的片段分配素材。返回剩余无素材片段列表。"""
all_clips = svc.list_clips(plan_id)
clips_without_asset = [c for c in all_clips if not c.asset_id]
config_asset_ids = (plan_check.config or {}).get("asset_ids", [])
if clips_without_asset and config_asset_ids:
logger.info(
"模板编辑器自动兜底3: plan=%s 为 %d 个无素材片段分配 %d 个指定素材",
plan_id,
len(clips_without_asset),
len(config_asset_ids),
)
for i, clip in enumerate(clips_without_asset):
asset_idx = i % len(config_asset_ids)
svc.assign_asset(clip.id, config_asset_ids[asset_idx])
logger.info("模板编辑器自动兜底3: plan=%s 素材分配完成", plan_id)
clips_without_asset = []
return clips_without_asset
def _auto_fallback_auto_material_mode(
svc: EditPlanService,
plan_id: str,
plan_check,
clips_without_asset: list,
asset_library_repo: Any,
asset_repo: Any,
) -> None:
"""自动兜底 4: 项目有视频素材库时自动选素材"""
if not clips_without_asset:
return
if not plan_check.project_id:
return
import random
logger.info(
"模板编辑器自动兜底4: plan=%s 自动选素材分配给 %d 个无素材片段",
plan_id,
len(clips_without_asset),
)
libs = asset_library_repo.find_by_project(plan_check.project_id)
video_lib = None
for lib in libs:
lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
if lib_kind == "video":
video_lib = lib
break
if video_lib:
assets = asset_repo.find_by_library(video_lib.id)
ready_videos = [
a
for a in assets
if (a.status.value if hasattr(a.status, "value") else a.status) == "ready"
and a.mime_type
and a.mime_type.startswith("video")
]
if ready_videos:
random.shuffle(ready_videos)
for i, clip in enumerate(clips_without_asset):
asset = ready_videos[i % len(ready_videos)]
svc.assign_asset(clip.id, asset.id)
logger.info(
"模板编辑器自动兜底4: plan=%s 从素材库 %s 分配了 %d 个素材",
plan_id,
video_lib.name,
len(ready_videos),
)
def _check_queue_limits(gen_task_repo, user_id: str) -> None:
"""队列限流预检查"""
try:
has_count = hasattr(gen_task_repo, "count_pending_by_user") and hasattr(gen_task_repo, "count_pending_total")
if has_count:
user_pending = gen_task_repo.count_pending_by_user(user_id)
global_pending = gen_task_repo.count_pending_total()
if user_pending >= USER_PENDING_LIMIT:
raise HTTPException(
status_code=429,
detail=f"您的待处理任务过多（当前 {user_pending}/{USER_PENDING_LIMIT}），请等待完成后再提交",
)
if global_pending >= GLOBAL_PENDING_LIMIT:
raise HTTPException(
status_code=503,
detail="系统繁忙，请稍后再试",
)
except HTTPException:
raise
except Exception as e:
logger.warning("[模板编辑器队列限流] 检查失败，跳过: %s", e)
@router.post("/generate", response_model=EditPlanGenerateResponse)
def generate_editor_draft(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
asset_library_repo: Any = Depends(get_asset_library_repository),
asset_repo: Any = Depends(get_asset_repository),
) -> EditPlanGenerateResponse:
"""触发模板草稿渲染生成"""
_, plan_svc = services
plan_check = plan_svc.get_plan_or_raise(plan_id)
# 自动兜底流程
_auto_fallback_draft_to_editing(plan_svc, plan_id, plan_check)
_auto_fallback_copy_template_clips(plan_svc, plan_id, plan_check, db)
clips_without_asset = _auto_fallback_assign_assets(plan_svc, plan_id, plan_check)
_auto_fallback_auto_material_mode(
plan_svc, plan_id, plan_check, clips_without_asset, asset_library_repo, asset_repo
)
# 检查是否可生成
try:
can_gen, reason = plan_svc.can_generate(plan_id)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
if not can_gen:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=reason)
try:
clip_count = plan_svc.mark_clips_ready(plan_id)
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
user_id = current_user.user_id
_check_queue_limits(gen_task_repo, user_id)
gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
plan = plan_svc.get_plan_or_raise(plan_id)
config_asset_ids = (plan.config or {}).get("asset_ids", [])
gen_task = gen_task_use_case.execute(
CreateGenerationTaskCommand(
project_id=plan.project_id or "",
template_id=plan.template_id,
created_by_user_id=current_user.user_id,
source_edit_plan_id=plan_id,
asset_ids=list(config_asset_ids) if config_asset_ids else [],
),
)
plan_svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id})
plan_svc.transition_status(plan_id, EditPlanStatus.RENDERING)
celery_app.send_task("worker.render_edit_plan", args=[plan_id])
updated_plan = plan_svc.get_plan_or_raise(plan_id)
logger.info(
"模板编辑器触发生成: template_id=%s plan_id=%s gen_task_id=%s clips=%d by user=%s",
template_id,
plan_id,
gen_task.id,
clip_count,
current_user.user_id,
)
return EditPlanGenerateResponse(
plan_id=plan_id,
plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
generation_task_id=gen_task.id,
clip_count=clip_count,
)
except HTTPException:
raise
except Exception as _e:
logger.exception("模板编辑器触发生成失败: template_id=%s plan_id=%s", template_id, plan_id)
try:
plan_svc.transition_status(plan_id, EditPlanStatus.FAILED)
except Exception:
pass
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="生成失败，请稍后重试",
) from _e
@router.get("/generation-status", response_model=EditPlanGenerationStatusResponse)
def get_editor_generation_status(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
storage_service: OSSStorageService = Depends(get_storage_service),
_: AuthenticatedUser = Depends(get_current_user),
) -> EditPlanGenerationStatusResponse:
"""查询草稿生成进度"""
_, plan_svc = services
try:
gen_status = plan_svc.get_generation_status(plan_id)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
plan = gen_status["plan"]
clips = gen_status["clips"]
clip_items = [
ClipStatusItem(
clip_id=c.id,
clip_type=c.clip_type,
order=c.order,
status=c.status.value if hasattr(c.status, "value") else c.status,
asset_id=c.asset_id or "",
text_content=c.text_content or "",
duration=c.duration,
)
for c in clips
]
raw_video_url = (plan.config or {}).get("rendered_url", "")
video_url = ""
if raw_video_url:
try:
video_url = storage_service.get_download_url(raw_video_url, expires_seconds=86400)
except Exception as e:
logger.warning("生成视频签名URL失败: template_id=%s error=%s", template_id, e)
video_url = raw_video_url
progress = gen_status.get("progress", 0.0)
error_message = gen_status.get("error_message", "")
gen_task_status = gen_status.get("generation_task_status")
plan_status_val = plan.status.value if hasattr(plan.status, "value") else plan.status
if plan_status_val == "completed" and progress < 100:
progress = 100.0
return EditPlanGenerationStatusResponse(
plan_id=plan_id,
plan_status=plan_status_val,
generation_task_id=gen_status["generation_task_id"],
generation_task_status=gen_task_status,
progress=progress,
video_url=video_url,
error_message=error_message,
clips=clip_items,
)
@router.get("/generations", response_model=EditPlanGenerationsResponse)
def list_editor_generations(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
db: Session = Depends(get_db_session),
_: AuthenticatedUser = Depends(get_current_user),
) -> EditPlanGenerationsResponse:
"""查询草稿关联的生成记录列表"""
_, plan_svc = services
plan_svc.get_plan_or_raise(plan_id)
from app.schemas.generation_task import GenerationTaskResponse
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
items = [
GenerationTaskResponse(
id=t.id,
project_id=t.project_id,
asset_library_id=t.asset_library_id,
strategy_id=t.strategy_id,
voice_library_id=t.voice_library_id,
template_id=t.template_id,
asset_ids=t.asset_ids,
title_ids=t.title_ids,
voice_ids=t.voice_ids,
source_edit_plan_id=t.source_edit_plan_id or "",
status=t.status.value if hasattr(t.status, "value") else t.status,
progress=t.progress,
result_count=t.result_count,
error_message=t.error_message,
)
for t in tasks
]
return EditPlanGenerationsResponse(items=items, total=len(items))
# ── 时间线模块 ────────────────────────────────────────────────────────────────
class EditorTimelineSceneResponse(BaseModel):
"""时间线场景"""
scene: str
time: str
duration: float
color: str
clip_id: str = ""
clip_type: str = ""
class EditorTimelineResponse(BaseModel):
"""时间线响应"""
plan_id: str
total_duration: float
scenes: List[EditorTimelineSceneResponse]
_CLIP_TYPE_COLORS = {
"intro": "#6366f1",
"title": "#6366f1",
"product": "#818cf8",
"showcase": "#10b981",
"scene": "#10b981",
"subtitle": "#f59e0b",
"text": "#f59e0b",
"cta": "#ef4444",
"outro": "#ef4444",
"voiceover": "#8b5cf6",
"transition": "#64748b",
}
_DEFAULT_COLOR = "#6366f1"
def _format_time(seconds: float) -> str:
m = int(seconds) // 60
s = int(seconds) % 60
return f"{m}:{s:02d}"
def _clip_type_to_scene_label(clip_type: str, text_content: str) -> str:
type_labels = {
"intro": "开场",
"title": "标题",
"product": "产品展示",
"showcase": "场景展示",
"scene": "场景",
"subtitle": "字幕",
"text": "文字",
"cta": "结尾 CTA",
"outro": "结尾",
"voiceover": "配音",
"transition": "转场",
}
label = type_labels.get(clip_type, clip_type or "片段")
if text_content:
short = text_content[:20].strip()
if short:
return f"{label} - {short}"
return label
@router.get("/timeline", response_model=EditorTimelineResponse)
def get_editor_timeline(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> EditorTimelineResponse:
"""获取草稿的时间线场景数据"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
clips = plan_svc.list_clips(plan_id=plan_id, skip=0, limit=200)
clips.sort(key=lambda c: c.order)
scenes = []
current_time = 0.0
for clip in clips:
start = current_time
end = start + clip.duration
color = _CLIP_TYPE_COLORS.get(clip.clip_type, _DEFAULT_COLOR)
scene_label = _clip_type_to_scene_label(clip.clip_type, clip.text_content)
scenes.append(
EditorTimelineSceneResponse(
scene=scene_label,
time=f"{_format_time(start)} - {_format_time(end)}",
duration=clip.duration,
color=color,
clip_id=clip.id,
clip_type=clip.clip_type,
)
)
current_time = end
total_duration = sum(s.duration for s in scenes) or plan.total_duration
return EditorTimelineResponse(
plan_id=plan_id,
total_duration=total_duration,
scenes=scenes,
)
# ── 转场模块 ────────────────────────────────────────────────────────────────
@router.get("/transition-presets", response_model=TransitionPresetListResponse)
def list_editor_transition_presets(
_: AuthenticatedUser = Depends(get_current_user),
) -> TransitionPresetListResponse:
"""获取转场预设列表"""
from packages.domain.transition_presets import TRANSITION_PRESETS
items = [
{
"id": p["id"],
"name": p["name"],
"category": p.get("category", "通用"),
"duration": p.get("default_duration", 0.5),
"description": p.get("description", ""),
}
for p in TRANSITION_PRESETS
]
return TransitionPresetListResponse(items=items, total=len(items))
@router.put("/clips/{clip_id}/transition", response_model=ClipTransitionResponse)
def update_editor_clip_transition(
template_id: str,
clip_id: str,
body: TransitionUpdateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> ClipTransitionResponse:
"""设置单个片段的转场效果"""
_, plan_svc = services
try:
clip = plan_svc.update_clip(
clip_id,
transition_effect=body.effect,
transition_duration=body.duration,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return ClipTransitionResponse(
clip_id=clip.id,
effect=clip.transition_effect.value if hasattr(clip.transition_effect, "value") else clip.transition_effect,
duration=clip.transition_duration or 0.5,
)
@router.post("/transitions/batch", response_model=BatchTransitionResponse)
def batch_update_editor_transitions(
template_id: str,
body: BatchTransitionRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> BatchTransitionResponse:
"""批量设置所有片段的转场效果"""
_, plan_svc = services
clips = plan_svc.list_clips(plan_id, limit=500)
updated = 0
for clip in clips:
if clip.order > 0: # 第一个片段不加转场
try:
plan_svc.update_clip(
clip.id,
transition_effect=body.effect,
transition_duration=body.duration,
)
updated += 1
except ValueError:
pass
return BatchTransitionResponse(
updated_count=updated,
effect=body.effect,
duration=body.duration,
)
# ── 滤镜模块 ────────────────────────────────────────────────────────────────
@router.get("/filter-presets", response_model=FilterPresetListResponse)
def list_editor_filter_presets(
_: AuthenticatedUser = Depends(get_current_user),
) -> FilterPresetListResponse:
"""获取滤镜预设列表"""
from packages.domain.filter_presets import FILTER_PRESETS
items = [
{
"id": p["id"],
"name": p["name"],
"category": p.get("category", "通用"),
"thumbnail": p.get("thumbnail", ""),
"description": p.get("description", ""),
}
for p in FILTER_PRESETS
]
return FilterPresetListResponse(items=items, total=len(items))
@router.get("/filter", response_model=FilterConfigResponse)
def get_editor_filter(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> FilterConfigResponse:
"""获取草稿的全局滤镜配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = plan.config or {}
filter_config = config.get("filter", {})
return FilterConfigResponse(
plan_id=plan.id,
enabled=filter_config.get("enabled", False),
preset_id=filter_config.get("preset_id", ""),
intensity=filter_config.get("intensity", 1.0),
)
@router.put("/filter", response_model=FilterConfigResponse)
def update_editor_filter(
template_id: str,
body: FilterUpdateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> FilterConfigResponse:
"""更新草稿的全局滤镜配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = dict(plan.config) if plan.config else {}
current_filter = dict(config.get("filter", {}))
update_data = body.model_dump(exclude_none=True)
current_filter.update(update_data)
config["filter"] = current_filter
updated_plan = plan_svc.update_plan_config(plan_id, normalize_plan_config(config))
return FilterConfigResponse(
plan_id=updated_plan.id,
enabled=current_filter.get("enabled", False),
preset_id=current_filter.get("preset_id", ""),
intensity=current_filter.get("intensity", 1.0),
)
# ── 导出模块 ────────────────────────────────────────────────────────────────
@router.get("/export-presets", response_model=ExportPresetListResponse)
def list_editor_export_presets(
_: AuthenticatedUser = Depends(get_current_user),
) -> ExportPresetListResponse:
"""获取导出预设列表"""
from packages.domain.export_presets import EXPORT_PRESETS
items = [
{
"id": p["id"],
"name": p["name"],
"resolution": p.get("resolution", "1080p"),
"fps": p.get("fps", 30),
"bitrate": p.get("bitrate", ""),
"description": p.get("description", ""),
}
for p in EXPORT_PRESETS
]
return ExportPresetListResponse(items=items, total=len(items))
@router.get("/export", response_model=ExportConfigResponse)
def get_editor_export(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> ExportConfigResponse:
"""获取草稿的导出配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = plan.config or {}
export_config = config.get("export", {})
return ExportConfigResponse(
plan_id=plan.id,
resolution=export_config.get("resolution", "1080p"),
fps=export_config.get("fps", 30),
bitrate=export_config.get("bitrate", ""),
format=export_config.get("format", "mp4"),
preset_id=export_config.get("preset_id", ""),
)
@router.put("/export", response_model=ExportConfigResponse)
def update_editor_export(
template_id: str,
body: ExportUpdateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> ExportConfigResponse:
"""更新草稿的导出配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = dict(plan.config) if plan.config else {}
current_export = dict(config.get("export", {}))
update_data = body.model_dump(exclude_none=True)
current_export.update(update_data)
config["export"] = current_export
updated_plan = plan_svc.update_plan_config(plan_id, normalize_plan_config(config))
return ExportConfigResponse(
plan_id=updated_plan.id,
resolution=current_export.get("resolution", "1080p"),
fps=current_export.get("fps", 30),
bitrate=current_export.get("bitrate", ""),
format=current_export.get("format", "mp4"),
preset_id=current_export.get("preset_id", ""),
)