"""模板片段转换器 — 纯函数集合. 从 edit_template_service.py 抽离的纯逻辑,负责在不同数据形态间转换: - 剪辑计划片段 (EditPlanClip) → 模板片段配置 (TemplateClipConfig) - 模板片段配置 → 版本快照 dict - 版本快照 dict → 模板片段配置 - 计划 config → 模板 config(过滤运行时字段) 所有函数均为纯函数,不依赖数据库或外部 IO。 """ from __future__ import annotations from typing import Any from packages.domain.template_clip_config import ( ClipType, TemplateClipConfig, TransitionEffect, ) # ── 安全枚举解析 ──────────────────────────────────────────────────────────── def safe_parse_transition_effect(value: Any, default: TransitionEffect = TransitionEffect.CUT) -> TransitionEffect: """安全解析转场效果枚举,解析失败返回默认值。 Args: value: 待解析的值(枚举、字符串或其他) default: 解析失败时的默认值 Returns: TransitionEffect 枚举值 """ if isinstance(value, TransitionEffect): return value try: return TransitionEffect(value) except (ValueError, TypeError): return default def safe_parse_clip_type(value: Any, default: ClipType = ClipType.MAIN) -> ClipType: """安全解析片段类型枚举,解析失败返回默认值。 Args: value: 待解析的值(枚举、字符串或其他) default: 解析失败时的默认值 Returns: ClipType 枚举值 """ if isinstance(value, ClipType): return value try: return ClipType(value) except (ValueError, TypeError): return default # ── Config 字段过滤 ───────────────────────────────────────────────────────── # 默认需要从 clip config 中移除的素材/运行时字段 _DEFAULT_CLIP_CONFIG_SKIP_KEYS = frozenset( { "asset_info", "source_asset_id", } ) # 默认需要从 plan config 中移除的运行时/实例字段 _DEFAULT_PLAN_CONFIG_SKIP_KEYS = frozenset( { "is_template_draft", "asset_ids", "source_edit_plan_id", "generation_task_id", } ) def filter_clip_config( clip_config: dict[str, Any] | None, playback_speed: float | None = None, skip_keys: frozenset[str] | None = None, ) -> dict[str, Any]: """构建模板片段的 config 字典。 处理逻辑: 1. 如果 playback_speed 存在且不等于 1.0,加入 config 2. 合并 clip 自身的 config 3. 移除素材相关字段 Args: clip_config: 原始片段 config(可为 None) playback_speed: 播放速度(可选,1.0 时不写入) skip_keys: 需要跳过的字段集合(None 时用默认) Returns: 过滤后的 config 字典 """ skip = skip_keys if skip_keys is not None else _DEFAULT_CLIP_CONFIG_SKIP_KEYS result: dict[str, Any] = {} if playback_speed is not None and playback_speed != 1.0: result["playback_speed"] = playback_speed if clip_config: result.update(clip_config) for key in skip: result.pop(key, None) return result def filter_plan_config_to_template( plan_config: dict[str, Any] | None, skip_keys: frozenset[str] | None = None, ) -> dict[str, Any]: """从计划 config 中提取模板 config(过滤运行时/实例字段)。 Args: plan_config: 原始计划 config(可为 None) skip_keys: 需要跳过的字段集合(None 时用默认) Returns: 过滤后的模板 config """ skip = skip_keys if skip_keys is not None else _DEFAULT_PLAN_CONFIG_SKIP_KEYS if not plan_config: return {} return {k: v for k, v in plan_config.items() if k not in skip} # ── Clip → TemplateClipConfig 转换 ──────────────────────────────────────── def clip_to_template_clip_config( template_id: str, clip: Any, ) -> TemplateClipConfig: """将剪辑计划片段转换为模板片段配置。 转换规则: - clip_type → 安全解析后映射 - order → 保持不变 - duration → min_duration = max_duration = duration(固定时长) - text_content → text_template - transition_effect → 安全解析后映射 - playback_speed → 存入 config(非 1.0 时) - clip.config → 合并入 config(过滤素材字段) Args: template_id: 目标模板 ID clip: 源片段对象(需有 clip_type/order/duration/text_content/ transition_effect/playback_speed/config 属性) Returns: 新创建的 TemplateClipConfig 实例 """ clip_type = safe_parse_clip_type(getattr(clip, "clip_type", None)) transition = safe_parse_transition_effect(getattr(clip, "transition_effect", None)) config = filter_clip_config( getattr(clip, "config", None), playback_speed=getattr(clip, "playback_speed", None), ) duration = getattr(clip, "duration", 0.0) or 0.0 return TemplateClipConfig.create( template_id=template_id, clip_type=clip_type, order=getattr(clip, "order", 0), min_duration=duration, max_duration=duration, text_template=getattr(clip, "text_content", "") or "", transition_effect=transition, config=config, ) def clips_to_template_clip_configs( template_id: str, clips: list[Any], ) -> list[TemplateClipConfig]: """批量将剪辑计划片段转换为模板片段配置列表。 Args: template_id: 目标模板 ID clips: 源片段对象列表 Returns: TemplateClipConfig 实例列表 """ return [clip_to_template_clip_config(template_id, c) for c in clips] # ── TemplateClipConfig → Snapshot 转换 ──────────────────────────────────── def _enum_value(value: Any) -> Any: """获取枚举的 value 值(兼容枚举和字符串)。""" if hasattr(value, "value"): return value.value return value def clip_config_to_snapshot(cfg: Any) -> dict[str, Any]: """将模板片段配置转换为版本快照 dict。 Args: cfg: TemplateClipConfig 对象(或有对应属性的对象) Returns: 快照字典,包含 clip_type/order/min_duration/max_duration/ text_template/transition_effect/config """ return { "clip_type": _enum_value(getattr(cfg, "clip_type", None)), "order": getattr(cfg, "order", 0), "min_duration": getattr(cfg, "min_duration", 0.0), "max_duration": getattr(cfg, "max_duration", 0.0), "text_template": getattr(cfg, "text_template", "") or "", "transition_effect": _enum_value(getattr(cfg, "transition_effect", None)), "config": dict(getattr(cfg, "config", {}) or {}), } def clip_configs_to_snapshots(configs: list[Any]) -> list[dict[str, Any]]: """批量将模板片段配置转换为版本快照列表。""" return [clip_config_to_snapshot(c) for c in configs] # ── Snapshot → TemplateClipConfig 转换 ──────────────────────────────────── def snapshot_to_template_clip_config( template_id: str, snapshot: dict[str, Any], ) -> TemplateClipConfig: """将版本快照 dict 转换为模板片段配置。 Args: template_id: 目标模板 ID snapshot: 快照字典 Returns: 新创建的 TemplateClipConfig 实例 """ clip_type = safe_parse_clip_type(snapshot.get("clip_type", "main")) transition = safe_parse_transition_effect(snapshot.get("transition_effect", "cut")) return TemplateClipConfig.create( template_id=template_id, clip_type=clip_type, order=snapshot.get("order", 0), min_duration=snapshot.get("min_duration", 0.0), max_duration=snapshot.get("max_duration", 0.0), text_template=snapshot.get("text_template", ""), transition_effect=transition, config=dict(snapshot.get("config", {}) or {}), ) def snapshots_to_template_clip_configs( template_id: str, snapshots: list[dict[str, Any]], ) -> list[TemplateClipConfig]: """批量将版本快照转换为模板片段配置列表。""" return [snapshot_to_template_clip_config(template_id, s) for s in snapshots] # ── 名称校验工具 ────────────────────────────────────────────────────────── def validate_template_name(name: str | None) -> str: """校验并清洗模板名称。 Args: name: 原始名称 Returns: 清洗后的名称(去除首尾空格) Raises: ValueError: 名称为空 """ clean_name = name.strip() if name else "" if not clean_name: raise ValueError("模板名称不能为空") return clean_name