xiaoxia-saas/packages/domain/edit_plan_clip.py

"""EditPlanClip domain entity for Phase 8 模板编排引擎.

剪辑计划中的具体片段实例，关联 EditPlan 和 TemplateClipConfig，
包含实际素材、实际文案、排序和时长等信息。
"""

from __future__ import annotations

import sys
from dataclasses import dataclass, field
from datetime import UTC, datetime

if sys.version_info >= (3, 11):
    from enum import StrEnum
else:
    from enum import Enum

    class StrEnum(str, Enum):
        pass


from typing import Any
from uuid import uuid4


class EditPlanClipStatus(StrEnum):
    """片段状态"""

    PENDING = "pending"  # 待处理（素材未就绪）
    READY = "ready"  # 就绪（素材已就绪，可渲染）
    RENDERED = "rendered"  # 已渲染
    FAILED = "failed"  # 渲染失败

    @classmethod
    def _missing_(cls, value: object) -> "EditPlanClipStatus":
        """兼容历史脏数据，避免枚举转换失败导致500。

        - success/done/finished/complete/rendered → RENDERED
        - fail/error/err → FAILED
        - ready/available → READY
        - 其他未知值 → PENDING（兜底，不阻塞业务）
        """
        if isinstance(value, str):
            normalized = value.strip().lower()
            if normalized in ("done", "success", "finished", "complete", "rendered", "render"):
                return cls.RENDERED
            if normalized in ("fail", "failed", "error", "err"):
                return cls.FAILED
            if normalized in ("ready", "available", "prepared"):
                return cls.READY
        return cls.PENDING


@dataclass(slots=True)
class EditPlanClip:
    """剪辑计划片段

    表示 EditPlan 中的一个具体片段实例，包含实际素材、文案和渲染状态。
    可选地关联 TemplateClipConfig 以继承模板规则。
    """

    id: str
    plan_id: str
    clip_type: str
    order: int
    template_clip_config_id: str = ""
    asset_id: str = ""
    atom_clip_id: str = ""
    text_content: str = ""
    start_time: float = 0.0
    duration: float = 0.0
    transition_effect: str = "cut"
    transition_duration: float = 0.0  # 0 表示使用全局默认值
    playback_speed: float = 1.0  # 0 或 1.0 表示原速，范围 0.25~4.0
    status: EditPlanClipStatus = EditPlanClipStatus.PENDING
    config: dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    @classmethod
    def create(
        cls,
        plan_id: str,
        clip_type: str,
        order: int,
        *,
        template_clip_config_id: str = "",
        asset_id: str = "",
        atom_clip_id: str = "",
        text_content: str = "",
        start_time: float = 0.0,
        duration: float = 0.0,
        transition_effect: str = "cut",
        transition_duration: float = 0.0,
        playback_speed: float = 1.0,
        config: dict[str, Any] | None = None,
    ) -> EditPlanClip:
        """创建剪辑计划片段"""
        if not plan_id.strip():
            raise ValueError("plan_id 不能为空")
        if not clip_type.strip():
            raise ValueError("clip_type 不能为空")
        if start_time < 0:
            raise ValueError("start_time 不能为负数")
        if duration < 0:
            raise ValueError("duration 不能为负数")
        # 速度边界钳制
        if playback_speed <= 0:
            playback_speed = 1.0
        elif playback_speed < 0.25:
            playback_speed = 0.25
        elif playback_speed > 4.0:
            playback_speed = 4.0

        return cls(
            id=uuid4().hex,
            plan_id=plan_id.strip(),
            clip_type=clip_type.strip(),
            order=order,
            template_clip_config_id=template_clip_config_id.strip() if template_clip_config_id else "",
            asset_id=asset_id.strip() if asset_id else "",
            atom_clip_id=atom_clip_id.strip() if atom_clip_id else "",
            text_content=text_content.strip(),
            start_time=start_time,
            duration=duration,
            transition_effect=transition_effect.strip() or "cut",
            transition_duration=max(0.0, transition_duration),
            playback_speed=playback_speed,
            status=EditPlanClipStatus.PENDING,
            config=config or {},
        )

    def assign_asset(
        self,
        asset_id: str,
        *,
        start_time: float | None = None,
        atom_clip_id: str | None = None,
    ) -> None:
        """分配素材

        Args:
            asset_id: 素材 ID
            start_time: 可选，素材播放起始时间（秒）。如果提供且在有效范围内，则设置；否则保持默认 0.0
            atom_clip_id: 可选，选中的原子片段 ID（#1970 原子化切片）。
        """
        if not asset_id.strip():
            raise ValueError("asset_id 不能为空")
        self.asset_id = asset_id.strip()
        if atom_clip_id is not None:
            self.atom_clip_id = atom_clip_id.strip() if atom_clip_id else ""
        if start_time is not None and start_time >= 0:
            self.start_time = start_time
        self.updated_at = datetime.now(UTC)

    def mark_ready(self) -> None:
        """标记为就绪"""
        if self.status != EditPlanClipStatus.PENDING:
            raise ValueError(f"只有 pending 状态的片段可以标记就绪，当前状态: {self.status}")
        self.status = EditPlanClipStatus.READY
        self.updated_at = datetime.now(UTC)

    def mark_rendered(self) -> None:
        """标记为已渲染"""
        if self.status != EditPlanClipStatus.READY:
            raise ValueError(f"只有 ready 状态的片段可以标记已渲染，当前状态: {self.status}")
        self.status = EditPlanClipStatus.RENDERED
        self.updated_at = datetime.now(UTC)

    def mark_failed(self) -> None:
        """标记为失败"""
        if self.status != EditPlanClipStatus.READY:
            raise ValueError(f"只有 ready 状态的片段可以标记失败，当前状态: {self.status}")
        self.status = EditPlanClipStatus.FAILED
        self.updated_at = datetime.now(UTC)

    @property
    def end_time(self) -> float:
        """片段结束时间"""
        return self.start_time + self.duration

    @property
    def has_asset(self) -> bool:
        """是否已分配素材"""
        return bool(self.asset_id)