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

"""Voice clone use cases."""

from __future__ import annotations

from typing import Optional

from packages.domain.voice_clone_profile import VoiceCloneProfile
from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository


class VoiceCloneNotFoundError(Exception):
    """音色克隆档案未找到。"""

    pass


class VoiceCloneNotRetryableError(Exception):
    """音色克隆档案不可重试。"""

    pass


class CreateVoiceCloneUseCase:
    """创建音色克隆档案。"""

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

    def execute(
        self,
        user_id: str,
        name: str,
        *,
        description: str = "",
        source_audio_url: str = "",
        voice_model: str = "",
        language: str = "zh-CN",
        gender: str = "unknown",
        max_retries: int = 3,
        metadata: Optional[dict] = None,
    ) -> VoiceCloneProfile:
        profile = VoiceCloneProfile.create(
            user_id=user_id,
            name=name,
            description=description,
            source_audio_url=source_audio_url,
            voice_model=voice_model,
            language=language,
            gender=gender,
            max_retries=max_retries,
            metadata=metadata,
        )
        return self.repository.create(profile)


class ListVoiceClonesUseCase:
    """列出用户的音色克隆档案。"""

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

    def execute(
        self,
        user_id: str,
        *,
        status: Optional[str] = None,
        skip: int = 0,
        limit: int = 50,
    ) -> tuple[list[VoiceCloneProfile], 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 GetVoiceCloneUseCase:
    """获取音色克隆档案详情。"""

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

    def execute(self, clone_id: str, user_id: str) -> VoiceCloneProfile:
        profile = self.repository.get(clone_id)
        if profile is None or profile.user_id != user_id:
            raise VoiceCloneNotFoundError(f"Voice clone {clone_id} not found")
        return profile


class GetVoiceCloneStatusUseCase:
    """查询音色克隆状态（用于轮询）。"""

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

    def execute(self, clone_id: str, user_id: str) -> VoiceCloneProfile:
        profile = self.repository.get(clone_id)
        if profile is None or profile.user_id != user_id:
            raise VoiceCloneNotFoundError(f"Voice clone {clone_id} not found")
        return profile


class DeleteVoiceCloneUseCase:
    """删除音色克隆档案（软删除）。"""

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

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


class RetryVoiceCloneUseCase:
    """重试失败的音色克隆。"""

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

    def execute(self, clone_id: str, user_id: str) -> VoiceCloneProfile:
        profile = self.repository.get(clone_id)
        if profile is None or profile.user_id != user_id:
            raise VoiceCloneNotFoundError(f"Voice clone {clone_id} not found")
        if not profile.is_retryable:
            raise VoiceCloneNotRetryableError(f"Voice clone {clone_id} is not retryable (status={profile.status})")
        profile.prepare_retry()
        return self.repository.update(profile)