Files
xiaoxia-saas/packages/domain/generation_task.py
T
xiaoxia 926d0fa272
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
Deploy / Deploy Staging (push) Has been cancelled
Deploy / Build Production Runtime Images (push) Has been cancelled
Deploy / Deploy Production (push) Has been cancelled
Deploy / Production Browser E2E (push) Has been cancelled
Tests / lint (push) Failing after 1926h42m54s
Tests / test (push) Failing after 1926h42m56s
fix: 为所有 ORM 模型添加外键约束及级联删除策略 (#65)
2026-06-28 08:47:51 +08:00

58 lines
1.7 KiB
Python

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import StrEnum
from uuid import uuid4
class GenerationTaskStatus(StrEnum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
@dataclass(slots=True)
class GenerationTask:
id: str
project_id: str
asset_library_id: str
strategy_id: str = ""
voice_library_id: str = ""
edit_plan_id: str = ""
status: GenerationTaskStatus = GenerationTaskStatus.PENDING
progress: float = 0.0
result_count: int = 0
error_message: str = ""
started_at: datetime | None = None
completed_at: datetime | None = None
created_by_user_id: str = ""
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@classmethod
def create(
cls,
project_id: str,
asset_library_id: str,
*,
strategy_id: str = "",
voice_library_id: str = "",
edit_plan_id: str = "",
created_by_user_id: str = "",
) -> "GenerationTask":
if not project_id.strip():
raise ValueError("project_id 不能为空")
if not asset_library_id.strip():
raise ValueError("asset_library_id 不能为空")
return cls(
id=uuid4().hex,
project_id=project_id.strip(),
asset_library_id=asset_library_id.strip(),
strategy_id=strategy_id.strip(),
voice_library_id=voice_library_id.strip(),
edit_plan_id=edit_plan_id.strip(),
created_by_user_id=created_by_user_id.strip(),
)