灵应
8935196fcd
Deploy / Staging E2E Tests (push) Has been skipped
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 138h4m33s
CI/CD Pipeline / Frontend Lint (push) Failing after 138h4m39s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 138h4m39s
129 lines
3.9 KiB
Python
129 lines
3.9 KiB
Python
"""Voice clone use cases."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import List, 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)
|