xiaoxia-saas/packages/domain/plan_generator_utils.py

"""剪辑计划生成 — 纯逻辑工具函数.

从 PlanGeneratorService 提取的纯业务逻辑：
- 素材分配策略（4 种 editing_mode）
- 默认 clip 结构生成
- clip_type 按模式映射
- 从 TemplateClipConfig 创建 EditPlanClip

纯函数，无副作用，不依赖 DB/外部服务。
"""

from __future__ import annotations

import logging
import random
from typing import Callable, List

logger = logging.getLogger(__name__)

from packages.domain.edit_plan_clip import EditPlanClip
from packages.domain.editing_mode import EditingMode
from packages.domain.template_clip_config import ClipType, TemplateClipConfig

# ── 默认片段时长（秒） ────────────────────────────────────────────────────────
DEFAULT_CLIP_DURATION = 5.0
DEFAULT_INTRO_DURATION = 3.0
DEFAULT_OUTRO_DURATION = 3.0


# ── SceneChange 镜头段工具 ────────────────────────────────────────────────────


def build_scene_segments(
    scene_changes: list[float],
    asset_duration: float,
) -> list[tuple[float, float]]:
    """根据场景切换点构建镜头段列表.

    Args:
        scene_changes: 场景切换点时间戳列表（已排序，首位为 0.0）
        asset_duration: 素材总时长

    Returns:
        镜头段列表 [(start, end), ...]，仅保留长度 >= 0.5s 的段
    """
    segments: list[tuple[float, float]] = []
    for i, ts in enumerate(scene_changes):
        end = scene_changes[i + 1] if i + 1 < len(scene_changes) else asset_duration
        # 只保留有效长度的镜头段（至少 0.5 秒）
        if end - ts >= 0.5:
            segments.append((ts, end))
    return segments


def pick_start_in_scene_segment(
    seg_start: float,
    seg_end: float,
    clip_duration: float,
) -> float | None:
    """在镜头段内随机选取一个起始时间点.

    确保 start + clip_duration <= seg_end。
    若镜头段长度不足以容纳片段，返回 None。
    """
    available = seg_end - seg_start - clip_duration
    if available < 0:
        return None
    max_start = seg_start + available
    return random.uniform(seg_start, max_start)


def _segments_overlap(
    start: float,
    duration: float,
    used: list[tuple[float, float]],
    edge_gap: float = 0.0,
) -> bool:
    """候选区间 [start, start+duration] 是否与已用区间冲突（含边缘间隙扩边）。"""
    end = start + duration
    for used_start, used_end in used:
        if start < used_end + edge_gap and end > used_start - edge_gap:
            return True
    return False


def pick_scene_aware_start(
    asset_id: str,
    clip_duration: float,
    asset_durations: dict[str, float],
    asset_scene_points: dict[str, list[float]] | None,
    used_segments: dict[str, list[tuple[float, float]]],
    *,
    edge_gap: float = 0.0,
) -> float | None:
    """基于缓存的场景切换点，从随机镜头段中选取不冲突的起始时间.

    流程：
    1. 读取 asset_scene_points 中该素材的场景切换点缓存 → 构建镜头段
    2. random.shuffle 镜头段（保证同一素材多次生成选不同镜头，而非固定第N段）
    3. 依次尝试：段内随机取点 → 越界检查 → 与 used_segments 冲突检查
    4. 全部冲突/无缓存 → 返回 None，由调用方回退 _calc_random_start_time

    Args:
        asset_id: 素材 ID
        clip_duration: 片段时长（秒）
        asset_durations: 素材 ID -> 总时长
        asset_scene_points: 素材 ID -> 场景切换点列表（metadata 缓存）
        used_segments: 素材 ID -> 已用区间列表（冲突避让）
        edge_gap: 冲突判定的边缘间隙（秒），已用区间按 [s-gap, e+gap] 扩边
    """
    asset_total = (asset_durations or {}).get(asset_id)
    if not asset_total or asset_total <= 0:
        return None
    scene_points = (asset_scene_points or {}).get(asset_id)
    if not scene_points:
        return None
    used = used_segments.get(asset_id, []) if used_segments else []

    scene_segments = build_scene_segments(scene_points, asset_total)
    if not scene_segments:
        return None
    random.shuffle(scene_segments)

    for seg_start, seg_end in scene_segments:
        candidate = pick_start_in_scene_segment(seg_start, seg_end, clip_duration)
        if candidate is None:
            continue
        # 越界检查（防御：场景点末尾段理论上不越界，metadata 脏数据兜底）
        if candidate + clip_duration > asset_total:
            continue
        # 与已用区间冲突检查
        if _segments_overlap(candidate, clip_duration, used, edge_gap):
            continue
        return candidate

    return None


def extract_scene_points_from_metadata(metadata: object) -> list[float] | None:
    """从素材 metadata 中提取并校验场景切换点缓存.

    合法缓存：list 类型、至少 2 个数值点、单调非负；否则返回 None（按未缓存处理）。
    """
    if not isinstance(metadata, dict):
        return None
    points = metadata.get("scene_change_points")
    if not isinstance(points, list) or len(points) < 2:
        return None
    try:
        cleaned = [float(p) for p in points]
    except (TypeError, ValueError):
        return None
    if any(p < 0 for p in cleaned):
        return None
    cleaned = sorted(cleaned)
    if cleaned[0] != 0.0:
        cleaned.insert(0, 0.0)
    return cleaned


# ── 素材分配 ────────────────────────────────────────────────────────────────


def distribute_assets(
    clips: List[EditPlanClip],
    asset_ids: List[str],
    editing_mode: str,
    *,
    random_selection: bool = False,
    asset_durations: dict[str, float] | None = None,
    asset_scene_points: dict[str, list[float]] | None = None,
    external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
    """按 editing_mode 将素材分配到 clips（就地修改）.

    分配策略:
    - ONE_TAKE: 素材按顺序依次分配给 main 类型 clips
    - PIP: 第1个素材→main(全屏背景)，其余→overlay clips
    - VOICE_OVER: 素材→main clips (B-roll)
    - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll

    start_time 选取：素材 metadata 中有场景切换点缓存时，优先从随机镜头段
    取起点（不同片段来自不同镜头）；无缓存或镜头段都冲突时回退随机起点。

    Args:
        clips: 剪辑片段列表（就地修改 asset_id）
        asset_ids: 素材 ID 列表
        editing_mode: 剪辑模式字符串
        random_selection: 是否随机选择素材（用于预览生成）
        asset_durations: 素材 ID -> 时长（秒）映射，用于设置 start_time
        asset_scene_points: 素材 ID -> 场景切换点列表（metadata 缓存）
        external_used_segments: 跨视频已用区间（来自其他视频的 clips），注入到分配逻辑中避让
    """
    if not asset_ids or not clips:
        return

    # 如果需要随机选择，先打乱素材顺序
    if random_selection:
        asset_ids = list(asset_ids)  # 复制避免修改原列表
        random.shuffle(asset_ids)

    if editing_mode == EditingMode.ONE_TAKE.value:
        _distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
    elif editing_mode == EditingMode.PIP.value:
        _distribute_pip(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
    elif editing_mode == EditingMode.VOICE_OVER.value:
        _distribute_voice_over(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
    elif editing_mode == EditingMode.VOICE_PIP.value:
        _distribute_voice_pip(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
    else:
        # 未知模式，退化为 one_take
        _distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)


def _resolve_start_time(
    asset_id: str,
    clip_duration: float,
    asset_durations: dict[str, float] | None,
    used_segments: dict[str, list[tuple[float, float]]],
    asset_scene_points: dict[str, list[float]] | None = None,
    on_exhausted: Callable[[str, float], tuple[float, float] | None] | None = None,
) -> float | None:
    """选取片段起点：场景缓存优先（随机镜头段），无缓存/全冲突回退随机起点.

    场景路径与随机路径共享 used_segments 冲突避让；场景路径返回 None 时
    （无缓存、镜头段全冲突）回退 _calc_random_start_time，其受控复用逻辑
    （on_exhausted）不受影响。
    """
    if asset_scene_points and asset_scene_points.get(asset_id):
        scene_start = pick_scene_aware_start(
            asset_id,
            clip_duration,
            asset_durations or {},
            asset_scene_points,
            used_segments,
        )
        if scene_start is not None:
            return scene_start
    return _calc_random_start_time(
        asset_id,
        clip_duration,
        asset_durations,
        used_segments,
        on_exhausted=on_exhausted,
    )


def _distribute_one_take(
    clips: List[EditPlanClip],
    asset_ids: List[str],
    asset_durations: dict[str, float] | None = None,
    asset_scene_points: dict[str, list[float]] | None = None,
    external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
    """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips."""
    used_segments: dict[str, list[tuple[float, float]]] = (
        {k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
    )
    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
    for i, clip in enumerate(main_clips):
        if i < len(asset_ids):
            asset_id = asset_ids[i]
            start_time = _resolve_start_time(
                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
            )
            clip.assign_asset(asset_id, start_time=start_time)
            # Record used segment
            if start_time is not None and asset_durations is not None:
                if asset_id not in used_segments:
                    used_segments[asset_id] = []
                used_segments[asset_id].append((start_time, start_time + clip.duration))


def _distribute_pip(
    clips: List[EditPlanClip],
    asset_ids: List[str],
    asset_durations: dict[str, float] | None = None,
    asset_scene_points: dict[str, list[float]] | None = None,
    external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
    """PIP: 第1个素材→main(全屏背景)，其余→overlay clips."""
    used_segments: dict[str, list[tuple[float, float]]] = (
        {k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
    )
    # 第1个素材 → main clip
    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
    if main_clips and asset_ids:
        asset_id = asset_ids[0]
        start_time = _resolve_start_time(
            asset_id, main_clips[0].duration, asset_durations, used_segments, asset_scene_points
        )
        main_clips[0].assign_asset(asset_id, start_time=start_time)
        # Record used segment
        if start_time is not None and asset_durations is not None:
            if asset_id not in used_segments:
                used_segments[asset_id] = []
            used_segments[asset_id].append((start_time, start_time + main_clips[0].duration))

    # 其余素材 → overlay clips
    overlay_clips = [c for c in clips if c.clip_type == "overlay"]
    remaining = asset_ids[1:]
    for i, clip in enumerate(overlay_clips):
        if i < len(remaining):
            asset_id = remaining[i]
            start_time = _resolve_start_time(
                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
            )
            clip.assign_asset(asset_id, start_time=start_time)
            # Record used segment
            if start_time is not None and asset_durations is not None:
                if asset_id not in used_segments:
                    used_segments[asset_id] = []
                used_segments[asset_id].append((start_time, start_time + clip.duration))


def _distribute_voice_over(
    clips: List[EditPlanClip],
    asset_ids: List[str],
    asset_durations: dict[str, float] | None = None,
    asset_scene_points: dict[str, list[float]] | None = None,
    external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
    """VOICE_OVER: 素材→main clips (B-roll)."""
    used_segments: dict[str, list[tuple[float, float]]] = (
        {k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
    )
    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
    for i, clip in enumerate(main_clips):
        if i < len(asset_ids):
            asset_id = asset_ids[i]
            start_time = _resolve_start_time(
                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
            )
            clip.assign_asset(asset_id, start_time=start_time)
            # Record used segment
            if start_time is not None and asset_durations is not None:
                if asset_id not in used_segments:
                    used_segments[asset_id] = []
                used_segments[asset_id].append((start_time, start_time + clip.duration))


def _distribute_voice_pip(
    clips: List[EditPlanClip],
    asset_ids: List[str],
    asset_durations: dict[str, float] | None = None,
    asset_scene_points: dict[str, list[float]] | None = None,
    external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
    """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll."""
    used_segments: dict[str, list[tuple[float, float]]] = (
        {k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
    )
    bg_clips = [c for c in clips if c.clip_type == "background"]
    voice_clips = [c for c in clips if c.clip_type == "corner_voice"]
    broll_clips = [c for c in clips if c.clip_type == "b_roll"]

    idx = 0

    # 第1个 → background
    if idx < len(asset_ids) and bg_clips:
        asset_id = asset_ids[idx]
        start_time = _resolve_start_time(
            asset_id, bg_clips[0].duration, asset_durations, used_segments, asset_scene_points
        )
        bg_clips[0].assign_asset(asset_id, start_time=start_time)
        # Record used segment
        if start_time is not None and asset_durations is not None:
            if asset_id not in used_segments:
                used_segments[asset_id] = []
            used_segments[asset_id].append((start_time, start_time + bg_clips[0].duration))
        idx += 1

    # 第2个 → corner_voice
    if idx < len(asset_ids) and voice_clips:
        asset_id = asset_ids[idx]
        start_time = _resolve_start_time(
            asset_id, voice_clips[0].duration, asset_durations, used_segments, asset_scene_points
        )
        voice_clips[0].assign_asset(asset_id, start_time=start_time)
        # Record used segment
        if start_time is not None and asset_durations is not None:
            if asset_id not in used_segments:
                used_segments[asset_id] = []
            used_segments[asset_id].append((start_time, start_time + voice_clips[0].duration))
        idx += 1

    # 剩余 → b_roll clips
    remaining = asset_ids[idx:]
    for i, clip in enumerate(broll_clips):
        if i < len(remaining):
            asset_id = remaining[i]
            start_time = _resolve_start_time(
                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
            )
            clip.assign_asset(asset_id, start_time=start_time)
            # Record used segment
            if start_time is not None and asset_durations is not None:
                if asset_id not in used_segments:
                    used_segments[asset_id] = []
                used_segments[asset_id].append((start_time, start_time + clip.duration))


# ── 随机 start_time 计算 ────────────────────────────────────────────────────


def _calc_random_start_time(
    asset_id: str,
    clip_duration: float,
    asset_durations: dict[str, float] | None,
    used_segments: dict[str, list[tuple[float, float]]] | None = None,
    on_exhausted: Callable[[str, float], tuple[float, float] | None] | None = None,
) -> float | None:
    """计算随机 start_time，避开已使用的时间段.

    在素材总时长范围内随机取点，确保 clip_duration 不超出素材边界。
    如果 asset_durations 为 None 或素材不在其中，返回 None（使用默认 0.0）。
    如果提供了 used_segments，会避开已使用的时间区间。

    Args:
        asset_id: 素材 ID
        clip_duration: 片段时长（秒）
        asset_durations: 素材 ID -> 时长映射
        used_segments: {asset_id: [(start1, end1), (start2, end2), ...]} 已使用的时间段
        on_exhausted: 100 次随机都找不到空闲区间时的受控复用回调，入参为
            (asset_id, clip_duration)，返回 (start, end) 复用区间或 None。
            历史记录永不自动清空；回调返回 None（全部达上限/复用占比超闸门）时
            本函数返回 None，由调用方轮询下一个素材或报错，不做重叠降级。

    Returns:
        随机 start_time 或 None
    """
    if asset_durations is None:
        return None

    total_duration = asset_durations.get(asset_id)
    if total_duration is None or total_duration <= 0:
        return None

    # 最大起始点 = 素材总时长 - 片段时长
    max_start = max(0.0, total_duration - clip_duration)
    if max_start <= 0:
        return 0.0

    # 如果没有已使用段，直接随机
    if not used_segments or asset_id not in used_segments:
        return random.uniform(0.0, max_start)

    # 尝试找到一个不与已使用段重叠的起始点
    used = sorted(used_segments[asset_id])
    max_attempts = 100

    for _ in range(max_attempts):
        candidate = random.uniform(0.0, max_start)
        candidate_end = candidate + clip_duration

        # 检查是否与任何已使用段重叠
        overlap = False
        for seg_start, seg_end in used:
            # 两个区间 [a, b] 和 [c, d] 重叠的条件: a < d and c < b
            if candidate < seg_end and seg_start < candidate_end:
                overlap = True
                break

        if not overlap:
            return candidate

    # 100 次都找不到空闲区间：进入受控复用，回调从历史区间中选最久未用且
    # 使用次数未达上限的区间返回（历史记录永不自动清空）
    if on_exhausted is not None:
        try:
            reused = on_exhausted(asset_id, clip_duration)
        except Exception:
            logger.warning(
                "on_exhausted 受控复用回调异常: asset_id=%s",
                asset_id,
                exc_info=True,
            )
            reused = None
        if reused is not None:
            reuse_start, reuse_end = reused
            # 边界保护：不越素材末尾、不为负
            reuse_start = max(0.0, min(float(reuse_start), max_start))
            logger.info(
                "素材可用区间耗尽，受控复用历史区间: asset_id=%s start=%.2f end=%.2f",
                asset_id,
                reuse_start,
                reuse_end,
            )
            return reuse_start
        # 回调存在但拒绝复用（区间全部达 use_count 上限，或复用占比将超 15% 闸门）：
        # 返回 None，由调用方轮询下一个素材；绝不能末尾/0.0 降级——那会把片段
        # 放回到已用过的画面，违反区间避让与重复率控制原则
        return None

    # 未提供 on_exhausted 回调（向后兼容）：降级使用素材末尾空闲位置；
    # 末尾也已占满时返回 0.0（旧行为，仅无持久化追踪的调用方会走到这里）
    last_used_end = 0.0
    for _seg_start, seg_end in used:
        last_used_end = max(last_used_end, seg_end)

    if last_used_end < total_duration:
        # 返回从最后使用点开始的位置
        return min(last_used_end, max_start)

    return 0.0


# ── clip_type 映射 ────────────────────────────────────────────────────────


def map_clip_types_for_mode(
    clips: List[EditPlanClip],
    editing_mode: str,
) -> None:
    """将 MAIN 类型片段按 editing_mode 映射为对应角色类型.

    模板的 clip_config 使用 ClipType 枚举（main/intro/outro 等），
    但 PIP / VOICE_PIP 模式的素材分配和渲染分层依赖特定的 clip_type 命名
    （overlay / background / corner_voice / b_roll）。

    映射规则（仅修改 MAIN 类型片段，非 MAIN 片段保持原类型）：
    - PIP: 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画）
    - VOICE_PIP: 第1个 → background，第2个 → corner_voice，第3+个 → b_roll
    - ONE_TAKE / VOICE_OVER: 保持 main 不变

    Args:
        clips: 剪辑片段列表（就地修改 clip_type）
        editing_mode: 剪辑模式字符串
    """
    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
    if not main_clips:
        return

    if editing_mode == EditingMode.PIP.value:
        # 第1个 main 保持（背景层），其余改为 overlay（画中画层）
        for i, clip in enumerate(main_clips):
            if i > 0:
                clip.clip_type = "overlay"

    elif editing_mode == EditingMode.VOICE_PIP.value:
        for i, clip in enumerate(main_clips):
            if i == 0:
                clip.clip_type = "background"
            elif i == 1:
                clip.clip_type = "corner_voice"
            else:
                clip.clip_type = "b_roll"

    # ONE_TAKE / VOICE_OVER: 保持 main 不变，无需处理


# ── 默认 clip 生成 ────────────────────────────────────────────────────────


def generate_default_clips(
    plan_id: str,
    editing_mode: str,
    asset_count: int,
) -> List[EditPlanClip]:
    """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构.

    - ONE_TAKE: N 个 main clips（N = asset_count，至少1个）
    - PIP: 1 个 main + (N-1) 个 overlay（N = asset_count）
    - VOICE_OVER: N 个 main clips + 标记需要配音
    - VOICE_PIP: 1 个 background + 1 个 corner_voice + (N-2) 个 b_roll

    Args:
        plan_id: 剪辑计划 ID
        editing_mode: 剪辑模式字符串
        asset_count: 素材数量

    Returns:
        List[EditPlanClip]: 生成的默认剪辑片段列表
    """
    n = max(asset_count, 1)
    clips: List[EditPlanClip] = []
    order = 0

    if editing_mode == EditingMode.PIP.value:
        # 1 个 main（全屏背景）
        clips.append(
            EditPlanClip.create(
                plan_id=plan_id,
                clip_type=ClipType.MAIN.value,
                order=order,
                duration=DEFAULT_CLIP_DURATION,
            )
        )
        order += 1
        # 剩余为 overlay
        for _ in range(1, n):
            clips.append(
                EditPlanClip.create(
                    plan_id=plan_id,
                    clip_type="overlay",
                    order=order,
                    duration=DEFAULT_CLIP_DURATION,
                )
            )
            order += 1

    elif editing_mode == EditingMode.VOICE_OVER.value:
        # N 个 main clips（B-roll）
        for _ in range(n):
            clips.append(
                EditPlanClip.create(
                    plan_id=plan_id,
                    clip_type=ClipType.MAIN.value,
                    order=order,
                    duration=DEFAULT_CLIP_DURATION,
                    config={"role": "b_roll"},
                )
            )
            order += 1

    elif editing_mode == EditingMode.VOICE_PIP.value:
        # 1 个 background
        clips.append(
            EditPlanClip.create(
                plan_id=plan_id,
                clip_type="background",
                order=order,
                duration=DEFAULT_CLIP_DURATION,
            )
        )
        order += 1
        # 1 个 corner_voice（至少有1个素材就有）
        if n >= 2:
            clips.append(
                EditPlanClip.create(
                    plan_id=plan_id,
                    clip_type="corner_voice",
                    order=order,
                    duration=DEFAULT_CLIP_DURATION,
                )
            )
            order += 1
        # 剩余为 b_roll
        for _ in range(2, n):
            clips.append(
                EditPlanClip.create(
                    plan_id=plan_id,
                    clip_type="b_roll",
                    order=order,
                    duration=DEFAULT_CLIP_DURATION,
                )
            )
            order += 1

    else:
        # ONE_TAKE / 未知模式：N 个 main clips
        for _ in range(n):
            clips.append(
                EditPlanClip.create(
                    plan_id=plan_id,
                    clip_type=ClipType.MAIN.value,
                    order=order,
                    duration=DEFAULT_CLIP_DURATION,
                )
            )
            order += 1

    return clips


# ── 从配置创建 clips ────────────────────────────────────────────────────────


def create_clips_from_configs(
    plan_id: str,
    clip_configs: List[TemplateClipConfig],
) -> List[EditPlanClip]:
    """从 TemplateClipConfig 列表创建 EditPlanClip 列表.

    Args:
        plan_id: 剪辑计划 ID
        clip_configs: 模板片段配置列表

    Returns:
        List[EditPlanClip]: 创建的剪辑片段列表（按 order 排序）
    """
    clips: List[EditPlanClip] = []
    # 按 order 排序
    sorted_configs = sorted(clip_configs, key=lambda c: c.order)

    for cfg in sorted_configs:
        # 计算时长：取 min_duration 和 max_duration 的中间值
        if cfg.min_duration > 0 and cfg.max_duration > 0:
            duration = (cfg.min_duration + cfg.max_duration) / 2
        elif cfg.min_duration > 0:
            duration = cfg.min_duration
        elif cfg.max_duration > 0:
            duration = cfg.max_duration
        else:
            duration = DEFAULT_CLIP_DURATION

        # clip_type 可能是枚举或字符串
        clip_type = cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type

        # transition_effect 可能是枚举或字符串
        transition = cfg.transition_effect.value if hasattr(cfg.transition_effect, "value") else cfg.transition_effect

        # 从 clip config 中解析 playback_speed（兼容 speed_ratio 字段名）
        clip_cfg = cfg.config or {}
        playback_speed = clip_cfg.get("playback_speed", clip_cfg.get("speed_ratio", 1.0)) or 1.0

        clip = EditPlanClip.create(
            plan_id=plan_id,
            clip_type=clip_type,
            order=cfg.order,
            template_clip_config_id=cfg.id,
            text_content=getattr(cfg, "text_template", "") or "",
            duration=duration,
            transition_effect=transition or "cut",
            playback_speed=playback_speed,
            config=clip_cfg,
        )
        clips.append(clip)

    return clips