xiaoxia-saas/packages/application/tts_job/use_cases.py

"""TTS Job use cases."""

from __future__ import annotations

from typing import List, Optional

from packages.application.tts_job.exceptions import TTSJobNotFoundError
from packages.domain.tts_job import TTSJob
from packages.ports.tts_job_repository import TTSJobRepository


class CreateTTSJobUseCase:
    """创建 TTS 合成任务。"""

    def __init__(self, repository: TTSJobRepository) -> None:
        self.repository = repository

    def execute(
        self,
        user_id: str,
        input_text: str,
        *,
        voice_id: str = "",
        voice_model: str = "",
        project_id: str = "",
        voice_clone_profile_id: str = "",
        sample_rate: int = 22050,
        format: str = "mp3",
        max_retries: int = 3,
        metadata: Optional[dict] = None,
    ) -> TTSJob:
        """创建 TTS 合成任务，状态为 pending。"""
        job = TTSJob.create(
            user_id=user_id,
            input_text=input_text,
            voice_id=voice_id,
            voice_model=voice_model,
            project_id=project_id,
            voice_clone_profile_id=voice_clone_profile_id,
            sample_rate=sample_rate,
            format=format,
            max_retries=max_retries,
            metadata=metadata,
        )
        return self.repository.create(job)


class ListTTSJobsUseCase:
    """列出用户的 TTS 合成任务。"""

    def __init__(self, repository: TTSJobRepository) -> None:
        self.repository = repository

    def execute(
        self,
        user_id: str,
        *,
        status: Optional[str] = None,
        skip: int = 0,
        limit: int = 50,
    ) -> tuple[List[TTSJob], int]:
        items = self.repository.list_by_user(user_id, status=status, limit=limit, offset=skip)
        total = self.repository.count_by_user(user_id, status=status)
        return items, total


class GetTTSJobUseCase:
    """获取 TTS 任务详情。"""

    def __init__(self, repository: TTSJobRepository) -> None:
        self.repository = repository

    def execute(self, job_id: str, user_id: str) -> TTSJob:
        job = self.repository.get(job_id)
        if job is None or job.user_id != user_id:
            raise TTSJobNotFoundError(f"TTS job {job_id} not found")
        return job


class GetTTSJobStatusUseCase:
    """查询 TTS 任务状态（用于轮询）。"""

    def __init__(self, repository: TTSJobRepository) -> None:
        self.repository = repository

    def execute(self, job_id: str, user_id: str) -> TTSJob:
        job = self.repository.get(job_id)
        if job is None or job.user_id != user_id:
            raise TTSJobNotFoundError(f"TTS job {job_id} not found")
        return job


class DeleteTTSJobUseCase:
    """删除 TTS 任务（软删除）。"""

    def __init__(self, repository: TTSJobRepository) -> None:
        self.repository = repository

    def execute(self, job_id: str, user_id: str) -> bool:
        job = self.repository.get(job_id)
        if job is None or job.user_id != user_id:
            return False
        return self.repository.delete(job_id)