"""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 ( SQLAlchemyEditTemplateRepository, SQLAlchemyTemplateClipConfigRepository, ) from packages.domain.edit_template import EditTemplate, EditTemplateStatus from packages.domain.template_clip_config import ( ClipType, TemplateClipConfig, TransitionEffect, ) logger = logging.getLogger(__name__) class EditTemplateService: """模板管理服务 职责: - 模板 CRUD(创建、查询、更新、软删除) - 模板片段配置管理(增删改查) - 业务校验(名称去重、状态合法性等) """ def __init__(self, db: Session) -> None: self._template_repo = SQLAlchemyEditTemplateRepository(db) self._clip_config_repo = SQLAlchemyTemplateClipConfigRepository(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 = 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}") 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, }