diff --git a/apps/api/app/api/routes/templates_editor.py b/apps/api/app/api/routes/templates_editor.py index 895a9808e..d7c9848f0 100755 --- a/apps/api/app/api/routes/templates_editor.py +++ b/apps/api/app/api/routes/templates_editor.py @@ -25,14 +25,89 @@ from __future__ import annotations import logging from typing import Any, List, Optional +from app.api.routes.edit_plans import ( + AIRecommendRequest, + AIRecommendResponse, + BGMConfigUpdateRequest, + ClipStatusItem, + EditPlanGenerateResponse, + EditPlanGenerationsResponse, + EditPlanGenerationStatusResponse, + GenerateCoverRequest, + GenerateCoverResponse, +) +from app.api.routes.edit_plans_adjustments import ( + BatchSpeedRequest, + BatchSpeedResponse, + ClipAdjustmentsRequest, + ClipAdjustResponse, + SpeedAdjustRequest, + TrimAdjustRequest, + VolumeAdjustRequest, +) +from app.api.routes.edit_plans_clips_batch import ( + ClipBatchDeleteRequest, + ClipBatchDeleteResponse, + ClipReorderRequest, + ClipReorderResponse, + ClipsFromAssetsRequest, + ClipsFromAssetsResponse, +) +from app.api.routes.edit_plans_cover import ( + CoverConfigResponse, + CoverExtractRequest, + CoverGenerateResponse, + CoverSmartRequest, + CoverUpdateRequest, +) +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 +426,1580 @@ def delete_draft_clip( if not success: raise HTTPException(status_code=404, detail="片段不存在") return None + + +@router.get("/clips/{clip_id}", response_model=EditorClipResponse) +def get_draft_clip_detail( + template_id: str, + clip_id: str, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +): + """获取草稿中的片段详情""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if clip is None: + raise HTTPException(status_code=404, detail="片段不存在") + if clip.plan_id != plan_id: + raise HTTPException(status_code=404, detail="片段不存在") + return EditorClipResponse( + id=clip.id, + plan_id=clip.plan_id, + clip_type=clip.clip_type.value if hasattr(clip.clip_type, "value") else str(clip.clip_type), + order=clip.order, + duration=clip.duration, + text_content=clip.text_content or "", + transition_effect=( + clip.transition_effect.value if hasattr(clip.transition_effect, "value") else str(clip.transition_effect) + ), + playback_speed=clip.playback_speed or 1.0, + config=clip.config or {}, + ) + + +# ── 片段分割与合并 ────────────────────────────────────────────────────────── + + +class SplitClipRequest(BaseModel): + """分割片段请求体""" + + split_time: float = Field(..., gt=0, description="分割点（秒，相对于片段起始）") + + +class MergeClipsRequest(BaseModel): + """合并片段请求体""" + + clip_ids: list[str] = Field(..., min_length=2, description="要合并的片段 ID 列表") + + +@router.post("/clips/{clip_id}/split", response_model=dict[str, Any], status_code=status.HTTP_200_OK) +def split_draft_clip( + template_id: str, + clip_id: str, + body: SplitClipRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +): + """将一个片段从指定时间点分割为两个片段""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if clip is None or clip.plan_id != plan_id: + raise HTTPException(status_code=404, detail="片段不存在") + try: + result = plan_svc.split_clip(clip_id, body.split_time) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + left = result["left_clip"] + right = result["right_clip"] + return { + "left_clip": { + "id": left.id, + "plan_id": left.plan_id, + "clip_type": left.clip_type, + "order": left.order, + "duration": left.duration, + "start_time": left.start_time, + }, + "right_clip": { + "id": right.id, + "plan_id": right.plan_id, + "clip_type": right.clip_type, + "order": right.order, + "duration": right.duration, + "start_time": right.start_time, + }, + } + + +@router.post("/clips/merge", response_model=dict[str, Any], status_code=status.HTTP_200_OK) +def merge_draft_clips( + template_id: str, + body: MergeClipsRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +): + """将多个连续的同类型片段合并为一个片段""" + _, plan_svc = services + for cid in body.clip_ids: + clip = plan_svc.get_clip(cid) + if clip is None or clip.plan_id != plan_id: + raise HTTPException(status_code=404, detail=f"片段不存在: {cid}") + try: + merged = plan_svc.merge_clips(body.clip_ids) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + return { + "id": merged.id, + "plan_id": merged.plan_id, + "clip_type": merged.clip_type, + "order": merged.order, + "duration": merged.duration, + "text_content": merged.text_content, + } + + +# ── 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", ""), + ) + + +# ── AI 推荐 & 封面生成 ──────────────────────────────────────────────────────── + + +@router.post("/ai-recommend", response_model=AIRecommendResponse) +def editor_ai_recommend( + template_id: str, + body: AIRecommendRequest, + 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), +) -> AIRecommendResponse: + """AI 推荐片段方案""" + _, plan_svc = services + plan = plan_svc.get_plan_or_raise(plan_id) + + plan_status = plan.status.value if hasattr(plan.status, "value") else plan.status + if plan_status not in ("draft", "editing"): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="当前草稿状态不支持AI推荐，请先编辑后再试", + ) + + from apps.worker.worker_app.tasks.ai_tasks import run_ai_recommend + + result = run_ai_recommend( + plan_id=plan_id, + template_id=plan.template_id, + asset_ids=body.asset_ids, + editing_mode=body.editing_mode, + target_duration=body.target_duration, + ) + + try: + plan_svc.delete_all_clips(plan_id) + + for clip_data in result["clips"]: + plan_svc.create_clip( + plan_id=plan_id, + clip_type=clip_data["clip_type"], + order=clip_data["order"], + text_content=clip_data.get("text_content", ""), + duration=clip_data["duration"], + transition_effect=clip_data.get("transition_effect", "cut"), + asset_id=clip_data.get("asset_id", ""), + start_time=clip_data.get("start_time", 0.0), + config=clip_data.get("config", {}), + ) + + normalized_config = normalize_plan_config(result.get("config", {})) + plan_svc.update_plan( + plan_id, + config=normalized_config, + total_duration=result["total_duration"], + ) + except Exception as _e: + logger.exception("模板编辑器AI推荐写入失败: template_id=%s plan_id=%s", template_id, plan_id) + try: + db.rollback() + except Exception: + pass + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="AI推荐结果保存失败，请稍后重试", + ) from _e + + logger.info( + "模板编辑器AI推荐: template_id=%s plan_id=%s clips=%d duration=%.1f by user=%s", + template_id, + plan_id, + len(result["clips"]), + result["total_duration"], + current_user.user_id, + ) + + return AIRecommendResponse( + plan_id=plan_id, + clips=[ + { + "clip_type": c["clip_type"], + "order": c["order"], + "text_content": c.get("text_content", ""), + "duration": c["duration"], + "transition_effect": c.get("transition_effect", "cut"), + "asset_id": c.get("asset_id", ""), + "start_time": c.get("start_time", 0.0), + "config": c.get("config", {}), + } + for c in result["clips"] + ], + config=normalized_config, + total_duration=result["total_duration"], + confidence=result["confidence"], + ) + + +@router.post("/generate-cover", response_model=GenerateCoverResponse) +def editor_generate_cover( + template_id: str, + body: GenerateCoverRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + current_user: AuthenticatedUser = Depends(get_current_user), +) -> GenerateCoverResponse: + """AI 生成封面""" + _, plan_svc = services + plan = plan_svc.get_plan_or_raise(plan_id) + + from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover + + cover_data = run_generate_cover( + plan_id=plan_id, + asset_ids=body.asset_ids, + cover_type=body.cover_type, + frame_time=body.frame_time, + ) + + current_config = dict(plan.config) if plan.config else {} + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + logger.info( + "模板编辑器封面生成: template_id=%s plan_id=%s type=%s by user=%s", + template_id, + plan_id, + body.cover_type, + current_user.user_id, + ) + + return GenerateCoverResponse(plan_id=plan_id, cover=cover_data) + + +# ── 字幕模块 ──────────────────────────────────────────────────────────────── + + +@router.get("/clips/{clip_id}/subtitles", response_model=list[dict[str, Any]]) +def get_editor_clip_subtitles( + template_id: str, + clip_id: str, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> list[dict[str, Any]]: + """获取片段的字幕列表""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + config = clip.config or {} + subtitles = config.get("subtitles", []) + if not isinstance(subtitles, list): + subtitles = [] + return subtitles + + +@router.post("/clips/{clip_id}/subtitles", response_model=dict[str, Any]) +def create_editor_clip_subtitle( + template_id: str, + clip_id: str, + body: dict[str, Any], + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> dict[str, Any]: + """新增片段字幕""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + config = dict(clip.config) if clip.config else {} + subtitles = config.get("subtitles", []) + if not isinstance(subtitles, list): + subtitles = [] + + new_id = f"sub_{len(subtitles) + 1}" + new_subtitle = { + "id": body.get("id", new_id), + "start_time": body.get("start_time", 0.0), + "end_time": body.get("end_time", 0.0), + "text": body.get("text", ""), + "style": body.get("style", {}), + } + subtitles.append(new_subtitle) + config["subtitles"] = subtitles + + plan_svc.update_clip(clip_id, config=config) + return new_subtitle + + +@router.put("/clips/{clip_id}/subtitles/{subtitle_id}", response_model=dict[str, Any]) +def update_editor_clip_subtitle( + template_id: str, + clip_id: str, + subtitle_id: str, + body: dict[str, Any], + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> dict[str, Any]: + """更新片段字幕""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + config = dict(clip.config) if clip.config else {} + subtitles = config.get("subtitles", []) + if not isinstance(subtitles, list): + subtitles = [] + + found = False + for i, sub in enumerate(subtitles): + if sub.get("id") == subtitle_id: + subtitles[i].update(body) + found = True + break + + if not found: + raise HTTPException(status_code=404, detail="字幕不存在") + + config["subtitles"] = subtitles + plan_svc.update_clip(clip_id, config=config) + return subtitles[i] + + +@router.delete("/clips/{clip_id}/subtitles/{subtitle_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_editor_clip_subtitle( + template_id: str, + clip_id: str, + subtitle_id: str, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +): + """删除片段字幕""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + config = dict(clip.config) if clip.config else {} + subtitles = config.get("subtitles", []) + if not isinstance(subtitles, list): + subtitles = [] + + new_subtitles = [s for s in subtitles if s.get("id") != subtitle_id] + if len(new_subtitles) == len(subtitles): + raise HTTPException(status_code=404, detail="字幕不存在") + + config["subtitles"] = new_subtitles + plan_svc.update_clip(clip_id, config=config) + return None + + +@router.put("/clips/{clip_id}/subtitles", response_model=list[dict[str, Any]]) +def batch_update_editor_clip_subtitles( + template_id: str, + clip_id: str, + body: list[dict[str, Any]], + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> list[dict[str, Any]]: + """批量更新片段字幕（全量替换）""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + config = dict(clip.config) if clip.config else {} + config["subtitles"] = body + plan_svc.update_clip(clip_id, config=config) + return body + + +# ── 片段调整模块 ────────────────────────────────────────────────────────────── + + +def _get_clip_config(clip) -> dict: + config = getattr(clip, "config", {}) or {} + if not isinstance(config, dict): + config = {} + return config + + +def _get_adjust_volume(clip) -> float: + config = _get_clip_config(clip) + return float(config.get("volume", 1.0)) + + +def _get_adjust_trim(clip) -> tuple[float, float]: + config = _get_clip_config(clip) + trim_start = float(config.get("trim_start", 0.0)) + trim_end = float(config.get("trim_end", 0.0)) + return trim_start, trim_end + + +def _build_adjust_response(clip) -> ClipAdjustResponse: + trim_start, trim_end = _get_adjust_trim(clip) + return ClipAdjustResponse( + clip_id=clip.id, + speed=clip.playback_speed, + volume=_get_adjust_volume(clip), + trim_start=trim_start, + trim_end=trim_end, + duration=clip.duration, + ) + + +def _validate_trim(trim_start: float, trim_end: float, total_duration: float) -> None: + if trim_start + trim_end >= total_duration: + raise ValueError(f"裁剪总时长（{trim_start + trim_end:.2f}s）不能大于等于片段总时长（{total_duration:.2f}s）") + + +@router.put("/clips/{clip_id}/speed", response_model=ClipAdjustResponse) +def adjust_editor_clip_speed( + template_id: str, + clip_id: str, + body: SpeedAdjustRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> ClipAdjustResponse: + """调整片段播放速度""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + updated = plan_svc.update_clip(clip_id, playback_speed=body.speed) + return _build_adjust_response(updated) + + +@router.put("/clips/{clip_id}/volume", response_model=ClipAdjustResponse) +def adjust_editor_clip_volume( + template_id: str, + clip_id: str, + body: VolumeAdjustRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> ClipAdjustResponse: + """调整片段音量""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + config = dict(_get_clip_config(clip)) + config["volume"] = body.volume + updated = plan_svc.update_clip(clip_id, config=config) + return _build_adjust_response(updated) + + +@router.put("/clips/{clip_id}/trim", response_model=ClipAdjustResponse) +def adjust_editor_clip_trim( + template_id: str, + clip_id: str, + body: TrimAdjustRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> ClipAdjustResponse: + """裁剪片段（trim in/out）""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + try: + _validate_trim(body.trim_start, body.trim_end, clip.duration) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + config = dict(_get_clip_config(clip)) + config["trim_start"] = body.trim_start + config["trim_end"] = body.trim_end + updated = plan_svc.update_clip(clip_id, config=config) + return _build_adjust_response(updated) + + +@router.put("/clips/{clip_id}/adjustments", response_model=ClipAdjustResponse) +def adjust_editor_clip_all( + template_id: str, + clip_id: str, + body: ClipAdjustmentsRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> ClipAdjustResponse: + """统一调整片段的 speed / volume / trim""" + _, plan_svc = services + clip = plan_svc.get_clip(clip_id) + if not clip: + raise HTTPException(status_code=404, detail="片段不存在") + + update_kwargs: dict[str, Any] = {} + config_updates: dict[str, Any] = {} + + if body.speed is not None: + update_kwargs["playback_speed"] = body.speed + if body.volume is not None: + config_updates["volume"] = body.volume + if body.trim_start is not None: + config_updates["trim_start"] = body.trim_start + if body.trim_end is not None: + config_updates["trim_end"] = body.trim_end + + current_trim_start, current_trim_end = _get_adjust_trim(clip) + new_trim_start = body.trim_start if body.trim_start is not None else current_trim_start + new_trim_end = body.trim_end if body.trim_end is not None else current_trim_end + + if body.trim_start is not None or body.trim_end is not None: + try: + _validate_trim(new_trim_start, new_trim_end, clip.duration) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + if config_updates: + config = dict(_get_clip_config(clip)) + config.update(config_updates) + update_kwargs["config"] = config + + if not update_kwargs: + return _build_adjust_response(clip) + + updated = plan_svc.update_clip(clip_id, **update_kwargs) + return _build_adjust_response(updated) + + +@router.post("/clips/batch-speed", response_model=BatchSpeedResponse) +def batch_adjust_editor_speed( + template_id: str, + body: BatchSpeedRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> BatchSpeedResponse: + """批量调整草稿内所有片段的播放速度""" + _, plan_svc = services + clips = plan_svc.list_clips(plan_id, limit=500, skip=0) + count = 0 + for clip in clips: + plan_svc.update_clip(clip.id, playback_speed=body.speed) + count += 1 + + return BatchSpeedResponse(updated_count=count, plan_id=plan_id) + + +# ── 封面管理模块 ────────────────────────────────────────────────────────────── + + +@router.get("/cover", response_model=CoverConfigResponse) +def get_editor_cover( + template_id: str, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> CoverConfigResponse: + """获取草稿封面配置""" + _, plan_svc = services + plan = plan_svc.get_plan_or_raise(plan_id) + config = plan.config or {} + cover_config = config.get("cover", {}) + + return CoverConfigResponse( + plan_id=plan.id, + cover_type=cover_config.get("cover_type", "auto"), + cover_image_url=cover_config.get("cover_image_url", ""), + clip_id=cover_config.get("clip_id", ""), + frame_time=cover_config.get("frame_time", 0.0), + ) + + +@router.put("/cover", response_model=CoverConfigResponse) +def update_editor_cover( + template_id: str, + body: CoverUpdateRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> CoverConfigResponse: + """更新草稿封面配置""" + _, plan_svc = services + plan = plan_svc.get_plan_or_raise(plan_id) + + config = dict(plan.config) if plan.config else {} + current_cover = dict(config.get("cover", {})) + update_data = body.model_dump(exclude_none=True) + current_cover.update(update_data) + + config["cover"] = current_cover + normalized = normalize_plan_config(config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + return CoverConfigResponse( + plan_id=plan.id, + cover_type=current_cover.get("cover_type", "auto"), + cover_image_url=current_cover.get("cover_image_url", ""), + clip_id=current_cover.get("clip_id", ""), + frame_time=current_cover.get("frame_time", 0.0), + ) + + +@router.post("/cover/extract", response_model=CoverGenerateResponse) +def extract_editor_cover( + template_id: str, + body: CoverExtractRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + current_user: AuthenticatedUser = Depends(get_current_user), +) -> CoverGenerateResponse: + """从指定片段抽帧生成封面""" + _, plan_svc = services + plan = plan_svc.get_plan_or_raise(plan_id) + + clip = plan_svc.get_clip(body.clip_id) + if not clip or clip.plan_id != plan_id: + raise HTTPException(status_code=400, detail="片段不存在或不属于当前草稿") + + cover_url = f"cover/extract/{plan_id}_{body.clip_id}_{body.frame_time}.jpg" + + config = dict(plan.config) if plan.config else {} + cover_config = dict(config.get("cover", {})) + cover_config.update( + { + "cover_type": "extract", + "cover_image_url": cover_url, + "clip_id": body.clip_id, + "frame_time": body.frame_time, + } + ) + config["cover"] = cover_config + normalized = normalize_plan_config(config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + logger.info( + "模板编辑器封面抽帧: template_id=%s plan_id=%s clip_id=%s by user=%s", + template_id, + plan_id, + body.clip_id, + current_user.user_id, + ) + + return CoverGenerateResponse( + plan_id=plan_id, + cover_url=cover_url, + cover_type="extract", + ) + + +@router.post("/cover/smart", response_model=CoverGenerateResponse) +def smart_editor_cover( + template_id: str, + body: CoverSmartRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + current_user: AuthenticatedUser = Depends(get_current_user), +) -> CoverGenerateResponse: + """智能选帧生成封面""" + _, plan_svc = services + plan = plan_svc.get_plan_or_raise(plan_id) + + cover_url = f"cover/smart/{plan_id}_smart.jpg" + + config = dict(plan.config) if plan.config else {} + cover_config = dict(config.get("cover", {})) + cover_config.update( + { + "cover_type": "smart", + "cover_image_url": cover_url, + "strategy": body.strategy, + } + ) + config["cover"] = cover_config + normalized = normalize_plan_config(config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + logger.info( + "模板编辑器智能封面: template_id=%s plan_id=%s strategy=%s by user=%s", + template_id, + plan_id, + body.strategy, + current_user.user_id, + ) + + return CoverGenerateResponse( + plan_id=plan_id, + cover_url=cover_url, + cover_type="smart", + ) + + +# ── 片段批量操作模块 ──────────────────────────────────────────────────────── + + +@router.post("/clips/reorder", response_model=ClipReorderResponse) +def reorder_editor_clips( + template_id: str, + body: ClipReorderRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> ClipReorderResponse: + """批量重排片段顺序""" + _, plan_svc = services + count = 0 + for item in body.items: + try: + plan_svc.update_clip(item.clip_id, order=item.order) + count += 1 + except ValueError: + pass + + return ClipReorderResponse(updated_count=count, plan_id=plan_id) + + +@router.post("/clips/batch-delete", response_model=ClipBatchDeleteResponse) +def batch_delete_editor_clips( + template_id: str, + body: ClipBatchDeleteRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +) -> ClipBatchDeleteResponse: + """批量删除片段""" + _, plan_svc = services + deleted = 0 + for clip_id in body.clip_ids: + if plan_svc.delete_clip(clip_id): + deleted += 1 + + return ClipBatchDeleteResponse(deleted_count=deleted, plan_id=plan_id) + + +@router.post("/clips/from-assets", response_model=ClipsFromAssetsResponse) +def create_clips_from_assets_editor( + template_id: str, + body: ClipsFromAssetsRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + current_user: AuthenticatedUser = Depends(get_current_user), +) -> ClipsFromAssetsResponse: + """从素材批量创建片段""" + _, plan_svc = services + clips = [] + for i, asset_id in enumerate(body.asset_ids): + try: + clip = plan_svc.create_clip( + plan_id, + clip_type="main", + order=body.start_order + i if hasattr(body, "start_order") else i, + duration=5.0, + asset_id=asset_id, + ) + clips.append(clip) + except ValueError: + pass + + logger.info( + "模板编辑器从素材创建片段: template_id=%s plan_id=%s count=%d by user=%s", + template_id, + plan_id, + len(clips), + current_user.user_id, + ) + + return ClipsFromAssetsResponse( + created_count=len(clips), + plan_id=plan_id, + clip_ids=[c.id for c in clips], + ) diff --git a/tests/unit/test_templates_editor_api.py b/tests/unit/test_templates_editor_api.py new file mode 100644 index 000000000..02446c31d --- /dev/null +++ b/tests/unit/test_templates_editor_api.py @@ -0,0 +1,451 @@ +""" +templates_editor.py 模板编辑器 API 端点单元测试 + +覆盖核心端点（25个测试用例）： +- 草稿：GET/PUT/发布 +- 片段：list/create/get/update/delete/split/merge +- BGM：GET/PUT +- 时间线：GET +- 生成状态查询 +- 预设：BGM预设 +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +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 + +# --------------------------------------------------------------------------- +# 测试常量 +# --------------------------------------------------------------------------- + +TEST_TEMPLATE_ID = "tmpl-test-001" +TEST_PLAN_ID = "plan-draft-001" +TEST_USER_ID = "user-001" + + +def _make_auth_user(): + """构造一个认证用户 stub（MagicMock 兼容不同属性名）""" + auth = MagicMock() + auth.user.id = TEST_USER_ID + auth.user.email = "test@example.com" + auth.user.display_name = "测试用户" + auth.user_id = TEST_USER_ID + return auth + + +def _make_mock_clip(clip_id="clip-001", order=0, duration=10.0, clip_type="video"): + """构造一个 mock 片段""" + clip = MagicMock() + clip.id = clip_id + clip.plan_id = TEST_PLAN_ID + clip.clip_type = clip_type + clip.order = order + clip.duration = duration + clip.start_time = 0.0 + clip.text_content = "" + clip.transition_effect = "none" + clip.playback_speed = 1.0 + clip.config = {} + clip.asset_id = "asset-001" + clip.status = "ready" + return clip + + +def _make_mock_plan(status="editing", config=None): + """构造一个 mock 剪辑计划""" + plan = MagicMock() + plan.id = TEST_PLAN_ID + plan.status = status + plan.config = config or {"is_template_draft": True, "asset_ids": []} + plan.template_id = TEST_TEMPLATE_ID + plan.project_id = "proj-001" + plan.name = "测试草稿" + plan.total_duration = 30.0 + plan.generation_task_id = None + return plan + + +# --------------------------------------------------------------------------- +# Test App Setup +# --------------------------------------------------------------------------- + + +def _create_test_app(): + """创建带 mock 注入的模板编辑器测试应用""" + from app.api.routes import templates_editor as editor_module + + mock_plan = _make_mock_plan() + mock_clip_1 = _make_mock_clip("clip-001", 0, 10.0) + mock_clip_2 = _make_mock_clip("clip-002", 1, 20.0) + + mock_template_svc = MagicMock() + mock_template_svc.publish_template_from_draft.return_value = MagicMock( + id=TEST_TEMPLATE_ID, + name="发布后的模板", + status="published", + ) + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_plan_svc.update_plan.return_value = mock_plan + mock_plan_svc.update_plan_config.return_value = mock_plan + mock_plan_svc.list_clips.return_value = [mock_clip_1, mock_clip_2] + mock_plan_svc.count_clips.return_value = 2 + mock_plan_svc.get_clip.return_value = mock_clip_1 + mock_plan_svc.create_clip.return_value = _make_mock_clip("clip-new", order=2) + mock_plan_svc.update_clip.return_value = _make_mock_clip("clip-001", duration=15.0) + mock_plan_svc.delete_clip.return_value = True + mock_plan_svc.split_clip.return_value = { + "left_clip": _make_mock_clip("clip-left", 0, 5.0), + "right_clip": _make_mock_clip("clip-right", 1, 5.0), + } + mock_plan_svc.merge_clips.return_value = _make_mock_clip("clip-merged", 0, 20.0) + mock_plan_svc.can_generate.return_value = (True, None) + mock_plan_svc.mark_clips_ready.return_value = 2 + mock_plan_svc.transition_status.return_value = mock_plan + + # 生成任务 mock + mock_gen_task = MagicMock() + mock_gen_task.id = "task-001" + mock_gen_task.status = "pending" + mock_plan_svc.create_generation_task.return_value = mock_gen_task + # 生成状态返回结构（需要 plan + clips） + mock_plan_svc.get_generation_status.return_value = { + "plan": mock_plan, + "clips": [mock_clip_1, mock_clip_2], + "task_id": "task-001", + "generation_task_id": "task-001", + "status": "processing", + } + mock_plan_svc.list_generation_tasks.return_value = {"items": [], "total": 0} + + # mock get_editor_services 依赖 + def _mock_get_editor_services(): + return mock_template_svc, mock_plan_svc + + app = FastAPI() + app.include_router( + editor_module.router, + prefix="/api/v1/templates/{template_id}/editor", + ) + + # 覆盖依赖 + app.dependency_overrides[editor_module.get_current_user] = _make_auth_user + app.dependency_overrides[editor_module.get_db_session] = lambda: MagicMock() + app.dependency_overrides[editor_module.get_draft_plan_id] = lambda: TEST_PLAN_ID + app.dependency_overrides[editor_module.get_editor_services] = _mock_get_editor_services + + return app, mock_template_svc, mock_plan_svc + + +@pytest.fixture +def client(): + app, mock_tpl_svc, mock_plan_svc = _create_test_app() + yield TestClient(app), mock_tpl_svc, mock_plan_svc + + +BASE = f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor" + + +# --------------------------------------------------------------------------- +# 草稿端点测试 +# --------------------------------------------------------------------------- + + +class TestDraftEndpoints: + """草稿查询/更新/发布端点测试""" + + def test_get_draft_success(self, client): + c, _, _ = client + resp = c.get(BASE + "/") + assert resp.status_code == 200 + data = resp.json() + assert data["plan_id"] == TEST_PLAN_ID + assert data["template_id"] == TEST_TEMPLATE_ID + assert "name" in data + assert "config" in data + assert "clip_count" in data + + def test_get_draft_returns_is_template_draft(self, client): + c, _, _ = client + resp = c.get(BASE + "/") + data = resp.json() + assert data["config"]["is_template_draft"] is True + + def test_update_draft_name(self, client): + c, _, mock_plan_svc = client + resp = c.put(BASE + "/", json={"name": "新名称"}) + assert resp.status_code == 200 + mock_plan_svc.update_plan.assert_called_once() + call_kwargs = mock_plan_svc.update_plan.call_args + assert call_kwargs.kwargs.get("name") == "新名称" or call_kwargs[1].get("name") == "新名称" + + def test_update_draft_empty_body_ok(self, client): + c, _, _ = client + resp = c.put(BASE + "/", json={}) + assert resp.status_code == 200 + + def test_publish_draft_success(self, client): + c, mock_tpl_svc, _ = client + resp = c.post(BASE + "/publish") + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "published" + assert data["template_id"] == TEST_TEMPLATE_ID + mock_tpl_svc.publish_template_from_draft.assert_called_once_with(TEST_TEMPLATE_ID, TEST_PLAN_ID) + + +# --------------------------------------------------------------------------- +# 片段端点测试 +# --------------------------------------------------------------------------- + + +class TestClipEndpoints: + """片段 CRUD 端点测试""" + + def test_list_clips_success(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/clips") + assert resp.status_code == 200 + data = resp.json() + assert "items" in data + assert "total" in data + assert data["total"] == 2 + assert len(data["items"]) == 2 + mock_plan_svc.list_clips.assert_called_once() + + def test_list_clips_pagination_params(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/clips?skip=10&limit=20") + assert resp.status_code == 200 + mock_plan_svc.list_clips.assert_called_once_with(TEST_PLAN_ID, skip=10, limit=20) + + def test_create_clip_success(self, client): + c, _, mock_plan_svc = client + resp = c.post( + BASE + "/clips", + json={"clip_type": "video", "order": 2, "duration": 5.0}, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["id"] == "clip-new" + mock_plan_svc.create_clip.assert_called_once() + + def test_create_clip_missing_type_422(self, client): + c, _, _ = client + resp = c.post(BASE + "/clips", json={"order": 0, "duration": 5.0}) + assert resp.status_code == 422 + + def test_get_clip_detail_success(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/clips/clip-001") + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == "clip-001" + mock_plan_svc.get_clip.assert_called_once_with("clip-001") + + def test_get_clip_not_found_404(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.get_clip.return_value = None + resp = c.get(BASE + "/clips/nonexistent") + assert resp.status_code == 404 + + def test_update_clip_success(self, client): + c, _, mock_plan_svc = client + resp = c.put( + BASE + "/clips/clip-001", + json={"duration": 15.0, "playback_speed": 2.0}, + ) + assert resp.status_code == 200 + mock_plan_svc.update_clip.assert_called_once() + + def test_delete_clip_success(self, client): + c, _, mock_plan_svc = client + resp = c.delete(BASE + "/clips/clip-001") + assert resp.status_code == 204 + mock_plan_svc.delete_clip.assert_called_once_with("clip-001") + + def test_delete_clip_not_found_404(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.delete_clip.return_value = False + resp = c.delete(BASE + "/clips/nonexistent") + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# 片段分割合并测试 +# --------------------------------------------------------------------------- + + +class TestClipSplitMerge: + """片段分割与合并端点测试""" + + def test_split_clip_success(self, client): + c, _, mock_plan_svc = client + resp = c.post(BASE + "/clips/clip-001/split", json={"split_time": 5.0}) + assert resp.status_code == 200 + data = resp.json() + assert "left_clip" in data + assert "right_clip" in data + mock_plan_svc.split_clip.assert_called_once_with("clip-001", 5.0) + + def test_split_clip_negative_time_422(self, client): + c, _, _ = client + resp = c.post(BASE + "/clips/clip-001/split", json={"split_time": -1.0}) + assert resp.status_code == 422 + + def test_split_clip_not_found_404(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.get_clip.return_value = None + resp = c.post(BASE + "/clips/nonexistent/split", json={"split_time": 5.0}) + assert resp.status_code == 404 + + def test_merge_clips_success(self, client): + c, _, mock_plan_svc = client + resp = c.post(BASE + "/clips/merge", json={"clip_ids": ["clip-001", "clip-002"]}) + assert resp.status_code == 200 + data = resp.json() + assert "id" in data + mock_plan_svc.merge_clips.assert_called_once_with(["clip-001", "clip-002"]) + + def test_merge_clips_single_422(self, client): + c, _, _ = client + resp = c.post(BASE + "/clips/merge", json={"clip_ids": ["clip-001"]}) + assert resp.status_code == 422 + + def test_merge_clips_not_found_404(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.get_clip.return_value = None + resp = c.post(BASE + "/clips/merge", json={"clip_ids": ["nope", "clip-002"]}) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# BGM 端点测试 +# --------------------------------------------------------------------------- + + +class TestBGMRoutes: + """BGM 配置端点测试""" + + def test_get_bgm_success(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/bgm") + assert resp.status_code == 200 + data = resp.json() + assert "plan_id" in data + assert "bgm" in data + mock_plan_svc.get_plan_or_raise.assert_called() + + def test_update_bgm_success(self, client): + c, _, mock_plan_svc = client + resp = c.put(BASE + "/bgm", json={"enabled": True, "asset_id": "asset-001", "volume": 0.6}) + assert resp.status_code == 200 + mock_plan_svc.update_plan_config.assert_called_once() + + def test_get_bgm_presets_success(self, client): + c, _, _ = client + resp = c.get(BASE + "/bgm/presets") + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, dict) + + +# --------------------------------------------------------------------------- +# 时间线端点测试 +# --------------------------------------------------------------------------- + + +class TestTimelineRoute: + """时间线条端点测试""" + + def test_get_timeline_success(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/timeline") + assert resp.status_code == 200 + data = resp.json() + assert "plan_id" in data + assert "scenes" in data + mock_plan_svc.list_clips.assert_called() + + +# --------------------------------------------------------------------------- +# 生成端点测试 +# --------------------------------------------------------------------------- + + +class TestGenerationRoutes: + """生成端点测试""" + + def test_generation_status_success(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/generation-status") + assert resp.status_code == 200 + data = resp.json() + assert "generation_task_id" in data + assert "clips" in data + + def test_generations_list_success(self, client): + c, _, mock_plan_svc = client + resp = c.get(BASE + "/generations") + assert resp.status_code == 200 + data = resp.json() + assert "items" in data or "tasks" in data or isinstance(data, dict) + + +# --------------------------------------------------------------------------- +# 字幕端点测试 +# --------------------------------------------------------------------------- + + +class TestSubtitleRoutes: + """字幕端点测试""" + + def test_get_subtitles_empty(self, client): + c, _, _ = client + resp = c.get(BASE + "/clips/clip-001/subtitles") + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_get_subtitles_not_found_404(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.get_clip.return_value = None + resp = c.get(BASE + "/clips/nonexistent/subtitles") + assert resp.status_code == 404 + + def test_create_subtitle(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.update_clip.return_value = _make_mock_clip() + resp = c.post( + BASE + "/clips/clip-001/subtitles", + json={"start_time": 0, "end_time": 2, "text": "hello"}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["text"] == "hello" + mock_plan_svc.update_clip.assert_called_once() + + def test_delete_subtitle_not_found_404(self, client): + c, _, mock_plan_svc = client + mock_plan_svc.get_clip.return_value = None + resp = c.delete(BASE + "/clips/nonexistent/subtitles/sub-001") + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# 片段调整端点测试 +# ---------------------------------------------------------------------------