diff --git a/alembic/versions/019_add_voice_clone_profiles_table.py b/alembic/versions/019_add_voice_clone_profiles_table.py new file mode 100644 index 000000000..0d2c6d342 --- /dev/null +++ b/alembic/versions/019_add_voice_clone_profiles_table.py @@ -0,0 +1,52 @@ +"""Task 3.05: Create voice_clone_profiles table + +Revision ID: 019 +Revises: 018 +Create Date: 2026-07-02 + +新增 voice_clone_profiles 表，用于存储音色克隆档案。 +""" + +from alembic import op +import sqlalchemy as sa + +revision = "019" +down_revision = "018" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "voice_clone_profiles", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("user_id", sa.String(36), nullable=False, index=True), + sa.Column("name", sa.String(100), nullable=False), + sa.Column("description", sa.Text(), nullable=False, server_default=""), + sa.Column("source_audio_url", sa.Text(), nullable=False, server_default=""), + sa.Column("voice_id", sa.String(100), nullable=False, server_default=""), + sa.Column("voice_model", sa.String(100), nullable=False, server_default=""), + sa.Column("language", sa.String(20), nullable=False, server_default="zh-CN"), + sa.Column("gender", sa.String(20), nullable=False, server_default="unknown"), + sa.Column("status", sa.String(20), nullable=False, server_default="pending", index=True), + sa.Column("error_message", sa.Text(), nullable=False, server_default=""), + sa.Column("retry_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("max_retries", sa.Integer(), nullable=False, server_default="3"), + sa.Column("metadata", sa.JSON(), nullable=False, server_default="{}"), + sa.Column( + "created_at", + sa.DateTime(), + nullable=False, + server_default=sa.func.now(), + ), + sa.Column( + "updated_at", + sa.DateTime(), + nullable=False, + server_default=sa.func.now(), + ), + ) + + +def downgrade() -> None: + op.drop_table("voice_clone_profiles") diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index 6a121b4c5..1b814ac7c 100755 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -20,6 +20,7 @@ from app.api.routes.task_center import router as task_center_router from app.api.routes.templates import router as templates_router from app.api.routes.titles import router as titles_router from app.api.routes.upload import router as upload_router +from app.api.routes.voice_clones import router as voice_clones_router from app.api.routes.voices import router as voices_router from fastapi import APIRouter @@ -98,6 +99,11 @@ api_router.include_router( prefix="/voices", tags=["VoiceLibrary"], ) +api_router.include_router( + voice_clones_router, + prefix="/voice-clones", + tags=["VoiceClone"], +) api_router.include_router( duplication_router, prefix="/duplication", diff --git a/apps/api/app/api/routes/voice_clones.py b/apps/api/app/api/routes/voice_clones.py new file mode 100644 index 000000000..ea8935246 --- /dev/null +++ b/apps/api/app/api/routes/voice_clones.py @@ -0,0 +1,180 @@ +"""音色克隆 API 路由。""" + +from __future__ import annotations + +from typing import Optional + +from app.auth import AuthenticatedUser, get_current_user +from app.dependencies import get_db_session +from app.schemas.voice_clone import ( + CreateVoiceCloneRequest, + ListVoiceCloneResponse, + VoiceCloneProfileResponse, + VoiceCloneStatusResponse, +) +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import ( + SQLAlchemyVoiceCloneProfileRepository, +) +from packages.application.voice_clone.use_cases import ( + CreateVoiceCloneUseCase, + DeleteVoiceCloneUseCase, + GetVoiceCloneStatusUseCase, + GetVoiceCloneUseCase, + ListVoiceClonesUseCase, + RetryVoiceCloneUseCase, + VoiceCloneNotFoundError, + VoiceCloneNotRetryableError, +) + +router = APIRouter() + + +def _get_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyVoiceCloneProfileRepository: + return SQLAlchemyVoiceCloneProfileRepository(session) + + +def _to_response(profile) -> VoiceCloneProfileResponse: + return VoiceCloneProfileResponse( + id=profile.id, + user_id=profile.user_id, + name=profile.name, + description=profile.description, + source_audio_url=profile.source_audio_url, + voice_id=profile.voice_id, + voice_model=profile.voice_model, + language=profile.language, + gender=profile.gender, + status=profile.status, + error_message=profile.error_message, + retry_count=profile.retry_count, + max_retries=profile.max_retries, + metadata=profile.metadata, + created_at=profile.created_at, + updated_at=profile.updated_at, + ) + + +@router.post("", response_model=VoiceCloneProfileResponse, status_code=status.HTTP_201_CREATED) +def create_voice_clone( + request: CreateVoiceCloneRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository), +) -> VoiceCloneProfileResponse: + """创建音色克隆任务。 + + 新建的克隆任务状态为 pending，等待后续 CosyVoice API 调用。 + """ + user_id = authenticated_user.user.id + use_case = CreateVoiceCloneUseCase(repository) + profile = use_case.execute( + user_id=user_id, + name=request.name, + description=request.description, + source_audio_url=request.source_audio_url, + voice_model=request.voice_model, + language=request.language, + gender=request.gender, + max_retries=request.max_retries, + metadata=request.metadata_, + ) + return _to_response(profile) + + +@router.get("", response_model=ListVoiceCloneResponse) +def list_voice_clones( + status_filter: Optional[str] = Query(None, alias="status"), + skip: int = Query(0, ge=0), + limit: int = Query(50, ge=1, le=200), + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository), +) -> ListVoiceCloneResponse: + """获取用户的音色克隆列表。""" + user_id = authenticated_user.user.id + use_case = ListVoiceClonesUseCase(repository) + items, total = use_case.execute( + user_id, status=status_filter, skip=skip, limit=limit + ) + return ListVoiceCloneResponse( + items=[_to_response(p) for p in items], + total=total, + ) + + +@router.get("/{clone_id}", response_model=VoiceCloneProfileResponse) +def get_voice_clone( + clone_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository), +) -> VoiceCloneProfileResponse: + """获取音色克隆详情。""" + user_id = authenticated_user.user.id + use_case = GetVoiceCloneUseCase(repository) + try: + profile = use_case.execute(clone_id, user_id) + except VoiceCloneNotFoundError: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") + return _to_response(profile) + + +@router.get("/{clone_id}/status", response_model=VoiceCloneStatusResponse) +def get_voice_clone_status( + clone_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository), +) -> VoiceCloneStatusResponse: + """查询音色克隆状态（用于前端轮询）。""" + user_id = authenticated_user.user.id + use_case = GetVoiceCloneStatusUseCase(repository) + try: + profile = use_case.execute(clone_id, user_id) + except VoiceCloneNotFoundError: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") + return VoiceCloneStatusResponse( + id=profile.id, + status=profile.status, + error_message=profile.error_message, + voice_id=profile.voice_id, + retry_count=profile.retry_count, + ) + + +@router.delete("/{clone_id}", status_code=status.HTTP_204_NO_CONTENT, response_class=Response) +def delete_voice_clone( + clone_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository), +) -> Response: + """删除音色克隆档案。""" + user_id = authenticated_user.user.id + use_case = DeleteVoiceCloneUseCase(repository) + deleted = use_case.execute(clone_id, user_id) + if not deleted: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") + return Response(status_code=204) + + +@router.post("/{clone_id}/retry", response_model=VoiceCloneProfileResponse) +def retry_voice_clone( + clone_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository), +) -> VoiceCloneProfileResponse: + """重试失败的音色克隆。 + + 仅当状态为 failed 时可重试，重试后状态变为 pending。 + """ + user_id = authenticated_user.user.id + use_case = RetryVoiceCloneUseCase(repository) + try: + profile = use_case.execute(clone_id, user_id) + except VoiceCloneNotFoundError: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") + except VoiceCloneNotRetryableError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Voice clone is not retryable (only failed clones can be retried)", + ) + return _to_response(profile) diff --git a/apps/api/app/schemas/voice_clone.py b/apps/api/app/schemas/voice_clone.py new file mode 100644 index 000000000..a1f7b7de5 --- /dev/null +++ b/apps/api/app/schemas/voice_clone.py @@ -0,0 +1,69 @@ +"""音色克隆 API Schema。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + + +class CreateVoiceCloneRequest(BaseModel): + """创建音色克隆请求。""" + + name: str = Field(..., min_length=1, max_length=100, description="音色名称") + description: str = Field("", description="音色描述") + source_audio_url: str = Field("", description="参考音频 URL") + voice_model: str = Field("", description="语音模型名称") + language: str = Field("zh-CN", description="语言") + gender: str = Field("unknown", description="性别") + max_retries: int = Field(3, ge=1, le=10, description="最大重试次数") + metadata_: Optional[Dict[str, Any]] = Field( + default=None, alias="metadata", description="额外元数据" + ) + + class Config: + populate_by_name = True + + +class VoiceCloneProfileResponse(BaseModel): + """音色克隆档案响应。""" + + id: str + user_id: str + name: str + description: str = "" + source_audio_url: str = "" + voice_id: str = "" + voice_model: str = "" + language: str = "zh-CN" + gender: str = "unknown" + status: str + error_message: str = "" + retry_count: int = 0 + max_retries: int = 3 + metadata_: Optional[Dict[str, Any]] = Field( + default=None, alias="metadata", description="额外元数据" + ) + created_at: datetime + updated_at: datetime + + class Config: + populate_by_name = True + + +class VoiceCloneStatusResponse(BaseModel): + """音色克隆状态响应（用于轮询）。""" + + id: str + status: str + error_message: str = "" + voice_id: str = "" + retry_count: int = 0 + + +class ListVoiceCloneResponse(BaseModel): + """音色克隆列表响应。""" + + items: List[VoiceCloneProfileResponse] + total: int diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 7519d0ea9..99497f8fb 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -267,6 +267,29 @@ class TitleLibraryModel(Base): updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) +class VoiceCloneProfileModel(Base): + """音色克隆档案 ORM 模型。""" + + __tablename__ = "voice_clone_profiles" + + id = Column(String(36), primary_key=True) + user_id = Column(String(36), nullable=False, index=True) + name = Column(String(100), nullable=False) + description = Column(Text, nullable=False, default="") + source_audio_url = Column(Text, nullable=False, default="") + voice_id = Column(String(100), nullable=False, default="") + voice_model = Column(String(100), nullable=False, default="") + language = Column(String(20), nullable=False, default="zh-CN") + gender = Column(String(20), nullable=False, default="unknown") + status = Column(String(20), nullable=False, default="pending", index=True) + error_message = Column(Text, nullable=False, default="") + retry_count = Column(Integer, nullable=False, default=0) + max_retries = Column(Integer, nullable=False, default=3) + metadata_ = Column("metadata", JSON, nullable=False, default=dict) + created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + + class VoiceLibraryModel(Base): __tablename__ = "voice_libraries" diff --git a/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py b/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py new file mode 100644 index 000000000..fa1a37f16 --- /dev/null +++ b/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py @@ -0,0 +1,152 @@ +"""SQLAlchemy implementation of VoiceCloneProfileRepository.""" + +from __future__ import annotations + +from typing import List, Optional + +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.models import VoiceCloneProfileModel +from packages.domain.voice_clone_profile import VoiceCloneProfile + + +class SQLAlchemyVoiceCloneProfileRepository: + """SQLAlchemy 音色克隆档案仓储。""" + + def __init__(self, session: Session) -> None: + self.session = session + + def create(self, profile: VoiceCloneProfile) -> VoiceCloneProfile: + model = VoiceCloneProfileModel( + id=profile.id, + user_id=profile.user_id, + name=profile.name, + description=profile.description, + source_audio_url=profile.source_audio_url, + voice_id=profile.voice_id, + voice_model=profile.voice_model, + language=profile.language, + gender=profile.gender, + status=profile.status, + error_message=profile.error_message, + retry_count=profile.retry_count, + max_retries=profile.max_retries, + metadata_=profile.metadata, + ) + self.session.add(model) + self.session.commit() + self.session.refresh(model) + return self._model_to_entity(model) + + def get(self, profile_id: str) -> Optional[VoiceCloneProfile]: + model = ( + self.session.query(VoiceCloneProfileModel) + .filter( + VoiceCloneProfileModel.id == profile_id, + VoiceCloneProfileModel.status != "deleted", + ) + .first() + ) + if model is None: + return None + return self._model_to_entity(model) + + def update(self, profile: VoiceCloneProfile) -> VoiceCloneProfile: + model = ( + self.session.query(VoiceCloneProfileModel) + .filter(VoiceCloneProfileModel.id == profile.id) + .first() + ) + if model is None: + raise ValueError(f"VoiceCloneProfile {profile.id} not found") + model.name = profile.name + model.description = profile.description + model.source_audio_url = profile.source_audio_url + model.voice_id = profile.voice_id + model.voice_model = profile.voice_model + model.language = profile.language + model.gender = profile.gender + model.status = profile.status + model.error_message = profile.error_message + model.retry_count = profile.retry_count + model.max_retries = profile.max_retries + model.metadata_ = profile.metadata + self.session.commit() + self.session.refresh(model) + return self._model_to_entity(model) + + def delete(self, profile_id: str) -> bool: + model = ( + self.session.query(VoiceCloneProfileModel) + .filter(VoiceCloneProfileModel.id == profile_id) + .first() + ) + if model is None: + return False + model.status = "deleted" + self.session.commit() + return True + + def list_by_user( + self, + user_id: str, + *, + status: Optional[str] = None, + limit: int = 50, + offset: int = 0, + ) -> List[VoiceCloneProfile]: + query = self.session.query(VoiceCloneProfileModel).filter( + VoiceCloneProfileModel.user_id == user_id, + VoiceCloneProfileModel.status != "deleted", + ) + if status: + query = query.filter(VoiceCloneProfileModel.status == status) + query = query.order_by(VoiceCloneProfileModel.created_at.desc()) + models = query.offset(offset).limit(limit).all() + return [self._model_to_entity(m) for m in models] + + def count_by_user(self, user_id: str, *, status: Optional[str] = None) -> int: + query = ( + self.session.query(VoiceCloneProfileModel) + .filter( + VoiceCloneProfileModel.user_id == user_id, + VoiceCloneProfileModel.status != "deleted", + ) + ) + if status: + query = query.filter(VoiceCloneProfileModel.status == status) + return query.count() + + def find_by_voice_id(self, voice_id: str) -> Optional[VoiceCloneProfile]: + model = ( + self.session.query(VoiceCloneProfileModel) + .filter( + VoiceCloneProfileModel.voice_id == voice_id, + VoiceCloneProfileModel.status != "deleted", + ) + .first() + ) + if model is None: + return None + return self._model_to_entity(model) + + @staticmethod + def _model_to_entity(model: VoiceCloneProfileModel) -> VoiceCloneProfile: + return VoiceCloneProfile( + id=model.id, + user_id=model.user_id, + name=model.name, + description=model.description or "", + source_audio_url=model.source_audio_url or "", + voice_id=model.voice_id or "", + voice_model=model.voice_model or "", + language=model.language or "zh-CN", + gender=model.gender or "unknown", + status=model.status, + error_message=model.error_message or "", + retry_count=model.retry_count or 0, + max_retries=model.max_retries or 3, + metadata=model.metadata_ or {}, + created_at=model.created_at, + updated_at=model.updated_at, + ) diff --git a/packages/application/voice_clone/__init__.py b/packages/application/voice_clone/__init__.py new file mode 100644 index 000000000..095d7e950 --- /dev/null +++ b/packages/application/voice_clone/__init__.py @@ -0,0 +1,23 @@ +"""Voice clone application package.""" + +from packages.application.voice_clone.use_cases import ( + CreateVoiceCloneUseCase, + DeleteVoiceCloneUseCase, + GetVoiceCloneStatusUseCase, + GetVoiceCloneUseCase, + ListVoiceClonesUseCase, + RetryVoiceCloneUseCase, + VoiceCloneNotFoundError, + VoiceCloneNotRetryableError, +) + +__all__ = [ + "CreateVoiceCloneUseCase", + "DeleteVoiceCloneUseCase", + "GetVoiceCloneStatusUseCase", + "GetVoiceCloneUseCase", + "ListVoiceClonesUseCase", + "RetryVoiceCloneUseCase", + "VoiceCloneNotFoundError", + "VoiceCloneNotRetryableError", +] diff --git a/packages/application/voice_clone/use_cases.py b/packages/application/voice_clone/use_cases.py new file mode 100644 index 000000000..f8adc4e48 --- /dev/null +++ b/packages/application/voice_clone/use_cases.py @@ -0,0 +1,132 @@ +"""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) diff --git a/tests/unit/test_voice_clone_api.py b/tests/unit/test_voice_clone_api.py new file mode 100644 index 000000000..3a9800d03 --- /dev/null +++ b/tests/unit/test_voice_clone_api.py @@ -0,0 +1,293 @@ +"""音色克隆 API 单元测试。""" + +from __future__ import annotations + +import pytest +from datetime import datetime, timezone +from unittest.mock import MagicMock + +from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus +from packages.application.voice_clone.use_cases import ( + CreateVoiceCloneUseCase, + DeleteVoiceCloneUseCase, + GetVoiceCloneStatusUseCase, + GetVoiceCloneUseCase, + ListVoiceClonesUseCase, + RetryVoiceCloneUseCase, + VoiceCloneNotFoundError, + VoiceCloneNotRetryableError, +) + + +def _make_profile(**kwargs) -> VoiceCloneProfile: + defaults = { + "id": "test_clone_001", + "user_id": "user_001", + "name": "测试音色", + "description": "测试描述", + "source_audio_url": "", + "voice_id": "", + "voice_model": "", + "language": "zh-CN", + "gender": "unknown", + "status": VoiceCloneStatus.PENDING, + "error_message": "", + "retry_count": 0, + "max_retries": 3, + "metadata": {}, + "created_at": datetime.now(timezone.utc), + "updated_at": datetime.now(timezone.utc), + } + defaults.update(kwargs) + return VoiceCloneProfile(**defaults) + + +class TestCreateVoiceCloneUseCase: + """测试创建音色克隆。""" + + def test_create_success(self) -> None: + """正常创建。""" + repo = MagicMock() + repo.create.side_effect = lambda p: p + + use_case = CreateVoiceCloneUseCase(repo) + profile = use_case.execute( + user_id="user_001", + name="我的音色", + description="克隆音色", + source_audio_url="https://example.com/audio.mp3", + ) + + assert profile.user_id == "user_001" + assert profile.name == "我的音色" + assert profile.status == VoiceCloneStatus.PENDING + repo.create.assert_called_once() + + def test_create_with_metadata(self) -> None: + """带元数据创建。""" + repo = MagicMock() + repo.create.side_effect = lambda p: p + + use_case = CreateVoiceCloneUseCase(repo) + profile = use_case.execute( + user_id="user_001", + name="测试", + metadata={"source": "upload"}, + ) + + assert profile.metadata == {"source": "upload"} + + +class TestListVoiceClonesUseCase: + """测试列出音色克隆。""" + + def test_list_empty(self) -> None: + """空列表。""" + repo = MagicMock() + repo.list_by_user.return_value = [] + repo.count_by_user.return_value = 0 + + use_case = ListVoiceClonesUseCase(repo) + items, total = use_case.execute("user_001") + + assert items == [] + assert total == 0 + repo.list_by_user.assert_called_once_with("user_001", status=None, limit=50, offset=0) + + def test_list_with_pagination(self) -> None: + """分页查询。""" + repo = MagicMock() + profiles = [_make_profile(id=f"clone_{i}") for i in range(3)] + repo.list_by_user.return_value = profiles + repo.count_by_user.return_value = 10 + + use_case = ListVoiceClonesUseCase(repo) + items, total = use_case.execute("user_001", skip=5, limit=3) + + assert len(items) == 3 + assert total == 10 + repo.list_by_user.assert_called_once_with("user_001", status=None, limit=3, offset=5) + + def test_list_with_status_filter(self) -> None: + """按状态过滤。""" + repo = MagicMock() + repo.list_by_user.return_value = [] + repo.count_by_user.return_value = 0 + + use_case = ListVoiceClonesUseCase(repo) + use_case.execute("user_001", status="pending") + + repo.list_by_user.assert_called_once_with("user_001", status="pending", limit=50, offset=0) + + +class TestGetVoiceCloneUseCase: + """测试获取音色克隆详情。""" + + def test_get_success(self) -> None: + """正常获取。""" + profile = _make_profile() + repo = MagicMock() + repo.get.return_value = profile + + use_case = GetVoiceCloneUseCase(repo) + result = use_case.execute("test_clone_001", "user_001") + + assert result.id == "test_clone_001" + assert result.user_id == "user_001" + + def test_get_not_found(self) -> None: + """档案不存在。""" + repo = MagicMock() + repo.get.return_value = None + + use_case = GetVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("nonexistent", "user_001") + + def test_get_wrong_user(self) -> None: + """用户不匹配。""" + profile = _make_profile(user_id="other_user") + repo = MagicMock() + repo.get.return_value = profile + + use_case = GetVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("test_clone_001", "user_001") + + +class TestGetVoiceCloneStatusUseCase: + """测试查询音色克隆状态。""" + + def test_status_success(self) -> None: + """正常查询状态。""" + profile = _make_profile(status=VoiceCloneStatus.PROCESSING) + repo = MagicMock() + repo.get.return_value = profile + + use_case = GetVoiceCloneStatusUseCase(repo) + result = use_case.execute("test_clone_001", "user_001") + + assert result.status == VoiceCloneStatus.PROCESSING + + def test_status_not_found(self) -> None: + """档案不存在。""" + repo = MagicMock() + repo.get.return_value = None + + use_case = GetVoiceCloneStatusUseCase(repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("nonexistent", "user_001") + + +class TestDeleteVoiceCloneUseCase: + """测试删除音色克隆。""" + + def test_delete_success(self) -> None: + """正常删除。""" + profile = _make_profile() + repo = MagicMock() + repo.get.return_value = profile + repo.delete.return_value = True + + use_case = DeleteVoiceCloneUseCase(repo) + result = use_case.execute("test_clone_001", "user_001") + + assert result is True + repo.delete.assert_called_once_with("test_clone_001") + + def test_delete_not_found(self) -> None: + """档案不存在。""" + repo = MagicMock() + repo.get.return_value = None + + use_case = DeleteVoiceCloneUseCase(repo) + result = use_case.execute("nonexistent", "user_001") + + assert result is False + + def test_delete_wrong_user(self) -> None: + """用户不匹配。""" + profile = _make_profile(user_id="other_user") + repo = MagicMock() + repo.get.return_value = profile + + use_case = DeleteVoiceCloneUseCase(repo) + result = use_case.execute("test_clone_001", "user_001") + + assert result is False + + +class TestRetryVoiceCloneUseCase: + """测试重试音色克隆。""" + + def test_retry_success(self) -> None: + """正常重试（failed → pending）。""" + profile = _make_profile(status=VoiceCloneStatus.FAILED, error_message="API error") + repo = MagicMock() + repo.get.return_value = profile + repo.update.side_effect = lambda p: p + + use_case = RetryVoiceCloneUseCase(repo) + result = use_case.execute("test_clone_001", "user_001") + + assert result.status == VoiceCloneStatus.PENDING + assert result.error_message == "" + assert result.retry_count == 1 + repo.update.assert_called_once() + + def test_retry_not_found(self) -> None: + """档案不存在。""" + repo = MagicMock() + repo.get.return_value = None + + use_case = RetryVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("nonexistent", "user_001") + + def test_retry_not_retryable_pending(self) -> None: + """pending 状态不可重试。""" + profile = _make_profile(status=VoiceCloneStatus.PENDING) + repo = MagicMock() + repo.get.return_value = profile + + use_case = RetryVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotRetryableError): + use_case.execute("test_clone_001", "user_001") + + def test_retry_not_retryable_ready(self) -> None: + """ready 状态不可重试。""" + profile = _make_profile(status=VoiceCloneStatus.READY) + repo = MagicMock() + repo.get.return_value = profile + + use_case = RetryVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotRetryableError): + use_case.execute("test_clone_001", "user_001") + + def test_retry_not_retryable_processing(self) -> None: + """processing 状态不可重试。""" + profile = _make_profile(status=VoiceCloneStatus.PROCESSING) + repo = MagicMock() + repo.get.return_value = profile + + use_case = RetryVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotRetryableError): + use_case.execute("test_clone_001", "user_001") + + def test_retry_wrong_user(self) -> None: + """用户不匹配。""" + profile = _make_profile(user_id="other_user", status=VoiceCloneStatus.FAILED) + repo = MagicMock() + repo.get.return_value = profile + + use_case = RetryVoiceCloneUseCase(repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("test_clone_001", "user_001")