saas-backend-agent
142 lines
4.8 KiB
Python
Executable File
142 lines
4.8 KiB
Python
Executable File
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from uuid import uuid4
|
||
|
||
from packages.domain import GenerationTask
|
||
from packages.ports.generation_task_repository import GenerationTaskRepository
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class CreateGenerationTaskCommand:
|
||
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)
|
||
created_by_user_id: str = ""
|
||
source_edit_plan_id: str = ""
|
||
asset_select_mode: str = ""
|
||
batch_id: str = ""
|
||
video_title: str = ""
|
||
resolution: str = ""
|
||
bgm_config: dict = field(default_factory=dict)
|
||
auto_retry_enabled: bool = False
|
||
auto_retry_max: int = 0
|
||
is_preview: bool = False
|
||
source_task_id: str = ""
|
||
output_width: int = 1280
|
||
output_height: int = 720
|
||
cover_url: str = ""
|
||
title_config: dict = field(default_factory=dict)
|
||
|
||
|
||
class CreateGenerationTaskUseCase:
|
||
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
||
self.generation_task_repository = generation_task_repository
|
||
|
||
def execute(self, command: CreateGenerationTaskCommand) -> GenerationTask:
|
||
task = GenerationTask(
|
||
id=uuid4().hex,
|
||
project_id=command.project_id,
|
||
asset_library_id=command.asset_library_id,
|
||
strategy_id=command.strategy_id,
|
||
voice_library_id=command.voice_library_id,
|
||
template_id=command.template_id,
|
||
asset_ids=command.asset_ids,
|
||
title_ids=command.title_ids,
|
||
voice_ids=[], # #1749：voice_ids 已废弃（冗余 voice_library_id），DB 列保留只读
|
||
status="pending", # type: ignore[arg-type]
|
||
progress=0.0,
|
||
result_count=0,
|
||
error_message="",
|
||
created_by_user_id=command.created_by_user_id,
|
||
source_edit_plan_id=command.source_edit_plan_id,
|
||
asset_select_mode=command.asset_select_mode,
|
||
batch_id=command.batch_id,
|
||
video_title=command.video_title,
|
||
resolution=command.resolution,
|
||
bgm_config=command.bgm_config,
|
||
auto_retry_enabled=command.auto_retry_enabled,
|
||
auto_retry_max=command.auto_retry_max,
|
||
is_preview=command.is_preview,
|
||
source_task_id=command.source_task_id,
|
||
output_width=command.output_width,
|
||
output_height=command.output_height,
|
||
cover_url=command.cover_url,
|
||
title_config=command.title_config,
|
||
)
|
||
return self.generation_task_repository.create(task)
|
||
|
||
|
||
class GetGenerationTaskUseCase:
|
||
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
||
self.generation_task_repository = generation_task_repository
|
||
|
||
def execute(self, task_id: str) -> GenerationTask | None:
|
||
return self.generation_task_repository.get(task_id)
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class ListTasksFilter:
|
||
"""任务列表筛选条件。"""
|
||
|
||
status: str | None = None # pending, running, completed, failed, cancelled
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class ListGenerationTasksResult:
|
||
"""带筛选和分页的任务列表结果。"""
|
||
|
||
items: list[GenerationTask]
|
||
total: int
|
||
|
||
|
||
class ListUserTasksFilteredUseCase:
|
||
"""按用户+筛选条件查询任务列表。"""
|
||
|
||
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
||
self.generation_task_repository = generation_task_repository
|
||
|
||
def execute(
|
||
self,
|
||
user_id: str,
|
||
*,
|
||
status: str | None = None,
|
||
limit: int | None = None,
|
||
offset: int = 0,
|
||
) -> ListGenerationTasksResult:
|
||
items = self.generation_task_repository.list_by_user_filtered(
|
||
user_id,
|
||
status=status,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
total = self.generation_task_repository.count_by_user_filtered(
|
||
user_id,
|
||
status=status,
|
||
)
|
||
return ListGenerationTasksResult(items=items, total=total)
|
||
|
||
|
||
class RetryGenerationTaskUseCase:
|
||
"""原地重试失败的任务（重置状态+递增retry_count）。
|
||
|
||
与创建新任务不同：复用同一个 task_id，保留历史关联。
|
||
"""
|
||
|
||
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
||
self.generation_task_repository = generation_task_repository
|
||
|
||
def execute(self, task_id: str) -> GenerationTask:
|
||
task = self.generation_task_repository.get(task_id)
|
||
if task is None:
|
||
raise ValueError(f"任务不存在: {task_id}")
|
||
if not task.is_failed:
|
||
raise ValueError(f"只有失败状态的任务才能重试，当前状态: {task.status.value}")
|
||
task.mark_pending_from_failed()
|
||
self.generation_task_repository.update(task)
|
||
return task
|