Compare commits

...

1 Commits

Author SHA1 Message Date
CI Bot c6bdd91a61 feat: 剪辑计划保存为模板 - API + Service + 11个单测
CI Build & Deploy Pipeline / Build Staging Web Image (push) Has been skipped
CI Build & Deploy Pipeline / Build Staging API Image (push) Has been skipped
CI Build & Deploy Pipeline / Build Staging Worker Image (push) Has been skipped
CI Build & Deploy Pipeline / Build Staging API Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been skipped
CI Build & Deploy Pipeline / Build Production API Image (push) Has been skipped
CI Build & Deploy Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production Web Image (push) Has been skipped
CI Build & Deploy Pipeline / Build Production Worker Image (push) Has been skipped
CI Build & Deploy Pipeline / Staging E2E Tests (push) Has been skipped
CI Build & Deploy Pipeline / Staging API Integration Tests (push) Has been skipped
CI Build & Deploy Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI Build & Deploy Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Production (push) Has been skipped
CI Build & Deploy Pipeline / Build Production API Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production Web Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Production Browser E2E (push) Has been skipped
CI Build & Deploy Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Production (pull_request) Has been skipped
CI Build & Deploy Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 1m4s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 1m8s
CI/CD Pipeline / Frontend Lint (push) Successful in 1m11s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m10s
CI/CD Pipeline / Unit Tests (push) Successful in 3m10s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m42s
CI/CD Pipeline / Integration Tests (push) Successful in 2m43s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 3m54s
- EditTemplateService.save_plan_as_template(): 计划→模板转换
  - 片段配置完整迁移(类型/时长/转场/文案/速度)
  - 自动剥离素材绑定和运行时字段
  - 未知枚举值安全回退
- POST /api/v1/edit-plans/{plan_id}/save-as-template
  - 项目级权限校验
  - 名称去重校验
- 11个单元测试全覆盖
2026-07-16 13:44:02 +08:00
3 changed files with 508 additions and 1 deletions
+79 -1
View File
@@ -24,7 +24,7 @@ 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.schemas.generation_task import GenerationTaskResponse
from app.services import EditPlanService
from app.services import EditPlanService, EditTemplateService
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
@@ -506,6 +506,84 @@ def copy_plan(
current_user.user.id,
)
return _to_response(new_plan)
# ── 保存为模板 ────────────────────────────────────────────────────────────────
class SaveAsTemplateRequest(BaseModel):
"""保存为模板请求体"""
name: str = Field(..., min_length=1, max_length=200, description="模板名称")
description: str = Field(default="", max_length=500, description="模板描述")
template_type: str = Field(default="custom", max_length=50, description="模板类型")
preview_url: str = Field(default="", max_length=500, description="预览图 URL")
@router.post(
"/{plan_id}/save-as-template",
response_model=dict[str, Any],
summary="将剪辑计划保存为模板",
status_code=status.HTTP_201_CREATED,
)
def save_plan_as_template(
plan_id: str,
body: SaveAsTemplateRequest,
current_user: AuthenticatedUser = Depends(get_current_user),
db: Session = Depends(get_db_session),
project_repo=Depends(get_project_repository),
) -> dict[str, Any]:
"""将指定剪辑计划的配置和片段结构保存为一个新模板。
新模板会复制计划的所有片段配置(类型、时长、转场、文案等),
但不绑定具体素材,可重复用于创建新的剪辑计划。
"""
# 校验计划存在性和项目权限
plan_service = EditPlanService(db)
plan = plan_service.get_plan(plan_id)
if plan is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"剪辑计划不存在: {plan_id}",
)
if plan.project_id:
check_project_access(project_repo, current_user, plan.project_id)
template_service = EditTemplateService(db)
try:
result = template_service.save_plan_as_template(
plan_id=plan_id,
name=body.name,
description=body.description,
template_type=body.template_type,
preview_url=body.preview_url,
)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e),
) from e
template = result["template"]
clip_configs = result["clip_configs"]
logger.info(
"保存计划为模板: plan_id=%s template_id=%s name=%s by user=%s",
plan_id,
template.id,
body.name,
current_user.user.id,
)
return {
"id": template.id,
"name": template.name,
"description": template.description,
"template_type": template.template_type,
"editing_mode": template.editing_mode,
"preview_url": template.preview_url,
"status": template.status.value,
"clip_count": len(clip_configs),
"created_at": template.created_at.isoformat(),
}
# ── Include sub-routers (拆分模块) ────────────────────────────────────────────
+141
View File
@@ -12,9 +12,13 @@ from typing import Any, List, Optional
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl import (
SQLAlchemyEditPlanClipRepository,
SQLAlchemyEditPlanRepository,
SQLAlchemyEditTemplateRepository,
SQLAlchemyTemplateClipConfigRepository,
)
from packages.domain.edit_plan import EditPlan
from packages.domain.edit_plan_clip import EditPlanClip
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.template_clip_config import (
ClipType,
@@ -37,6 +41,9 @@ class EditTemplateService:
def __init__(self, db: Session) -> None:
self._template_repo = SQLAlchemyEditTemplateRepository(db)
self._clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db)
self._plan_repo = SQLAlchemyEditPlanRepository(db)
self._plan_clip_repo = SQLAlchemyEditPlanClipRepository(db)
self._db = db
# ── 模板 CRUD ──────────────────────────────────────────────────────────
@@ -394,3 +401,137 @@ class EditTemplateService:
"template": template,
"clip_configs": clip_configs,
}
# ── 从剪辑计划保存为模板 ──────────────────────────────────────────────
def save_plan_as_template(
self,
plan_id: str,
name: str,
*,
description: str = "",
template_type: str = "custom",
preview_url: str = "",
) -> dict[str, Any]:
"""将剪辑计划保存为模板
将指定剪辑计划的配置和片段结构另存为一个新模板,
方便后续基于该模板快速创建新的剪辑计划。
转换规则:
- 计划名称 → 模板名称(调用方传入,支持自定义)
- 计划 config → 模板 config(整体迁移)
- 计划 editing_mode 从 config 中提取,默认 one_take
- 每个片段转换为模板片段配置:
- clip_type 直接映射
- order 保持不变
- duration → min_duration = max_duration = duration(固定时长)
- text_content → text_template
- transition_effect 直接映射
- playback_speed 等播放参数存入 config
- 不保留 asset_id(模板不绑定具体素材)
Args:
plan_id: 源剪辑计划 ID
name: 新模板名称
description: 模板描述
template_type: 模板类型,默认 custom(用户自定义)
preview_url: 预览图 URL
Returns:
dict: {"template": EditTemplate, "clip_configs": List[TemplateClipConfig]}
Raises:
ValueError: 计划不存在或名称为空/重复
"""
# 1. 读取源计划
plan = self._plan_repo.get(plan_id)
if plan is None:
raise ValueError(f"剪辑计划不存在: {plan_id}")
# 2. 读取所有片段(按 order 排序)
clips = self._plan_clip_repo.list_by_plan(plan_id)
clips.sort(key=lambda c: c.order)
# 3. 提取 editing_mode
editing_mode = plan.config.get("editing_mode", "one_take") if plan.config else "one_take"
# 4. 创建模板(复用 create_template 的校验逻辑,但手动构建避免重复查询)
clean_name = name.strip()
if not clean_name:
raise ValueError("模板名称不能为空")
# 名称重复检查
existing = self._template_repo.list_all(skip=0, limit=1000)
for t in existing:
if t.name == clean_name and t.status == EditTemplateStatus.ACTIVE:
raise ValueError(f"模板名称已存在: {clean_name}")
# 从计划 config 中提取模板级配置,去掉运行时/素材相关字段
plan_config = plan.config or {}
template_config: dict[str, Any] = {}
for key, value in plan_config.items():
# 跳过明显的运行时/实例字段,保留风格/模式类配置
if key not in {"asset_ids", "source_edit_plan_id", "generation_task_id"}:
template_config[key] = value
template = EditTemplate.create(
name=clean_name,
description=description,
template_type=template_type,
editing_mode=editing_mode,
config=template_config,
preview_url=preview_url,
)
created_template = self._template_repo.create(template)
logger.info(
"从剪辑计划创建模板: plan_id=%s template_id=%s name=%s clip_count=%d",
plan_id,
created_template.id,
clean_name,
len(clips),
)
# 5. 转换每个片段为模板片段配置
created_configs: List[TemplateClipConfig] = []
for clip in clips:
clip_config: dict[str, Any] = {}
# 播放速度存入 config
if clip.playback_speed and clip.playback_speed != 1.0:
clip_config["playback_speed"] = clip.playback_speed
# 片段自有 config 合并(优先级:clip.config 覆盖上面的)
if clip.config:
clip_config.update(clip.config)
# 去掉素材相关字段
clip_config.pop("asset_info", None)
clip_config.pop("source_asset_id", None)
# 转场效果兼容校验
try:
transition = TransitionEffect(clip.transition_effect)
except ValueError:
transition = TransitionEffect.CUT
# 片段类型兼容校验
try:
clip_type = ClipType(clip.clip_type)
except ValueError:
clip_type = ClipType.MAIN
clip_config_obj = TemplateClipConfig.create(
template_id=created_template.id,
clip_type=clip_type,
order=clip.order,
min_duration=clip.duration,
max_duration=clip.duration,
text_template=clip.text_content or "",
transition_effect=transition,
config=clip_config,
)
created = self._clip_config_repo.create(clip_config_obj)
created_configs.append(created)
return {
"template": created_template,
"clip_configs": created_configs,
}
+288
View File
@@ -463,3 +463,291 @@ class TestCompositeQueries:
t = svc.create_template(name="空模板")
result = svc.get_template_with_configs(t.id)
assert len(result["clip_configs"]) == 0
# ===========================================================================
# Stub Repositories for EditPlan (save_as_template 测试用)
# ===========================================================================
class StubEditPlanRepository:
"""内存中的 EditPlan 仓储 stub"""
def __init__(self) -> None:
self._plans: dict[str, EditPlan] = {}
self._counter = 0
def _next_id(self) -> str:
self._counter += 1
return f"plan-{self._counter:03d}"
def get(self, plan_id: str) -> Optional[EditPlan]:
return self._plans.get(plan_id)
def create(self, plan: EditPlan) -> EditPlan:
if not plan.id:
plan.id = self._next_id()
self._plans[plan.id] = plan
return plan
class StubEditPlanClipRepository:
"""内存中的 EditPlanClip 仓储 stub"""
def __init__(self) -> None:
self._clips: dict[str, EditPlanClip] = {}
self._counter = 0
def _next_id(self) -> str:
self._counter += 1
return f"clip-{self._counter:03d}"
def list_by_plan(
self,
plan_id: str,
*,
status: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[EditPlanClip]:
items = [c for c in self._clips.values() if c.plan_id == plan_id]
if status:
items = [c for c in items if c.status.value == status]
items.sort(key=lambda c: c.order)
return items[skip : skip + limit]
def create(self, clip: EditPlanClip) -> EditPlanClip:
if not clip.id:
clip.id = self._next_id()
self._clips[clip.id] = clip
return clip
def _make_service_with_plan_stubs():
"""创建使用 stub 仓储的 EditTemplateService(含 plan 相关 stub"""
from app.services.edit_template_service import EditTemplateService
db = MagicMock()
svc = EditTemplateService(db)
svc._template_repo = StubEditTemplateRepository()
svc._clip_config_repo = StubTemplateClipConfigRepository()
svc._plan_repo = StubEditPlanRepository()
svc._plan_clip_repo = StubEditPlanClipRepository()
return svc
def _make_test_plan_with_clips(svc, *, clip_count: int = 3, plan_config=None):
"""辅助方法:创建一个带片段的测试计划,返回 plan 对象"""
from packages.domain.edit_plan import EditPlan
from packages.domain.edit_plan_clip import EditPlanClip
plan = EditPlan.create(
template_id="tpl-source",
name="我的剪辑计划",
config=plan_config or {"editing_mode": "one_take", "theme": "minimal"},
project_id="proj-001",
created_by_user_id="user-001",
)
plan.id = "plan-test-001"
svc._plan_repo.create(plan)
for i in range(clip_count):
clip = EditPlanClip.create(
plan_id=plan.id,
clip_type=ClipType.MAIN.value,
order=i,
asset_id=f"asset-{i:03d}",
text_content=f"片段{i}的文案",
duration=10.0 + i * 5,
transition_effect="cut" if i == 0 else "fade",
playback_speed=1.0 if i == 0 else 1.5,
config={"filter": "vivid"} if i == 1 else {},
)
svc._plan_clip_repo.create(clip)
return plan
# ===========================================================================
# 保存为模板测试
# ===========================================================================
class TestSavePlanAsTemplate:
"""从剪辑计划保存为模板测试"""
def test_basic_save_as_template(self):
"""基础场景:将有3个片段的计划保存为模板"""
svc = _make_service_with_plan_stubs()
plan = _make_test_plan_with_clips(svc, clip_count=3)
result = svc.save_plan_as_template(plan.id, name="我的自定义模板")
assert result["template"].name == "我的自定义模板"
assert result["template"].template_type == "custom"
assert result["template"].editing_mode == "one_take"
assert result["template"].status == EditTemplateStatus.ACTIVE
assert len(result["clip_configs"]) == 3
def test_clip_configs_correctly_converted(self):
"""片段正确转换为模板片段配置"""
svc = _make_service_with_plan_stubs()
plan = _make_test_plan_with_clips(svc, clip_count=2)
result = svc.save_plan_as_template(plan.id, name="转换测试模板")
configs = result["clip_configs"]
configs.sort(key=lambda c: c.order)
# 第0个片段
assert configs[0].clip_type == ClipType.MAIN
assert configs[0].order == 0
assert configs[0].min_duration == 10.0
assert configs[0].max_duration == 10.0
assert configs[0].text_template == "片段0的文案"
assert configs[0].transition_effect.value == "cut"
# playback_speed=1.0 不存
assert "playback_speed" not in configs[0].config
# 第1个片段
assert configs[1].order == 1
assert configs[1].min_duration == 15.0
assert configs[1].max_duration == 15.0
assert configs[1].transition_effect.value == "fade"
# playback_speed=1.5 存入config
assert configs[1].config.get("playback_speed") == 1.5
# config 中的 filter 保留
assert configs[1].config.get("filter") == "vivid"
def test_no_asset_id_in_template(self):
"""模板不保留具体素材ID"""
svc = _make_service_with_plan_stubs()
plan = _make_test_plan_with_clips(svc, clip_count=2)
result = svc.save_plan_as_template(plan.id, name="素材剥离测试")
for cfg in result["clip_configs"]:
# 模板片段配置没有 asset_id 字段
assert not hasattr(cfg, "asset_id") or not getattr(cfg, "asset_id", "")
# config 中也不应有素材相关字段
assert "asset_info" not in cfg.config
assert "source_asset_id" not in cfg.config
def test_template_config_stripped_of_runtime_fields(self):
"""模板config剥离运行时字段"""
svc = _make_service_with_plan_stubs()
plan_config = {
"editing_mode": "one_take",
"theme": "cinematic",
"asset_ids": ["a1", "a2"],
"source_edit_plan_id": "old-plan",
"generation_task_id": "task-123",
}
plan = _make_test_plan_with_clips(svc, clip_count=1, plan_config=plan_config)
result = svc.save_plan_as_template(plan.id, name="配置剥离测试")
tpl_config = result["template"].config
assert tpl_config.get("theme") == "cinematic"
assert "asset_ids" not in tpl_config
assert "source_edit_plan_id" not in tpl_config
assert "generation_task_id" not in tpl_config
def test_plan_not_found_raises_error(self):
"""计划不存在时报错"""
svc = _make_service_with_plan_stubs()
with pytest.raises(ValueError, match="剪辑计划不存在"):
svc.save_plan_as_template("nonexistent-plan", name="不存在的计划")
def test_empty_name_raises_error(self):
"""模板名称为空时报错"""
svc = _make_service_with_plan_stubs()
plan = _make_test_plan_with_clips(svc, clip_count=1)
with pytest.raises(ValueError, match="模板名称不能为空"):
svc.save_plan_as_template(plan.id, name=" ")
def test_duplicate_name_raises_error(self):
"""模板名称重复时报错"""
svc = _make_service_with_plan_stubs()
svc.create_template(name="重名模板")
plan = _make_test_plan_with_clips(svc, clip_count=1)
with pytest.raises(ValueError, match="模板名称已存在"):
svc.save_plan_as_template(plan.id, name="重名模板")
def test_save_zero_clip_plan(self):
"""零片段计划也能保存为模板"""
svc = _make_service_with_plan_stubs()
from packages.domain.edit_plan import EditPlan
plan = EditPlan.create(
template_id="tpl-source",
name="空计划",
config={"editing_mode": "one_take"},
)
plan.id = "plan-empty"
svc._plan_repo.create(plan)
result = svc.save_plan_as_template(plan.id, name="空模板")
assert result["template"].name == "空模板"
assert len(result["clip_configs"]) == 0
def test_custom_description_and_type(self):
"""自定义描述和模板类型"""
svc = _make_service_with_plan_stubs()
plan = _make_test_plan_with_clips(svc, clip_count=1)
result = svc.save_plan_as_template(
plan.id,
name="自定义模板",
description="这是一个测试模板",
template_type="vlog",
)
assert result["template"].description == "这是一个测试模板"
assert result["template"].template_type == "vlog"
def test_unknown_transition_effect_falls_back_to_cut(self):
"""未知转场效果回退到cut"""
svc = _make_service_with_plan_stubs()
from packages.domain.edit_plan import EditPlan
from packages.domain.edit_plan_clip import EditPlanClip
plan = EditPlan.create(template_id="tpl-src", name="转场测试计划")
plan.id = "plan-transition-test"
svc._plan_repo.create(plan)
clip = EditPlanClip.create(
plan_id=plan.id,
clip_type="main",
order=0,
duration=10.0,
transition_effect="weird_effect_that_does_not_exist",
)
svc._plan_clip_repo.create(clip)
result = svc.save_plan_as_template(plan.id, name="转场兼容模板")
assert result["clip_configs"][0].transition_effect.value == "cut"
def test_unknown_clip_type_falls_back_to_main(self):
"""未知片段类型回退到main"""
svc = _make_service_with_plan_stubs()
from packages.domain.edit_plan import EditPlan
from packages.domain.edit_plan_clip import EditPlanClip
plan = EditPlan.create(template_id="tpl-src", name="类型测试计划")
plan.id = "plan-type-test"
svc._plan_repo.create(plan)
clip = EditPlanClip.create(
plan_id=plan.id,
clip_type="unknown_clip_type",
order=0,
duration=10.0,
)
svc._plan_clip_repo.create(clip)
result = svc.save_plan_as_template(plan.id, name="类型兼容模板")
assert result["clip_configs"][0].clip_type == ClipType.MAIN