294 lines
9.2 KiB
Python
Executable File
294 lines
9.2 KiB
Python
Executable File
"""模板片段转换器 — 纯函数集合.
|
||
|
||
从 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
|