diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 4285125ab..cdadd7a33 100755 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -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 (拆分模块) ──────────────────────────────────────────── diff --git a/apps/api/app/services/edit_template_service.py b/apps/api/app/services/edit_template_service.py old mode 100644 new mode 100755 index e5bbc23ce..7ee940258 --- a/apps/api/app/services/edit_template_service.py +++ b/apps/api/app/services/edit_template_service.py @@ -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, + } diff --git a/tests/unit/test_edit_template_service.py b/tests/unit/test_edit_template_service.py old mode 100644 new mode 100755 index b2cbbbd0c..41eb14f90 --- a/tests/unit/test_edit_template_service.py +++ b/tests/unit/test_edit_template_service.py @@ -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