feat: 任务3.05 音色克隆API — CRUD + 状态管理 #165
@@ -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")
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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)
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user