xiaoxia-saas/packages/domain/clip_operations.py

"""片段操作工具 — EditPlanClip 分割/合并等纯逻辑操作。

从 edit_plan_service.py 抽离的纯函数集合，专门负责：
- 片段分割：将一个片段从指定位置拆分为两个
- 片段合并：将多个连续片段合并为一个
- Order 重排计算

所有函数均为纯函数，不依赖数据库或外部 IO。
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

# ── 常量 ──────────────────────────────────────────────────────────────────────

DEFAULT_SPLIT_DURATION = 5.0
ROUND_PRECISION = 3


# ── 数据结构 ──────────────────────────────────────────────────────────────────


@dataclass(frozen=True)
class SplitResult:
    """片段分割结果。"""

    left_duration: float
    right_duration: float
    right_start_time: float
    left_trim_end: float
    right_trim_start: float


@dataclass(frozen=True)
class MergeResult:
    """片段合并结果。"""

    total_duration: float
    merged_text: str
    merged_config: dict[str, Any]
    first_order: int
    shift_amount: int


# ── 分割 ──────────────────────────────────────────────────────────────────────


def validate_split_time(split_time: float, duration: float) -> None:
    """校验分割时间是否合法。

    Args:
        split_time: 分割点（秒）
        duration: 原片段时长（秒）

    Raises:
        ValueError: 分割时间不在 (0, duration) 范围内
    """
    if split_time <= 0 or split_time >= duration:
        raise ValueError(f"分割时间必须在 (0, {duration:.3f}) 范围内，当前: {split_time}")


def calculate_split(
    duration: float,
    split_time: float,
    start_time: float = 0.0,
    *,
    precision: int = ROUND_PRECISION,
) -> SplitResult:
    """计算片段分割后的各项参数。

    左半部分：从 0 到 split_time
    右半部分：从 split_time 到 duration

    Args:
        duration: 原片段时长（秒）
        split_time: 分割点（秒）
        start_time: 原片段起始时间（秒），右半部分 start_time 需要加上 left_duration
        precision: 小数精度（默认 3 位，即毫秒）

    Returns:
        SplitResult 包含左右部分的时长、右半部分 start_time、trim 信息
    """
    validate_split_time(split_time, duration)

    left_duration = round(split_time, precision)
    right_duration = round(duration - split_time, precision)
    right_start_time = round(start_time + left_duration, precision)

    return SplitResult(
        left_duration=left_duration,
        right_duration=right_duration,
        right_start_time=right_start_time,
        left_trim_end=right_duration,
        right_trim_start=left_duration,
    )


# ── 合并 ──────────────────────────────────────────────────────────────────────


def validate_merge_clips(clips: list[Any]) -> tuple[str, int]:
    """校验待合并的片段列表。

    校验项：
    1. 至少 2 个片段
    2. 属于同一计划
    3. order 连续
    4. 类型一致

    Args:
        clips: 按任意顺序排列的片段列表（会自动按 order 排序）

    Returns:
        (plan_id, first_order) 元组

    Raises:
        ValueError: 校验失败
    """
    if len(clips) < 2:
        raise ValueError("至少需要 2 个片段才能合并")

    # 校验：同一计划
    plan_id = clips[0].plan_id
    for c in clips[1:]:
        if c.plan_id != plan_id:
            raise ValueError("只能合并同一计划下的片段")

    # 按 order 排序
    sorted_clips = sorted(clips, key=lambda c: c.order)

    # 校验：order 连续
    for i in range(1, len(sorted_clips)):
        if sorted_clips[i].order != sorted_clips[i - 1].order + 1:
            raise ValueError(f"片段不连续：order {sorted_clips[i-1].order} → {sorted_clips[i].order}")

    # 校验：类型一致
    clip_type = sorted_clips[0].clip_type
    for c in sorted_clips[1:]:
        if c.clip_type != clip_type:
            raise ValueError("只能合并相同类型的片段")

    return plan_id, sorted_clips[0].order


def calculate_merge(
    clips: list[Any],
    *,
    precision: int = ROUND_PRECISION,
) -> MergeResult:
    """计算多个片段合并后的参数。

    合并规则：
    - 时长：所有片段时长之和
    - 文案：用换行连接非空文案
    - config：后面的覆盖前面的，移除 trim_start/trim_end
    - first_order：第一个片段的 order
    - shift_amount：合并后 order 前移位数（n-1）

    Args:
        clips: 待合并片段列表（会自动按 order 排序）
        precision: 时长精度（默认 3 位）

    Returns:
        MergeResult 合并结果
    """
    if not clips:
        raise ValueError("合并的片段列表不能为空")

    # 按 order 排序
    sorted_clips = sorted(clips, key=lambda c: c.order)

    # 总时长
    total_duration = round(sum(c.duration for c in sorted_clips), precision)

    # 合并文案
    merged_text = "\n".join(c.text_content for c in sorted_clips if c.text_content and c.text_content.strip())

    # 合并 config（后面的覆盖前面的）
    merged_config: dict[str, Any] = {}
    for c in sorted_clips:
        if c.config:
            merged_config.update(c.config)
    # 清理 trim 相关字段（合并后就是完整片段了）
    merged_config.pop("trim_start", None)
    merged_config.pop("trim_end", None)

    first_order = sorted_clips[0].order
    shift_amount = len(sorted_clips) - 1

    return MergeResult(
        total_duration=total_duration,
        merged_text=merged_text,
        merged_config=merged_config,
        first_order=first_order,
        shift_amount=shift_amount,
    )


# ── Order 重排 ───────────────────────────────────────────────────────────────


def calculate_reorder_new_orders(
    ordered_ids: list[str],
    current_items: list[Any],
    *,
    id_attr: str = "id",
    order_attr: str = "order",
) -> dict[str, int]:
    """根据新顺序计算每个 item 的新 order 值。

    Args:
        ordered_ids: 按新顺序排列的 ID 列表
        current_items: 当前所有 item 列表
        id_attr: ID 属性名
        order_attr: order 属性名

    Returns:
        {item_id: new_order} 映射

    Raises:
        ValueError: ID 列表与当前 items 不匹配
    """
    current_ids = {getattr(c, id_attr) for c in current_items}
    ordered_id_set = set(ordered_ids)

    if ordered_id_set != current_ids:
        raise ValueError("ID 列表与当前 items 不匹配")

    return {item_id: idx for idx, item_id in enumerate(ordered_ids)}


def calculate_shift_orders(
    items: list[Any],
    threshold_order: int,
    shift: int,
    *,
    excluded_ids: set[str] | None = None,
    order_attr: str = "order",
    id_attr: str = "id",
) -> list[tuple[Any, int]]:
    """计算 order 需要偏移的 items 及新 order 值。

    Args:
        items: 所有 item 列表
        threshold_order: 只处理 order > threshold_order 的 item
        shift: 偏移量（正数加，负数减）
        excluded_ids: 排除的 ID 集合
        order_attr: order 属性名
        id_attr: ID 属性名

    Returns:
        [(item, new_order), ...] 列表
    """
    excluded = excluded_ids or set()
    result: list[tuple[Any, int]] = []

    for item in items:
        item_id = getattr(item, id_attr)
        if item_id in excluded:
            continue
        current_order = getattr(item, order_attr)
        if current_order > threshold_order:
            result.append((item, current_order + shift))

    return result