814 lines
29 KiB
Python
Executable File
814 lines
29 KiB
Python
Executable File
"""EditTemplateService — 模板管理业务逻辑.
|
||
|
||
封装 EditTemplate 和 TemplateClipConfig 的 CRUD 操作，
|
||
提供统一的业务接口供 API 路由层调用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
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_template import EditTemplate, EditTemplateStatus
|
||
from packages.domain.template_clip_config import (
|
||
ClipType,
|
||
TemplateClipConfig,
|
||
TransitionEffect,
|
||
)
|
||
from packages.domain.template_clip_converter import (
|
||
clip_configs_to_snapshots,
|
||
clips_to_template_clip_configs,
|
||
filter_plan_config_to_template,
|
||
snapshots_to_template_clip_configs,
|
||
validate_template_name,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class EditTemplateService:
|
||
"""模板管理服务
|
||
|
||
职责:
|
||
- 模板 CRUD（创建、查询、更新、软删除）
|
||
- 模板片段配置管理（增删改查）
|
||
- 业务校验（名称去重、状态合法性等）
|
||
"""
|
||
|
||
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)
|
||
from packages.adapters.sqlalchemy_impl.template_version_repository import (
|
||
SQLAlchemyTemplateVersionRepository,
|
||
)
|
||
|
||
self._version_repo = SQLAlchemyTemplateVersionRepository(db)
|
||
self._db = db
|
||
|
||
# ── 模板 CRUD ──────────────────────────────────────────────────────────
|
||
|
||
def list_templates(
|
||
self,
|
||
*,
|
||
template_type: Optional[str] = None,
|
||
status: Optional[EditTemplateStatus] = None,
|
||
active_only: bool = False,
|
||
skip: int = 0,
|
||
limit: int = 50,
|
||
) -> List[EditTemplate]:
|
||
"""列出模板
|
||
|
||
Args:
|
||
template_type: 按类型筛选
|
||
status: 按状态筛选
|
||
active_only: 仅返回激活模板
|
||
skip: 分页偏移
|
||
limit: 每页数量
|
||
"""
|
||
if active_only:
|
||
return self._template_repo.list_active(
|
||
template_type=template_type,
|
||
skip=skip,
|
||
limit=limit,
|
||
)
|
||
return self._template_repo.list_all(
|
||
template_type=template_type,
|
||
status=status,
|
||
skip=skip,
|
||
limit=limit,
|
||
)
|
||
|
||
def count_templates(
|
||
self,
|
||
*,
|
||
template_type: Optional[str] = None,
|
||
status: Optional[EditTemplateStatus] = None,
|
||
) -> int:
|
||
"""统计模板数量"""
|
||
return self._template_repo.count(
|
||
template_type=template_type,
|
||
status=status,
|
||
)
|
||
|
||
def get_template(self, template_id: str) -> Optional[EditTemplate]:
|
||
"""获取模板详情"""
|
||
return self._template_repo.get(template_id)
|
||
|
||
def get_template_or_raise(self, template_id: str) -> EditTemplate:
|
||
"""获取模板，不存在则抛出 ValueError"""
|
||
template = self._template_repo.get(template_id)
|
||
if template is None:
|
||
raise ValueError(f"模板不存在: {template_id}")
|
||
return template
|
||
|
||
def create_template(
|
||
self,
|
||
name: str,
|
||
*,
|
||
description: str = "",
|
||
template_type: str = "default",
|
||
editing_mode: str = "one_take",
|
||
config: Optional[dict[str, Any]] = None,
|
||
preview_url: str = "",
|
||
sort_weight: int = 0,
|
||
) -> EditTemplate:
|
||
"""创建模板
|
||
|
||
Raises:
|
||
ValueError: 名称为空或重复
|
||
"""
|
||
# 名称校验
|
||
clean_name = validate_template_name(name)
|
||
|
||
# 名称重复检查
|
||
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}")
|
||
|
||
template = EditTemplate.create(
|
||
name=clean_name,
|
||
description=description,
|
||
template_type=template_type,
|
||
editing_mode=editing_mode,
|
||
config=config,
|
||
preview_url=preview_url,
|
||
sort_weight=sort_weight,
|
||
)
|
||
created = self._template_repo.create(template)
|
||
logger.info("创建模板: id=%s name=%s", created.id, created.name)
|
||
return created
|
||
|
||
def update_template(
|
||
self,
|
||
template_id: str,
|
||
*,
|
||
name: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
template_type: Optional[str] = None,
|
||
editing_mode: Optional[str] = None,
|
||
config: Optional[dict[str, Any]] = None,
|
||
preview_url: Optional[str] = None,
|
||
sort_weight: Optional[int] = None,
|
||
status: Optional[EditTemplateStatus] = None,
|
||
) -> EditTemplate:
|
||
"""更新模板
|
||
|
||
Raises:
|
||
ValueError: 模板不存在或名称重复
|
||
"""
|
||
existing = self.get_template_or_raise(template_id)
|
||
|
||
# 名称重复检查（排除自身）
|
||
new_name = name.strip() if name is not None else existing.name
|
||
if name is not None and new_name != existing.name:
|
||
all_templates = self._template_repo.list_all(skip=0, limit=1000)
|
||
for t in all_templates:
|
||
if t.id != template_id and t.name == new_name and t.status == EditTemplateStatus.ACTIVE:
|
||
raise ValueError(f"模板名称已存在: {new_name}")
|
||
|
||
# 构建更新后的实体
|
||
updated = EditTemplate(
|
||
id=existing.id,
|
||
name=new_name,
|
||
description=description.strip() if description is not None else existing.description,
|
||
template_type=template_type.strip() if template_type is not None else existing.template_type,
|
||
editing_mode=editing_mode.strip() if editing_mode is not None else existing.editing_mode,
|
||
config=config if config is not None else existing.config,
|
||
preview_url=preview_url.strip() if preview_url is not None else existing.preview_url,
|
||
sort_weight=sort_weight if sort_weight is not None else existing.sort_weight,
|
||
status=status if status is not None else existing.status,
|
||
created_at=existing.created_at,
|
||
updated_at=existing.updated_at,
|
||
)
|
||
result = self._template_repo.update(updated)
|
||
logger.info("更新模板: id=%s", template_id)
|
||
return result
|
||
|
||
def deactivate_template(self, template_id: str) -> EditTemplate:
|
||
"""软删除模板（设为 inactive）
|
||
|
||
Raises:
|
||
ValueError: 模板不存在
|
||
"""
|
||
existing = self.get_template_or_raise(template_id)
|
||
existing.deactivate()
|
||
result = self._template_repo.update(existing)
|
||
logger.info("停用模板: id=%s", template_id)
|
||
return result
|
||
|
||
# ── 模板片段配置管理 ────────────────────────────────────────────────────
|
||
|
||
def list_clip_configs(
|
||
self,
|
||
template_id: str,
|
||
*,
|
||
clip_type: Optional[ClipType] = None,
|
||
skip: int = 0,
|
||
limit: int = 100,
|
||
) -> List[TemplateClipConfig]:
|
||
"""列出模板的片段配置"""
|
||
# 确保模板存在
|
||
self.get_template_or_raise(template_id)
|
||
return self._clip_config_repo.list_by_template(
|
||
template_id,
|
||
clip_type=clip_type,
|
||
skip=skip,
|
||
limit=limit,
|
||
)
|
||
|
||
def get_clip_config(self, config_id: str) -> Optional[TemplateClipConfig]:
|
||
"""获取片段配置详情"""
|
||
return self._clip_config_repo.get(config_id)
|
||
|
||
def get_clip_config_or_raise(self, config_id: str) -> TemplateClipConfig:
|
||
"""获取片段配置，不存在则抛出 ValueError"""
|
||
config = self._clip_config_repo.get(config_id)
|
||
if config is None:
|
||
raise ValueError(f"片段配置不存在: {config_id}")
|
||
return config
|
||
|
||
def create_clip_config(
|
||
self,
|
||
template_id: str,
|
||
clip_type: ClipType | str,
|
||
order: int,
|
||
*,
|
||
min_duration: float = 0.0,
|
||
max_duration: float = 0.0,
|
||
text_template: str = "",
|
||
material_requirements: Optional[dict[str, Any]] = None,
|
||
transition_effect: TransitionEffect | str = TransitionEffect.CUT,
|
||
config: Optional[dict[str, Any]] = None,
|
||
) -> TemplateClipConfig:
|
||
"""创建片段配置
|
||
|
||
Raises:
|
||
ValueError: 模板不存在或参数校验失败
|
||
"""
|
||
# 确保模板存在
|
||
self.get_template_or_raise(template_id)
|
||
|
||
clip_config = TemplateClipConfig.create(
|
||
template_id=template_id,
|
||
clip_type=clip_type,
|
||
order=order,
|
||
min_duration=min_duration,
|
||
max_duration=max_duration,
|
||
text_template=text_template,
|
||
material_requirements=material_requirements,
|
||
transition_effect=transition_effect,
|
||
config=config,
|
||
)
|
||
created = self._clip_config_repo.create(clip_config)
|
||
logger.info(
|
||
"创建片段配置: id=%s template_id=%s clip_type=%s order=%d",
|
||
created.id,
|
||
template_id,
|
||
created.clip_type,
|
||
created.order,
|
||
)
|
||
return created
|
||
|
||
def update_clip_config(
|
||
self,
|
||
config_id: str,
|
||
*,
|
||
clip_type: Optional[ClipType | str] = None,
|
||
order: Optional[int] = None,
|
||
min_duration: Optional[float] = None,
|
||
max_duration: Optional[float] = None,
|
||
text_template: Optional[str] = None,
|
||
material_requirements: Optional[dict[str, Any]] = None,
|
||
transition_effect: Optional[TransitionEffect | str] = None,
|
||
config: Optional[dict[str, Any]] = None,
|
||
) -> TemplateClipConfig:
|
||
"""更新片段配置
|
||
|
||
Raises:
|
||
ValueError: 配置不存在或参数校验失败
|
||
"""
|
||
existing = self.get_clip_config_or_raise(config_id)
|
||
|
||
# 解析枚举类型
|
||
new_clip_type = ClipType(clip_type) if clip_type is not None else existing.clip_type
|
||
new_transition = (
|
||
TransitionEffect(transition_effect) if transition_effect is not None else existing.transition_effect
|
||
)
|
||
|
||
updated = TemplateClipConfig(
|
||
id=existing.id,
|
||
template_id=existing.template_id,
|
||
clip_type=new_clip_type,
|
||
order=order if order is not None else existing.order,
|
||
min_duration=min_duration if min_duration is not None else existing.min_duration,
|
||
max_duration=max_duration if max_duration is not None else existing.max_duration,
|
||
text_template=text_template.strip() if text_template is not None else existing.text_template,
|
||
material_requirements=(
|
||
material_requirements if material_requirements is not None else existing.material_requirements
|
||
),
|
||
transition_effect=new_transition,
|
||
config=config if config is not None else existing.config,
|
||
created_at=existing.created_at,
|
||
updated_at=existing.updated_at,
|
||
)
|
||
result = self._clip_config_repo.update(updated)
|
||
logger.info("更新片段配置: id=%s", config_id)
|
||
return result
|
||
|
||
def delete_clip_config(self, config_id: str) -> bool:
|
||
"""删除片段配置
|
||
|
||
Returns:
|
||
bool: 是否删除成功
|
||
"""
|
||
deleted = self._clip_config_repo.delete(config_id)
|
||
if deleted:
|
||
logger.info("删除片段配置: id=%s", config_id)
|
||
return deleted
|
||
|
||
def reorder_clip_configs(
|
||
self,
|
||
template_id: str,
|
||
config_ids: List[str],
|
||
) -> List[TemplateClipConfig]:
|
||
"""重新排序片段配置
|
||
|
||
Args:
|
||
template_id: 模板 ID
|
||
config_ids: 按新顺序排列的配置 ID 列表
|
||
|
||
Returns:
|
||
更新后的配置列表
|
||
|
||
Raises:
|
||
ValueError: 模板不存在或配置 ID 不匹配
|
||
"""
|
||
# 确保模板存在
|
||
self.get_template_or_raise(template_id)
|
||
|
||
# 获取当前配置
|
||
current_configs = self._clip_config_repo.list_by_template(template_id)
|
||
current_ids = {c.id for c in current_configs}
|
||
|
||
# 校验 ID 列表
|
||
if set(config_ids) != current_ids:
|
||
raise ValueError("配置 ID 列表与模板下的配置不匹配")
|
||
|
||
# 更新 order
|
||
results = []
|
||
for new_order, config_id in enumerate(config_ids):
|
||
config = self._clip_config_repo.get(config_id)
|
||
if config is None:
|
||
continue
|
||
updated = TemplateClipConfig(
|
||
id=config.id,
|
||
template_id=config.template_id,
|
||
clip_type=config.clip_type,
|
||
order=new_order,
|
||
min_duration=config.min_duration,
|
||
max_duration=config.max_duration,
|
||
text_template=config.text_template,
|
||
material_requirements=config.material_requirements,
|
||
transition_effect=config.transition_effect,
|
||
config=config.config,
|
||
created_at=config.created_at,
|
||
updated_at=config.updated_at,
|
||
)
|
||
results.append(self._clip_config_repo.update(updated))
|
||
|
||
logger.info(
|
||
"重排序片段配置: template_id=%s count=%d",
|
||
template_id,
|
||
len(config_ids),
|
||
)
|
||
return results
|
||
|
||
def get_template_with_configs(
|
||
self,
|
||
template_id: str,
|
||
) -> dict:
|
||
"""获取模板及其所有片段配置
|
||
|
||
Returns:
|
||
dict: {"template": EditTemplate, "clip_configs": List[TemplateClipConfig]}
|
||
"""
|
||
template = self.get_template_or_raise(template_id)
|
||
clip_configs = self._clip_config_repo.list_by_template(template_id)
|
||
return {
|
||
"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 中提取模板级配置，去掉运行时/素材相关字段
|
||
template_config = filter_plan_config_to_template(plan.config)
|
||
|
||
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_config_obj in clips_to_template_clip_configs(created_template.id, clips):
|
||
created = self._clip_config_repo.create(clip_config_obj)
|
||
created_configs.append(created)
|
||
|
||
return {
|
||
"template": created_template,
|
||
"clip_configs": created_configs,
|
||
}
|
||
|
||
# ── 模板草稿（编辑器）相关 ──────────────────────────────────────────────────
|
||
|
||
def get_template_draft(self, template_id: str) -> Optional[Any]:
|
||
"""获取模板的草稿剪辑计划
|
||
|
||
通过 template_id + config.is_template_draft=True 标记查找。
|
||
每个模板有且仅有一个草稿计划。
|
||
|
||
Args:
|
||
template_id: 模板 ID
|
||
|
||
Returns:
|
||
EditPlan | None: 草稿剪辑计划，不存在则返回 None
|
||
"""
|
||
from packages.domain.edit_plan import EditPlan # noqa: F401
|
||
|
||
plans = self._plan_repo.list_by_template(template_id, limit=50)
|
||
for plan in plans:
|
||
config = plan.config or {}
|
||
if config.get("is_template_draft") is True:
|
||
return plan
|
||
return None
|
||
|
||
def create_template_draft(
|
||
self,
|
||
template_id: str,
|
||
user_id: str,
|
||
*,
|
||
project_id: str = "",
|
||
) -> Any:
|
||
"""基于模板创建草稿剪辑计划
|
||
|
||
草稿与普通剪辑计划的区别：
|
||
- config.is_template_draft = True
|
||
- 不绑定具体素材（空素材列表）
|
||
- 用于模板编辑器的编辑上下文
|
||
|
||
Args:
|
||
template_id: 模板 ID
|
||
user_id: 创建者用户 ID
|
||
project_id: 所属项目 ID（可选）
|
||
|
||
Returns:
|
||
EditPlan: 创建的草稿剪辑计划
|
||
|
||
Raises:
|
||
ValueError: 模板不存在，或草稿已存在
|
||
"""
|
||
from .plan_generator_service import PlanGeneratorService
|
||
|
||
# 检查模板是否存在
|
||
template = self.get_template_or_raise(template_id)
|
||
|
||
# 检查草稿是否已存在
|
||
existing = self.get_template_draft(template_id)
|
||
if existing is not None:
|
||
raise ValueError(f"模板草稿已存在: {template_id}")
|
||
|
||
# 读取模板片段配置
|
||
clip_configs = self.list_clip_configs(template_id)
|
||
|
||
# 基于模板生成计划（空素材）
|
||
generator = PlanGeneratorService(self._db)
|
||
result = generator.generate_from_template(
|
||
template=template,
|
||
clip_configs=clip_configs,
|
||
asset_ids=[],
|
||
project_id=project_id,
|
||
created_by_user_id=user_id,
|
||
name=f"{template.name} - 草稿",
|
||
)
|
||
plan = result["plan"]
|
||
|
||
# 标记为模板草稿
|
||
plan_config = plan.config or {}
|
||
plan_config["is_template_draft"] = True
|
||
plan.config = plan_config
|
||
plan = self._plan_repo.update(plan)
|
||
|
||
logger.info(
|
||
"创建模板草稿: template_id=%s draft_plan_id=%s user_id=%s",
|
||
template_id,
|
||
plan.id,
|
||
user_id,
|
||
)
|
||
return plan
|
||
|
||
def get_or_create_draft(
|
||
self,
|
||
template_id: str,
|
||
user_id: str,
|
||
*,
|
||
project_id: str = "",
|
||
) -> Any:
|
||
"""获取或创建模板草稿
|
||
|
||
首次访问模板编辑器时自动创建草稿。
|
||
|
||
Args:
|
||
template_id: 模板 ID
|
||
user_id: 操作用户 ID
|
||
project_id: 所属项目 ID（可选）
|
||
|
||
Returns:
|
||
EditPlan: 草稿剪辑计划
|
||
"""
|
||
draft = self.get_template_draft(template_id)
|
||
if draft is not None:
|
||
return draft
|
||
return self.create_template_draft(template_id, user_id, project_id=project_id)
|
||
|
||
def publish_template_from_draft(
|
||
self,
|
||
template_id: str,
|
||
draft_plan_id: str,
|
||
*,
|
||
change_note: str = "",
|
||
published_by: str = "",
|
||
) -> Any:
|
||
"""将草稿剪辑计划的内容发布（同步）到模板
|
||
|
||
将草稿的配置和片段结构同步到模板，相当于"保存"编辑结果。
|
||
使用事务保证一致性，失败则回滚。
|
||
|
||
同步规则:
|
||
- 草稿 plan.config → template.config（过滤掉草稿特有字段）
|
||
- 草稿 clips → template_clip_configs（先删后插）
|
||
- 草稿 editing_mode → template.editing_mode
|
||
- 不更新模板名称、描述等元信息（由专门的接口处理）
|
||
|
||
Args:
|
||
template_id: 模板 ID
|
||
draft_plan_id: 草稿剪辑计划 ID
|
||
|
||
Returns:
|
||
EditTemplate: 更新后的模板
|
||
|
||
Raises:
|
||
ValueError: 模板/草稿不存在，或草稿不属于该模板
|
||
"""
|
||
# 1. 校验模板和草稿
|
||
template = self.get_template_or_raise(template_id)
|
||
draft = self._plan_repo.get(draft_plan_id)
|
||
if draft is None:
|
||
raise ValueError(f"草稿计划不存在: {draft_plan_id}")
|
||
if draft.template_id != template_id:
|
||
raise ValueError(f"草稿不属于该模板: plan_template_id={draft.template_id}")
|
||
config = draft.config or {}
|
||
if config.get("is_template_draft") is not True:
|
||
raise ValueError("指定的计划不是模板草稿")
|
||
|
||
# 2. 读取草稿片段
|
||
draft_clips = self._plan_clip_repo.list_by_plan(draft_plan_id)
|
||
draft_clips.sort(key=lambda c: c.order)
|
||
|
||
# 3. 提取 editing_mode
|
||
editing_mode = config.get("editing_mode", "one_take")
|
||
|
||
# 4. 提取模板配置（去掉草稿/运行时字段）
|
||
template_config = filter_plan_config_to_template(draft.config)
|
||
|
||
# 5. 事务更新
|
||
try:
|
||
# 5.0 先保存旧版快照（发布前的状态），用于回滚
|
||
old_version = template.version or 1
|
||
old_clip_configs = self._clip_config_repo.list_by_template(template_id)
|
||
old_clip_snapshots = clip_configs_to_snapshots(old_clip_configs)
|
||
|
||
from packages.domain.template_version import EditTemplateVersion
|
||
|
||
old_snapshot = EditTemplateVersion.create(
|
||
template_id=template_id,
|
||
version=old_version,
|
||
name=template.name,
|
||
editing_mode=template.editing_mode,
|
||
config=dict(template.config) if template.config else {},
|
||
clip_configs=old_clip_snapshots,
|
||
change_note=f"v{old_version} 快照（发布前）",
|
||
published_by=published_by,
|
||
)
|
||
self._version_repo.create(old_snapshot)
|
||
|
||
# 更新模板元信息
|
||
template.config = template_config
|
||
template.editing_mode = editing_mode
|
||
template.bump_version() # 版本号 +1
|
||
updated_template = self._template_repo.update(template)
|
||
|
||
# 批量删除旧的片段配置（外层事务统一提交）
|
||
self._clip_config_repo.delete_by_template(template_id, commit=False)
|
||
|
||
# 创建新的片段配置
|
||
created_configs: list[TemplateClipConfig] = []
|
||
for config_obj in clips_to_template_clip_configs(template_id, draft_clips):
|
||
created = self._clip_config_repo.create(config_obj)
|
||
created_configs.append(created)
|
||
|
||
self._db.commit()
|
||
logger.info(
|
||
"发布模板草稿: template_id=%s draft_plan_id=%s clip_count=%d",
|
||
template_id,
|
||
draft_plan_id,
|
||
len(created_configs),
|
||
)
|
||
return updated_template
|
||
|
||
except Exception as exc:
|
||
self._db.rollback()
|
||
logger.error(
|
||
"发布模板草稿失败: template_id=%s draft_plan_id=%s error=%s",
|
||
template_id,
|
||
draft_plan_id,
|
||
exc,
|
||
)
|
||
raise
|
||
|
||
# ── 版本历史与回滚 ────────────────────────────────────────────────────
|
||
|
||
def list_template_versions(self, template_id: str, limit: int = 50) -> list[Any]:
|
||
"""列出模板的发布版本历史（按版本号倒序）"""
|
||
self.get_template_or_raise(template_id) # 校验存在性
|
||
return self._version_repo.list_by_template(template_id, limit=limit)
|
||
|
||
def rollback_to_version(self, template_id: str, version: int) -> Any:
|
||
"""回滚模板到指定历史版本
|
||
|
||
流程：
|
||
1. 校验目标版本存在
|
||
2. 保存当前状态为新版本快照（当前版本号）
|
||
3. 用目标版本的快照覆盖模板 config + clip_configs
|
||
4. 版本号 +1（回滚本身也是一次发布）
|
||
|
||
Returns:
|
||
EditTemplate: 回滚后的模板
|
||
|
||
Raises:
|
||
ValueError: 模板/版本不存在
|
||
"""
|
||
template = self.get_template_or_raise(template_id)
|
||
|
||
# 1. 读取目标版本快照
|
||
target_version = self._version_repo.get_by_version(template_id, version)
|
||
if target_version is None:
|
||
raise ValueError(f"模板 {template_id} 不存在版本 {version}")
|
||
|
||
current_version = template.version or 1
|
||
|
||
try:
|
||
# 2. 先保存当前状态快照（当前版本号），确保回滚可撤销
|
||
old_clip_configs = self._clip_config_repo.list_by_template(template_id)
|
||
old_clip_snapshots = clip_configs_to_snapshots(old_clip_configs)
|
||
|
||
from packages.domain.template_version import EditTemplateVersion
|
||
|
||
current_snapshot = EditTemplateVersion.create(
|
||
template_id=template_id,
|
||
version=current_version,
|
||
name=template.name,
|
||
editing_mode=template.editing_mode,
|
||
config=dict(template.config) if template.config else {},
|
||
clip_configs=old_clip_snapshots,
|
||
change_note=f"v{current_version} 快照（回滚到 v{version} 前）",
|
||
published_by="rollback",
|
||
)
|
||
self._version_repo.create(current_snapshot)
|
||
|
||
# 3. 覆盖模板配置 + editing_mode + name + preview_url
|
||
template.config = dict(target_version.config)
|
||
template.editing_mode = target_version.editing_mode
|
||
if target_version.name:
|
||
template.name = target_version.name
|
||
template.bump_version() # 版本号 +1
|
||
updated_template = self._template_repo.update(template)
|
||
|
||
# 4. 先删后插 clip_configs（批量删除避免N+1）
|
||
from packages.adapters.sqlalchemy_impl.models import (
|
||
TemplateClipConfigModel,
|
||
)
|
||
|
||
self._db.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == template_id).delete(
|
||
synchronize_session=False
|
||
)
|
||
|
||
for config_obj in snapshots_to_template_clip_configs(template_id, target_version.clip_configs):
|
||
self._clip_config_repo.create(config_obj)
|
||
|
||
self._db.commit()
|
||
logger.info(
|
||
"模板回滚成功: template_id=%s from_v=%d to_v=%d new_v=%d",
|
||
template_id,
|
||
current_version,
|
||
version,
|
||
updated_template.version,
|
||
)
|
||
return updated_template
|
||
|
||
except Exception as exc:
|
||
self._db.rollback()
|
||
logger.error(
|
||
"模板回滚失败: template_id=%s target_version=%d error=%s",
|
||
template_id,
|
||
version,
|
||
exc,
|
||
)
|
||
raise
|