Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c6bdd91a61 |
@@ -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 (拆分模块) ────────────────────────────────────────────
|
||||
|
||||
Regular → Executable
+141
@@ -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,
|
||||
}
|
||||
|
||||
Regular → Executable
+288
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user