xiaoxia-saas/packages/domain/edit_plan.py

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

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 EditPlanStatus(StrEnum):
    """剪辑计划状态"""

    DRAFT = "draft"
    EDITING = "editing"
    RENDERING = "rendering"
    COMPLETED = "completed"
    FAILED = "failed"

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

        - success/done/finished/complete → COMPLETED
        - fail/error/err → FAILED
        - render/rendering → RENDERING
        - edit/editing → EDITING
        - 其他未知值 → DRAFT（兜底，不阻塞业务）
        """
        if isinstance(value, str):
            normalized = value.strip().lower()
            if normalized in ("done", "success", "finished", "complete", "completed"):
                return cls.COMPLETED
            if normalized in ("fail", "failed", "error", "err"):
                return cls.FAILED
            if normalized in ("render", "rendering", "generating", "generating_video"):
                return cls.RENDERING
            if normalized in ("edit", "editing", "working"):
                return cls.EDITING
        return cls.DRAFT


@dataclass(slots=True)
class EditPlan:
    """Phase 8 剪辑计划实体

    基于某个 EditTemplate 创建的剪辑计划，包含具体的配置和状态追踪。
    状态流转：draft → editing → rendering → completed / failed
    """

    id: str
    template_id: str
    name: str
    status: EditPlanStatus = EditPlanStatus.DRAFT
    total_duration: float = 0.0
    source_edit_plan_id: str = ""
    project_id: str = ""
    created_by_user_id: str = ""
    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,
        template_id: str,
        name: str,
        *,
        config: dict[str, Any] | None = None,
        total_duration: float = 0.0,
        source_edit_plan_id: str = "",
        project_id: str = "",
        created_by_user_id: str = "",
    ) -> EditPlan:
        """创建新剪辑计划实例"""
        clean_name = name.strip()
        if not clean_name:
            raise ValueError("计划名称不能为空")
        if not template_id.strip():
            raise ValueError("template_id 不能为空")
        return cls(
            id=uuid4().hex,
            template_id=template_id.strip(),
            name=clean_name,
            status=EditPlanStatus.DRAFT,
            total_duration=total_duration,
            source_edit_plan_id=source_edit_plan_id.strip(),
            project_id=project_id.strip(),
            created_by_user_id=created_by_user_id.strip(),
            config=config or {},
        )

    def start_editing(self) -> None:
        """开始编辑"""
        if self.status != EditPlanStatus.DRAFT:
            raise ValueError(f"只有 draft 状态的计划可以开始编辑，当前状态: {self.status}")
        self.status = EditPlanStatus.EDITING
        self.updated_at = datetime.now(UTC)

    def start_rendering(self) -> None:
        """开始渲染"""
        if self.status != EditPlanStatus.EDITING:
            raise ValueError(f"只有 editing 状态的计划可以开始渲染，当前状态: {self.status}")
        self.status = EditPlanStatus.RENDERING
        self.updated_at = datetime.now(UTC)

    def mark_completed(self) -> None:
        """标记为完成"""
        if self.status != EditPlanStatus.RENDERING:
            raise ValueError(f"只有 rendering 状态的计划可以标记完成，当前状态: {self.status}")
        self.status = EditPlanStatus.COMPLETED
        self.updated_at = datetime.now(UTC)

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

    def resume_editing(self) -> None:
        """重新进入编辑状态（完成/失败后重新编辑）"""
        if self.status not in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED):
            raise ValueError(f"只有 completed/failed 状态的计划可以重新编辑，当前状态: {self.status}")
        self.status = EditPlanStatus.EDITING
        self.updated_at = datetime.now(UTC)

    def reset_to_draft(self) -> None:
        """重置为草稿状态（仅从 failed 状态可重置）"""
        if self.status != EditPlanStatus.FAILED:
            raise ValueError(f"只有 failed 状态的计划可以重置，当前状态: {self.status}")
        self.status = EditPlanStatus.DRAFT
        self.updated_at = datetime.now(UTC)