灵应
356df4663e
Deploy / Staging E2E Tests (push) Has been cancelled
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
Deploy / Deploy Staging (push) Failing after 111h7m48s
CI/CD Pipeline / Frontend Lint (push) Failing after 111h7m57s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 111h7m57s
Task #105: 修复剪辑计划/模板页面报服务器繁忙 - templates.py: list_templates/get_template/list_categories 加 try/except - templates.py: 新增 POST /{template_id}/toggle-favorite 兼容端点 - edit_plans.py: list_plans 加 try/except，ai_tasks 加 ImportError 守卫 Task #106: 修复素材库诊断按钮 + 缩略图/视频URL - asset_diagnosis.py: get_project_asset_diagnosis 加 try/except - assets.py: 注入 storage_service，生成签名 file_url - asset.py schema: 新增 file_url 字段 - 视频素材 thumbnail_url 为空时复用 file_url 作为封面 其他: - edit_plans/generation_tasks 支持 source_edit_plan_id - Alembic 迁移 022: 两表加 source_edit_plan_id 字段
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
|
|
if sys.version_info >= (3, 11):
|
|
from enum import StrEnum
|
|
else:
|
|
from enum import Enum
|
|
|
|
class StrEnum(str, Enum):
|
|
pass
|
|
|
|
|
|
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 = ""
|
|
template_id: str = ""
|
|
asset_ids: list[str] = field(default_factory=list)
|
|
title_ids: list[str] = field(default_factory=list)
|
|
voice_ids: list[str] = field(default_factory=list)
|
|
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
|
|
source_edit_plan_id: str = ""
|
|
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 = "",
|
|
template_id: str = "",
|
|
asset_ids: list[str] | None = None,
|
|
title_ids: list[str] | None = None,
|
|
voice_ids: list[str] | None = None,
|
|
created_by_user_id: str = "",
|
|
source_edit_plan_id: str = "",
|
|
) -> "GenerationTask":
|
|
if not project_id.strip() and not template_id.strip():
|
|
raise ValueError("project_id 或 template_id 至少需要提供一个")
|
|
if not asset_library_id.strip() and not (asset_ids or title_ids or voice_ids):
|
|
raise ValueError("asset_library_id 或 asset_ids/title_ids/voice_ids 至少需要提供一个")
|
|
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(),
|
|
template_id=template_id.strip(),
|
|
asset_ids=list(asset_ids) if asset_ids else [],
|
|
title_ids=list(title_ids) if title_ids else [],
|
|
voice_ids=list(voice_ids) if voice_ids else [],
|
|
created_by_user_id=created_by_user_id.strip(),
|
|
source_edit_plan_id=source_edit_plan_id.strip(),
|
|
)
|