xiaoxia-saas/packages/application/generation_tasks.py

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)
    voice_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 = ""
    custom_title: 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=command.voice_ids,
            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,
            custom_title=command.custom_title,
        )
        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