Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 28745c401d |
@@ -23,6 +23,13 @@ from packages.domain.template_clip_config import (
|
||||
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__)
|
||||
|
||||
@@ -121,9 +128,7 @@ class EditTemplateService:
|
||||
ValueError: 名称为空或重复
|
||||
"""
|
||||
# 名称校验
|
||||
clean_name = name.strip()
|
||||
if not clean_name:
|
||||
raise ValueError("模板名称不能为空")
|
||||
clean_name = validate_template_name(name)
|
||||
|
||||
# 名称重复检查
|
||||
existing = self._template_repo.list_all(skip=0, limit=1000)
|
||||
@@ -471,12 +476,7 @@ class EditTemplateService:
|
||||
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_config = filter_plan_config_to_template(plan.config)
|
||||
|
||||
template = EditTemplate.create(
|
||||
name=clean_name,
|
||||
@@ -497,40 +497,7 @@ class EditTemplateService:
|
||||
|
||||
# 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,
|
||||
)
|
||||
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)
|
||||
|
||||
@@ -679,8 +646,6 @@ class EditTemplateService:
|
||||
Raises:
|
||||
ValueError: 模板/草稿不存在,或草稿不属于该模板
|
||||
"""
|
||||
from packages.domain.template_clip_config import TemplateClipConfig
|
||||
|
||||
# 1. 校验模板和草稿
|
||||
template = self.get_template_or_raise(template_id)
|
||||
draft = self._plan_repo.get(draft_plan_id)
|
||||
@@ -700,39 +665,14 @@ class EditTemplateService:
|
||||
editing_mode = config.get("editing_mode", "one_take")
|
||||
|
||||
# 4. 提取模板配置(去掉草稿/运行时字段)
|
||||
draft_config = draft.config or {}
|
||||
template_config: dict[str, Any] = {}
|
||||
skip_keys = {
|
||||
"is_template_draft",
|
||||
"asset_ids",
|
||||
"source_edit_plan_id",
|
||||
"generation_task_id",
|
||||
}
|
||||
for key, value in draft_config.items():
|
||||
if key not in skip_keys:
|
||||
template_config[key] = value
|
||||
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_type": cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
|
||||
"order": cfg.order,
|
||||
"min_duration": cfg.min_duration,
|
||||
"max_duration": cfg.max_duration,
|
||||
"text_template": cfg.text_template or "",
|
||||
"transition_effect": (
|
||||
cfg.transition_effect.value
|
||||
if hasattr(cfg.transition_effect, "value")
|
||||
else cfg.transition_effect
|
||||
),
|
||||
"config": cfg.config or {},
|
||||
}
|
||||
for cfg in old_clip_configs
|
||||
]
|
||||
old_clip_snapshots = clip_configs_to_snapshots(old_clip_configs)
|
||||
|
||||
from packages.domain.template_version import EditTemplateVersion
|
||||
|
||||
@@ -759,46 +699,7 @@ class EditTemplateService:
|
||||
|
||||
# 创建新的片段配置
|
||||
created_configs: list[TemplateClipConfig] = []
|
||||
for clip in draft_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 合并
|
||||
if clip.config:
|
||||
clip_config.update(clip.config)
|
||||
# 去掉素材相关字段
|
||||
clip_config.pop("asset_info", None)
|
||||
clip_config.pop("source_asset_id", None)
|
||||
|
||||
# 转场效果兼容校验
|
||||
try:
|
||||
from packages.domain.template_clip_config import (
|
||||
TransitionEffect,
|
||||
)
|
||||
|
||||
transition = TransitionEffect(clip.transition_effect)
|
||||
except (ValueError, ImportError):
|
||||
transition = TransitionEffect.CUT # type: ignore
|
||||
|
||||
# 片段类型兼容校验
|
||||
try:
|
||||
from packages.domain.template_clip_config import ClipType
|
||||
|
||||
clip_type = ClipType(clip.clip_type)
|
||||
except (ValueError, ImportError):
|
||||
clip_type = ClipType.MAIN # type: ignore
|
||||
|
||||
config_obj = TemplateClipConfig.create(
|
||||
template_id=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,
|
||||
)
|
||||
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)
|
||||
|
||||
@@ -843,8 +744,6 @@ class EditTemplateService:
|
||||
Raises:
|
||||
ValueError: 模板/版本不存在
|
||||
"""
|
||||
from packages.domain.template_clip_config import TemplateClipConfig
|
||||
|
||||
template = self.get_template_or_raise(template_id)
|
||||
|
||||
# 1. 读取目标版本快照
|
||||
@@ -857,22 +756,7 @@ class EditTemplateService:
|
||||
try:
|
||||
# 2. 先保存当前状态快照(当前版本号),确保回滚可撤销
|
||||
old_clip_configs = self._clip_config_repo.list_by_template(template_id)
|
||||
old_clip_snapshots = [
|
||||
{
|
||||
"clip_type": cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
|
||||
"order": cfg.order,
|
||||
"min_duration": cfg.min_duration,
|
||||
"max_duration": cfg.max_duration,
|
||||
"text_template": cfg.text_template or "",
|
||||
"transition_effect": (
|
||||
cfg.transition_effect.value
|
||||
if hasattr(cfg.transition_effect, "value")
|
||||
else cfg.transition_effect
|
||||
),
|
||||
"config": cfg.config or {},
|
||||
}
|
||||
for cfg in old_clip_configs
|
||||
]
|
||||
old_clip_snapshots = clip_configs_to_snapshots(old_clip_configs)
|
||||
|
||||
from packages.domain.template_version import EditTemplateVersion
|
||||
|
||||
@@ -905,37 +789,7 @@ class EditTemplateService:
|
||||
synchronize_session=False
|
||||
)
|
||||
|
||||
for clip_snap in target_version.clip_configs:
|
||||
# 转场效果兼容校验
|
||||
try:
|
||||
from packages.domain.template_clip_config import TransitionEffect
|
||||
|
||||
transition = TransitionEffect(clip_snap.get("transition_effect", "cut"))
|
||||
except (ValueError, ImportError):
|
||||
from packages.domain.template_clip_config import TransitionEffect
|
||||
|
||||
transition = TransitionEffect.CUT
|
||||
|
||||
# 片段类型兼容校验
|
||||
try:
|
||||
from packages.domain.template_clip_config import ClipType
|
||||
|
||||
clip_type = ClipType(clip_snap.get("clip_type", "main"))
|
||||
except (ValueError, ImportError):
|
||||
from packages.domain.template_clip_config import ClipType
|
||||
|
||||
clip_type = ClipType.MAIN
|
||||
|
||||
config_obj = TemplateClipConfig.create(
|
||||
template_id=template_id,
|
||||
clip_type=clip_type,
|
||||
order=clip_snap.get("order", 0),
|
||||
min_duration=clip_snap.get("min_duration", 0.0),
|
||||
max_duration=clip_snap.get("max_duration", 0.0),
|
||||
text_template=clip_snap.get("text_template", ""),
|
||||
transition_effect=transition,
|
||||
config=clip_snap.get("config", {}) or {},
|
||||
)
|
||||
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()
|
||||
|
||||
Executable
+293
@@ -0,0 +1,293 @@
|
||||
"""模板片段转换器 — 纯函数集合.
|
||||
|
||||
从 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
|
||||
Executable
+480
@@ -0,0 +1,480 @@
|
||||
"""template_clip_converter 模块单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.template_clip_config import ClipType, TransitionEffect
|
||||
from packages.domain.template_clip_converter import (
|
||||
clip_config_to_snapshot,
|
||||
clip_configs_to_snapshots,
|
||||
clip_to_template_clip_config,
|
||||
clips_to_template_clip_configs,
|
||||
filter_clip_config,
|
||||
filter_plan_config_to_template,
|
||||
safe_parse_clip_type,
|
||||
safe_parse_transition_effect,
|
||||
snapshot_to_template_clip_config,
|
||||
snapshots_to_template_clip_configs,
|
||||
validate_template_name,
|
||||
)
|
||||
|
||||
# ── 辅助数据类 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeClip:
|
||||
"""模拟剪辑计划片段对象."""
|
||||
|
||||
clip_type: Any = "main"
|
||||
order: int = 0
|
||||
duration: float = 5.0
|
||||
text_content: str = ""
|
||||
transition_effect: Any = "cut"
|
||||
playback_speed: float | None = None
|
||||
config: dict[str, Any] | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeClipConfig:
|
||||
"""模拟模板片段配置对象."""
|
||||
|
||||
clip_type: Any = "main"
|
||||
order: int = 0
|
||||
min_duration: float = 0.0
|
||||
max_duration: float = 0.0
|
||||
text_template: str = ""
|
||||
transition_effect: Any = "cut"
|
||||
config: dict[str, Any] | None = None
|
||||
|
||||
|
||||
# ── safe_parse_transition_effect ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSafeParseTransitionEffect:
|
||||
def test_enum_value_passthrough(self):
|
||||
assert safe_parse_transition_effect(TransitionEffect.DISSOLVE) == TransitionEffect.DISSOLVE
|
||||
|
||||
def test_valid_string(self):
|
||||
assert safe_parse_transition_effect("dissolve") == TransitionEffect.DISSOLVE
|
||||
|
||||
def test_invalid_string_defaults_to_cut(self):
|
||||
assert safe_parse_transition_effect("invalid_effect") == TransitionEffect.CUT
|
||||
|
||||
def test_none_defaults_to_cut(self):
|
||||
assert safe_parse_transition_effect(None) == TransitionEffect.CUT
|
||||
|
||||
def test_custom_default(self):
|
||||
assert safe_parse_transition_effect("bad", default=TransitionEffect.FADE) == TransitionEffect.FADE
|
||||
|
||||
def test_int_value(self):
|
||||
assert safe_parse_transition_effect(123) == TransitionEffect.CUT
|
||||
|
||||
|
||||
# ── safe_parse_clip_type ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSafeParseClipType:
|
||||
def test_enum_value_passthrough(self):
|
||||
assert safe_parse_clip_type(ClipType.INTRO) == ClipType.INTRO
|
||||
|
||||
def test_valid_string(self):
|
||||
assert safe_parse_clip_type("intro") == ClipType.INTRO
|
||||
|
||||
def test_invalid_string_defaults_to_main(self):
|
||||
assert safe_parse_clip_type("invalid_type") == ClipType.MAIN
|
||||
|
||||
def test_none_defaults_to_main(self):
|
||||
assert safe_parse_clip_type(None) == ClipType.MAIN
|
||||
|
||||
def test_custom_default(self):
|
||||
assert safe_parse_clip_type("bad", default=ClipType.OUTRO) == ClipType.OUTRO
|
||||
|
||||
def test_int_value(self):
|
||||
assert safe_parse_clip_type(42) == ClipType.MAIN
|
||||
|
||||
|
||||
# ── filter_clip_config ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestFilterClipConfig:
|
||||
def test_none_config_no_speed(self):
|
||||
result = filter_clip_config(None)
|
||||
assert result == {}
|
||||
|
||||
def test_empty_config_no_speed(self):
|
||||
result = filter_clip_config({})
|
||||
assert result == {}
|
||||
|
||||
def test_playback_speed_added_when_not_default(self):
|
||||
result = filter_clip_config(None, playback_speed=1.5)
|
||||
assert result == {"playback_speed": 1.5}
|
||||
|
||||
def test_playback_speed_skipped_when_default(self):
|
||||
result = filter_clip_config(None, playback_speed=1.0)
|
||||
assert result == {}
|
||||
|
||||
def test_playback_speed_none_skipped(self):
|
||||
result = filter_clip_config(None, playback_speed=None)
|
||||
assert result == {}
|
||||
|
||||
def test_config_merged(self):
|
||||
result = filter_clip_config({"filter": "vintage", "intensity": 0.5})
|
||||
assert result == {"filter": "vintage", "intensity": 0.5}
|
||||
|
||||
def test_asset_info_removed(self):
|
||||
result = filter_clip_config({"asset_info": {"name": "test.mp4"}, "filter": "vintage"})
|
||||
assert "asset_info" not in result
|
||||
assert result["filter"] == "vintage"
|
||||
|
||||
def test_source_asset_id_removed(self):
|
||||
result = filter_clip_config({"source_asset_id": "abc123", "filter": "vintage"})
|
||||
assert "source_asset_id" not in result
|
||||
assert result["filter"] == "vintage"
|
||||
|
||||
def test_speed_overrides_config_playback_speed(self):
|
||||
result = filter_clip_config({"playback_speed": 2.0}, playback_speed=0.5)
|
||||
assert result["playback_speed"] == 2.0 # config 优先级更高
|
||||
|
||||
def test_custom_skip_keys(self):
|
||||
skip = frozenset({"custom_field"})
|
||||
result = filter_clip_config(
|
||||
{"custom_field": "x", "asset_info": "keep_it"},
|
||||
skip_keys=skip,
|
||||
)
|
||||
assert "custom_field" not in result
|
||||
assert "asset_info" in result # 自定义 skip 覆盖默认
|
||||
|
||||
|
||||
# ── filter_plan_config_to_template ──────────────────────────────────────────
|
||||
|
||||
|
||||
class TestFilterPlanConfigToTemplate:
|
||||
def test_none_config(self):
|
||||
assert filter_plan_config_to_template(None) == {}
|
||||
|
||||
def test_empty_config(self):
|
||||
assert filter_plan_config_to_template({}) == {}
|
||||
|
||||
def test_draft_flag_removed(self):
|
||||
result = filter_plan_config_to_template({"is_template_draft": True, "editing_mode": "one_take"})
|
||||
assert "is_template_draft" not in result
|
||||
assert result["editing_mode"] == "one_take"
|
||||
|
||||
def test_asset_ids_removed(self):
|
||||
result = filter_plan_config_to_template({"asset_ids": ["a", "b"], "resolution": "1080p"})
|
||||
assert "asset_ids" not in result
|
||||
assert result["resolution"] == "1080p"
|
||||
|
||||
def test_source_edit_plan_id_removed(self):
|
||||
result = filter_plan_config_to_template({"source_edit_plan_id": "plan123", "theme": "dark"})
|
||||
assert "source_edit_plan_id" not in result
|
||||
assert result["theme"] == "dark"
|
||||
|
||||
def test_generation_task_id_removed(self):
|
||||
result = filter_plan_config_to_template({"generation_task_id": "task123", "bgm": "on"})
|
||||
assert "generation_task_id" not in result
|
||||
assert result["bgm"] == "on"
|
||||
|
||||
def test_normal_fields_preserved(self):
|
||||
config = {
|
||||
"editing_mode": "pip",
|
||||
"resolution": "720p",
|
||||
"duration": 30,
|
||||
"style": "cinematic",
|
||||
}
|
||||
result = filter_plan_config_to_template(config)
|
||||
assert result == config
|
||||
|
||||
def test_custom_skip_keys(self):
|
||||
skip = frozenset({"secret_field"})
|
||||
result = filter_plan_config_to_template(
|
||||
{"secret_field": "x", "is_template_draft": "keep"},
|
||||
skip_keys=skip,
|
||||
)
|
||||
assert "secret_field" not in result
|
||||
assert "is_template_draft" in result
|
||||
|
||||
|
||||
# ── clip_to_template_clip_config ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestClipToTemplateClipConfig:
|
||||
def test_basic_conversion(self):
|
||||
clip = FakeClip(
|
||||
clip_type="main",
|
||||
order=2,
|
||||
duration=3.5,
|
||||
text_content="Hello world",
|
||||
transition_effect="dissolve",
|
||||
)
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.template_id == "tmpl_001"
|
||||
assert result.clip_type == ClipType.MAIN
|
||||
assert result.order == 2
|
||||
assert result.min_duration == 3.5
|
||||
assert result.max_duration == 3.5
|
||||
assert result.text_template == "Hello world"
|
||||
assert result.transition_effect == TransitionEffect.DISSOLVE
|
||||
|
||||
def test_playback_speed_in_config(self):
|
||||
clip = FakeClip(playback_speed=2.0)
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.config["playback_speed"] == 2.0
|
||||
|
||||
def test_default_speed_not_in_config(self):
|
||||
clip = FakeClip(playback_speed=1.0)
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert "playback_speed" not in result.config
|
||||
|
||||
def test_config_preserved_and_filtered(self):
|
||||
clip = FakeClip(config={"filter": "vintage", "asset_info": {"id": "x"}})
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.config["filter"] == "vintage"
|
||||
assert "asset_info" not in result.config
|
||||
|
||||
def test_text_content_none_becomes_empty(self):
|
||||
clip = FakeClip(text_content=None)
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.text_template == ""
|
||||
|
||||
def test_invalid_type_falls_back(self):
|
||||
clip = FakeClip(clip_type="nonexistent")
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.clip_type == ClipType.MAIN
|
||||
|
||||
def test_invalid_transition_falls_back(self):
|
||||
clip = FakeClip(transition_effect="nonexistent")
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.transition_effect == TransitionEffect.CUT
|
||||
|
||||
def test_enum_type_input(self):
|
||||
clip = FakeClip(clip_type=ClipType.INTRO, transition_effect=TransitionEffect.FADE)
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.clip_type == ClipType.INTRO
|
||||
assert result.transition_effect == TransitionEffect.FADE
|
||||
|
||||
def test_duration_none_defaults_zero(self):
|
||||
clip = FakeClip(duration=None)
|
||||
result = clip_to_template_clip_config("tmpl_001", clip)
|
||||
assert result.min_duration == 0.0
|
||||
assert result.max_duration == 0.0
|
||||
|
||||
|
||||
class TestClipsToTemplateClipConfigs:
|
||||
def test_empty_list(self):
|
||||
result = clips_to_template_clip_configs("tmpl_001", [])
|
||||
assert result == []
|
||||
|
||||
def test_multiple_clips(self):
|
||||
clips = [
|
||||
FakeClip(clip_type="intro", order=0, duration=2.0),
|
||||
FakeClip(clip_type="main", order=1, duration=5.0),
|
||||
FakeClip(clip_type="outro", order=2, duration=3.0),
|
||||
]
|
||||
result = clips_to_template_clip_configs("tmpl_001", clips)
|
||||
assert len(result) == 3
|
||||
assert result[0].clip_type == ClipType.INTRO
|
||||
assert result[1].clip_type == ClipType.MAIN
|
||||
assert result[2].clip_type == ClipType.OUTRO
|
||||
assert all(r.template_id == "tmpl_001" for r in result)
|
||||
|
||||
|
||||
# ── clip_config_to_snapshot ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestClipConfigToSnapshot:
|
||||
def test_basic_snapshot(self):
|
||||
cfg = FakeClipConfig(
|
||||
clip_type="main",
|
||||
order=1,
|
||||
min_duration=2.0,
|
||||
max_duration=5.0,
|
||||
text_template="hello",
|
||||
transition_effect="dissolve",
|
||||
config={"filter": "vintage"},
|
||||
)
|
||||
snap = clip_config_to_snapshot(cfg)
|
||||
assert snap["clip_type"] == "main"
|
||||
assert snap["order"] == 1
|
||||
assert snap["min_duration"] == 2.0
|
||||
assert snap["max_duration"] == 5.0
|
||||
assert snap["text_template"] == "hello"
|
||||
assert snap["transition_effect"] == "dissolve"
|
||||
assert snap["config"] == {"filter": "vintage"}
|
||||
|
||||
def test_enum_values_converted_to_strings(self):
|
||||
cfg = FakeClipConfig(
|
||||
clip_type=ClipType.INTRO,
|
||||
transition_effect=TransitionEffect.FADE,
|
||||
)
|
||||
snap = clip_config_to_snapshot(cfg)
|
||||
assert snap["clip_type"] == "intro"
|
||||
assert snap["transition_effect"] == "fade"
|
||||
|
||||
def test_none_text_template_becomes_empty(self):
|
||||
cfg = FakeClipConfig(text_template=None)
|
||||
snap = clip_config_to_snapshot(cfg)
|
||||
assert snap["text_template"] == ""
|
||||
|
||||
def test_none_config_becomes_empty_dict(self):
|
||||
cfg = FakeClipConfig(config=None)
|
||||
snap = clip_config_to_snapshot(cfg)
|
||||
assert snap["config"] == {}
|
||||
|
||||
def test_config_is_copy_not_reference(self):
|
||||
original = {"key": "value"}
|
||||
cfg = FakeClipConfig(config=original)
|
||||
snap = clip_config_to_snapshot(cfg)
|
||||
snap["config"]["key"] = "modified"
|
||||
assert original["key"] == "value"
|
||||
|
||||
|
||||
class TestClipConfigsToSnapshots:
|
||||
def test_empty_list(self):
|
||||
assert clip_configs_to_snapshots([]) == []
|
||||
|
||||
def test_multiple_configs(self):
|
||||
configs = [
|
||||
FakeClipConfig(clip_type="intro", order=0),
|
||||
FakeClipConfig(clip_type="main", order=1),
|
||||
]
|
||||
result = clip_configs_to_snapshots(configs)
|
||||
assert len(result) == 2
|
||||
assert result[0]["clip_type"] == "intro"
|
||||
assert result[1]["order"] == 1
|
||||
|
||||
|
||||
# ── snapshot_to_template_clip_config ────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSnapshotToTemplateClipConfig:
|
||||
def test_basic_conversion(self):
|
||||
snap = {
|
||||
"clip_type": "intro",
|
||||
"order": 2,
|
||||
"min_duration": 1.0,
|
||||
"max_duration": 3.0,
|
||||
"text_template": "hi",
|
||||
"transition_effect": "dissolve",
|
||||
"config": {"filter": "bw"},
|
||||
}
|
||||
result = snapshot_to_template_clip_config("tmpl_001", snap)
|
||||
assert result.template_id == "tmpl_001"
|
||||
assert result.clip_type == ClipType.INTRO
|
||||
assert result.order == 2
|
||||
assert result.min_duration == 1.0
|
||||
assert result.max_duration == 3.0
|
||||
assert result.text_template == "hi"
|
||||
assert result.transition_effect == TransitionEffect.DISSOLVE
|
||||
assert result.config == {"filter": "bw"}
|
||||
|
||||
def test_missing_fields_get_defaults(self):
|
||||
result = snapshot_to_template_clip_config("tmpl_001", {})
|
||||
assert result.clip_type == ClipType.MAIN
|
||||
assert result.order == 0
|
||||
assert result.min_duration == 0.0
|
||||
assert result.max_duration == 0.0
|
||||
assert result.text_template == ""
|
||||
assert result.transition_effect == TransitionEffect.CUT
|
||||
assert result.config == {}
|
||||
|
||||
def test_invalid_type_falls_back(self):
|
||||
snap = {"clip_type": "invalid"}
|
||||
result = snapshot_to_template_clip_config("tmpl_001", snap)
|
||||
assert result.clip_type == ClipType.MAIN
|
||||
|
||||
def test_invalid_transition_falls_back(self):
|
||||
snap = {"transition_effect": "invalid"}
|
||||
result = snapshot_to_template_clip_config("tmpl_001", snap)
|
||||
assert result.transition_effect == TransitionEffect.CUT
|
||||
|
||||
def test_none_config_becomes_empty_dict(self):
|
||||
snap = {"config": None}
|
||||
result = snapshot_to_template_clip_config("tmpl_001", snap)
|
||||
assert result.config == {}
|
||||
|
||||
|
||||
class TestSnapshotsToTemplateClipConfigs:
|
||||
def test_empty_list(self):
|
||||
assert snapshots_to_template_clip_configs("tmpl_001", []) == []
|
||||
|
||||
def test_multiple_snapshots(self):
|
||||
snaps = [
|
||||
{"clip_type": "intro", "order": 0},
|
||||
{"clip_type": "outro", "order": 2},
|
||||
]
|
||||
result = snapshots_to_template_clip_configs("tmpl_001", snaps)
|
||||
assert len(result) == 2
|
||||
assert result[0].clip_type == ClipType.INTRO
|
||||
assert result[1].clip_type == ClipType.OUTRO
|
||||
assert all(r.template_id == "tmpl_001" for r in result)
|
||||
|
||||
|
||||
# ── 往返一致性测试 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestRoundTrip:
|
||||
def test_snapshot_clip_config_roundtrip(self):
|
||||
"""snapshot → TemplateClipConfig → snapshot 应保持一致."""
|
||||
original = {
|
||||
"clip_type": "intro",
|
||||
"order": 3,
|
||||
"min_duration": 1.5,
|
||||
"max_duration": 4.0,
|
||||
"text_template": "test text",
|
||||
"transition_effect": "dissolve",
|
||||
"config": {"key": "value", "nested": {"a": 1}},
|
||||
}
|
||||
cfg = snapshot_to_template_clip_config("tmpl_test", original)
|
||||
result = clip_config_to_snapshot(cfg)
|
||||
assert result == original
|
||||
|
||||
def test_clip_to_config_to_snapshot(self):
|
||||
"""clip → TemplateClipConfig → snapshot 的预期结果."""
|
||||
clip = FakeClip(
|
||||
clip_type="main",
|
||||
order=1,
|
||||
duration=5.0,
|
||||
text_content="hello",
|
||||
transition_effect="fade",
|
||||
playback_speed=1.5,
|
||||
config={"filter": "vintage", "asset_info": "should_remove"},
|
||||
)
|
||||
cfg = clip_to_template_clip_config("tmpl_001", clip)
|
||||
snap = clip_config_to_snapshot(cfg)
|
||||
assert snap["clip_type"] == "main"
|
||||
assert snap["order"] == 1
|
||||
assert snap["min_duration"] == 5.0
|
||||
assert snap["max_duration"] == 5.0
|
||||
assert snap["text_template"] == "hello"
|
||||
assert snap["transition_effect"] == "fade"
|
||||
assert snap["config"]["playback_speed"] == 1.5
|
||||
assert snap["config"]["filter"] == "vintage"
|
||||
assert "asset_info" not in snap["config"]
|
||||
|
||||
|
||||
# ── validate_template_name ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestValidateTemplateName:
|
||||
def test_valid_name(self):
|
||||
assert validate_template_name("My Template") == "My Template"
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
assert validate_template_name(" Hello ") == "Hello"
|
||||
|
||||
def test_empty_string_raises(self):
|
||||
with pytest.raises(ValueError, match="名称不能为空"):
|
||||
validate_template_name("")
|
||||
|
||||
def test_whitespace_only_raises(self):
|
||||
with pytest.raises(ValueError, match="名称不能为空"):
|
||||
validate_template_name(" ")
|
||||
|
||||
def test_none_raises(self):
|
||||
with pytest.raises(ValueError, match="名称不能为空"):
|
||||
validate_template_name(None)
|
||||
Reference in New Issue
Block a user