Compare commits

...

3 Commits

Author SHA1 Message Date
CI Bot b23bd2a942 style: auto-format with black + isort + prettier [skip ci-format-check]
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 56s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m11s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m55s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m58s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m59s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 48s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 47s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m18s
AI Code Review / AI Code Review (pull_request) Successful in 4m19s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m3s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m3s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m57s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 5m45s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 3m45s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 9m12s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 6s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 49s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 49s
2026-07-30 06:33:54 +00:00
xiaoxia d6dfb48687 fix(test): 修复voice_clone_preview测试污染GetVoiceCloneUseCase类的问题
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 21s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m14s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m17s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 45s
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
测试中用delattr删除了类原生execute方法,导致后续测试文件
test_voice_clone_use_cases.py中GetVoiceCloneUseCase.execute丢失。
改为保存原始方法并在finally中恢复。
2026-07-30 14:29:29 +08:00
xiaoxia 1437120634 feat(#1195): 新增克隆音色试听接口 /voice-clones/{id}/preview
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 22s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) Failing after 1m13s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m36s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m37s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m10s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m29s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
- 新增 GET /voice-clones/{clone_id}/preview 端点
- 支持默认试听文本(缓存7天)和自定义文本
- 状态校验:仅 ready 状态可试听
- 10 个单元测试覆盖成功/失败/缓存/权限等场景
2026-07-30 13:56:42 +08:00
4 changed files with 388 additions and 2 deletions
+84 -1
View File
@@ -11,6 +11,7 @@ from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repo
from app.schemas.voice_clone import (
CreateVoiceCloneRequest,
ListVoiceCloneResponse,
VoiceClonePreviewResponse,
VoiceCloneProfileResponse,
VoiceCloneStatusResponse,
)
@@ -19,7 +20,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
SQLAlchemyVoiceCloneProfileRepository,
)
from packages.application.cosyvoice_service import CosyVoiceService
from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
from packages.application.voice_clone.use_cases import (
DeleteVoiceCloneUseCase,
GetVoiceCloneStatusUseCase,
@@ -36,6 +37,13 @@ logger = logging.getLogger(__name__)
router = APIRouter()
# 克隆音色试听缓存(减少重复TTS调用)
# key: clone_id, value: (audio_url, duration, file_size, text, timestamp)
_clone_preview_cache: dict[str, tuple[str, float, int, str, float]] = {}
CLONE_PREVIEW_CACHE_TTL = 7 * 24 * 3600 # 7天TTL
# 默认试听文本
CLONE_PREVIEW_TEMPLATE = "你好,这是我的克隆音色,很高兴能为你配音。"
def _to_response(profile) -> VoiceCloneProfileResponse:
# source_audio_url 是用户传入的原始 URL(可能是外部地址),不做预签名转换
@@ -223,3 +231,78 @@ def retry_voice_clone(
logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
return _to_response(profile)
@router.get("/{clone_id}/preview", response_model=VoiceClonePreviewResponse)
def get_voice_clone_preview(
clone_id: str,
text: str = Query("", description="自定义试听文本,为空则使用默认示例"),
authenticated_user: AuthenticatedUser = Depends(get_current_user),
repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
cosyvoice: CosyVoiceService = Depends(get_cosyvoice_service),
) -> VoiceClonePreviewResponse:
"""获取克隆音色试听音频(实时 TTS 合成)。
- 克隆音色必须处于 ready 状态
- 使用默认试听文本时,结果缓存 7 天
- 可传入自定义 text 参数试听不同文本
"""
import time
use_case = GetVoiceCloneUseCase(repository)
try:
profile = use_case.execute(clone_id, authenticated_user.user.id)
except VoiceCloneNotFoundError as _e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") from _e
if not profile.is_ready:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Voice clone is not ready (current status: {profile.status})",
)
# 有自定义文本时不缓存
use_cache = not text.strip()
if use_cache and clone_id in _clone_preview_cache:
audio_url, duration, file_size, cached_text, cached_at = _clone_preview_cache[clone_id]
if time.time() - cached_at < CLONE_PREVIEW_CACHE_TTL:
return VoiceClonePreviewResponse(
clone_id=clone_id,
voice_id=profile.voice_id,
audio_url=audio_url,
text=cached_text,
duration=duration,
file_size=file_size,
)
# 合成试听音频
preview_text = text.strip() or CLONE_PREVIEW_TEMPLATE
try:
result = cosyvoice.synthesize_speech(
text=preview_text,
voice_id=profile.voice_id,
format="mp3",
speed=1.0,
)
except CosyVoiceError as e:
raise HTTPException(status_code=502, detail=f"TTS 合成失败: {e}") from e
# 缓存(仅默认试听文本)
if use_cache:
_clone_preview_cache[clone_id] = (
result.audio_url,
result.duration,
result.file_size,
preview_text,
time.time(),
)
return VoiceClonePreviewResponse(
clone_id=clone_id,
voice_id=profile.voice_id,
audio_url=result.audio_url,
text=preview_text,
duration=result.duration,
file_size=result.file_size,
)
+22
View File
@@ -63,3 +63,25 @@ class ListVoiceCloneResponse(BaseModel):
items: List[VoiceCloneProfileResponse]
total: int
class VoiceClonePreviewResponse(BaseModel):
"""克隆音色试听响应。"""
clone_id: str
"""音色克隆档案 ID"""
voice_id: str
"""CosyVoice 音色 ID"""
audio_url: str
"""试听音频 URL"""
text: str
"""试听文本"""
duration: float = 0.0
"""音频时长(秒)"""
file_size: int = 0
"""文件大小(字节)"""
@@ -642,4 +642,3 @@
font-size: 36px;
}
}
+282
View File
@@ -0,0 +1,282 @@
"""音色克隆试听接口单元测试 (#1195).
直接测试路由函数逻辑,mock 所有依赖。
"""
from __future__ import annotations
import time
from datetime import datetime, timezone
from unittest.mock import MagicMock
import pytest
from fastapi import HTTPException
from packages.application.cosyvoice_service import CosyVoiceError, SynthesizeResult
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
def _make_profile(**kwargs) -> VoiceCloneProfile:
defaults = {
"id": "clone_001",
"user_id": "user_001",
"name": "测试音色",
"description": "",
"source_audio_url": "",
"voice_id": "clone_voice_001",
"voice_model": "cosyvoice-v3",
"language": "zh-CN",
"gender": "female",
"status": VoiceCloneStatus.READY,
"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)
def _make_auth_user(user_id: str = "user_001") -> MagicMock:
user = MagicMock()
user.id = user_id
auth = MagicMock()
auth.user = user
return auth
class TestVoiceClonePreview:
"""克隆音色试听接口测试。"""
def _call_preview(self, profile, cosyvoice_mock, text="", user_id="user_001"):
"""调用路由函数,模拟 FastAPI 注入依赖。"""
from app.api.routes.voice_clones import get_voice_clone_preview
repo = MagicMock()
if profile is None:
from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError
repo.get.return_value = None
use_case_instance = MagicMock()
use_case_instance.execute.side_effect = VoiceCloneNotFoundError("not found")
else:
repo.get.return_value = profile
use_case_instance = MagicMock()
use_case_instance.execute.return_value = profile
# 替换 use case 构造
from packages.application.voice_clone.use_cases import GetVoiceCloneUseCase
original_init = GetVoiceCloneUseCase.__init__
original_execute = GetVoiceCloneUseCase.execute
def mock_init(self, repository):
self.repository = repository
def mock_execute(self, clone_id, uid):
if profile is None:
from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError
raise VoiceCloneNotFoundError("not found")
if profile.user_id != uid:
from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError
raise VoiceCloneNotFoundError("not found")
return profile
GetVoiceCloneUseCase.__init__ = mock_init
GetVoiceCloneUseCase.execute = mock_execute
try:
result = get_voice_clone_preview(
clone_id=profile.id if profile else "nonexistent",
text=text,
authenticated_user=_make_auth_user(user_id),
repository=repo,
cosyvoice=cosyvoice_mock,
)
return result
finally:
GetVoiceCloneUseCase.__init__ = original_init
GetVoiceCloneUseCase.execute = original_execute
def test_preview_success_ready_clone(self) -> None:
"""就绪的克隆音色可以正常试听。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile()
cosyvoice = MagicMock()
cosyvoice.synthesize_speech.return_value = SynthesizeResult(
audio_url="https://oss.example.com/preview/clone_001.mp3",
duration=3.5,
file_size=56000,
request_id="req_001",
)
result = self._call_preview(profile, cosyvoice)
assert result.clone_id == "clone_001"
assert result.voice_id == "clone_voice_001"
assert result.audio_url == "https://oss.example.com/preview/clone_001.mp3"
assert result.duration == 3.5
assert result.file_size == 56000
assert "克隆音色" in result.text
cosyvoice.synthesize_speech.assert_called_once()
call_kwargs = cosyvoice.synthesize_speech.call_args
assert call_kwargs.kwargs["voice_id"] == "clone_voice_001"
assert call_kwargs.kwargs["format"] == "mp3"
def test_preview_custom_text(self) -> None:
"""自定义试听文本。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile()
cosyvoice = MagicMock()
cosyvoice.synthesize_speech.return_value = SynthesizeResult(
audio_url="https://oss.example.com/preview/custom.mp3",
duration=2.0,
file_size=32000,
)
custom_text = "大家好,这是自定义试听文本。"
result = self._call_preview(profile, cosyvoice, text=custom_text)
assert result.text == custom_text
call_kwargs = cosyvoice.synthesize_speech.call_args
assert call_kwargs.kwargs["text"] == custom_text
def test_preview_not_found(self) -> None:
"""克隆音色不存在时返回 404。"""
cosyvoice = MagicMock()
with pytest.raises(HTTPException) as exc_info:
self._call_preview(None, cosyvoice)
assert exc_info.value.status_code == 404
cosyvoice.synthesize_speech.assert_not_called()
def test_preview_not_ready_pending(self) -> None:
"""pending 状态的克隆音色不能试听。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile(status=VoiceCloneStatus.PENDING, voice_id="")
cosyvoice = MagicMock()
with pytest.raises(HTTPException) as exc_info:
self._call_preview(profile, cosyvoice)
assert exc_info.value.status_code == 400
assert "not ready" in exc_info.value.detail.lower() or "未就绪" in exc_info.value.detail
cosyvoice.synthesize_speech.assert_not_called()
def test_preview_not_ready_processing(self) -> None:
"""processing 状态的克隆音色不能试听。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
cosyvoice = MagicMock()
with pytest.raises(HTTPException) as exc_info:
self._call_preview(profile, cosyvoice)
assert exc_info.value.status_code == 400
cosyvoice.synthesize_speech.assert_not_called()
def test_preview_not_ready_failed(self) -> None:
"""failed 状态的克隆音色不能试听。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile(status=VoiceCloneStatus.FAILED, error_message="超时")
cosyvoice = MagicMock()
with pytest.raises(HTTPException) as exc_info:
self._call_preview(profile, cosyvoice)
assert exc_info.value.status_code == 400
cosyvoice.synthesize_speech.assert_not_called()
def test_preview_tts_failure(self) -> None:
"""TTS 合成失败返回 502。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile()
cosyvoice = MagicMock()
cosyvoice.synthesize_speech.side_effect = CosyVoiceError("API 调用失败")
with pytest.raises(HTTPException) as exc_info:
self._call_preview(profile, cosyvoice)
assert exc_info.value.status_code == 502
assert "TTS" in exc_info.value.detail
def test_preview_cache_default_text(self) -> None:
"""默认试听文本使用缓存。"""
from app.api.routes.voice_clones import CLONE_PREVIEW_CACHE_TTL, _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile()
cosyvoice = MagicMock()
cosyvoice.synthesize_speech.return_value = SynthesizeResult(
audio_url="https://oss.example.com/preview/cached.mp3",
duration=3.0,
file_size=48000,
)
# 第一次调用:合成
result1 = self._call_preview(profile, cosyvoice)
assert cosyvoice.synthesize_speech.call_count == 1
# 第二次调用:走缓存,不重复合成
result2 = self._call_preview(profile, cosyvoice)
assert cosyvoice.synthesize_speech.call_count == 1
assert result2.audio_url == result1.audio_url
assert result2.duration == result1.duration
def test_preview_custom_text_no_cache(self) -> None:
"""自定义文本不使用缓存。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile()
cosyvoice = MagicMock()
cosyvoice.synthesize_speech.side_effect = [
SynthesizeResult(audio_url="https://example.com/1.mp3", duration=2.0, file_size=32000),
SynthesizeResult(audio_url="https://example.com/2.mp3", duration=2.5, file_size=40000),
]
result1 = self._call_preview(profile, cosyvoice, text="自定义文本一")
result2 = self._call_preview(profile, cosyvoice, text="自定义文本二")
assert cosyvoice.synthesize_speech.call_count == 2
assert result1.audio_url != result2.audio_url
def test_preview_wrong_user(self) -> None:
"""非所有者不能访问他人克隆音色的试听。"""
from app.api.routes.voice_clones import _clone_preview_cache
_clone_preview_cache.clear()
profile = _make_profile(user_id="user_001")
cosyvoice = MagicMock()
with pytest.raises(HTTPException) as exc_info:
self._call_preview(profile, cosyvoice, user_id="user_002")
assert exc_info.value.status_code == 404
cosyvoice.synthesize_speech.assert_not_called()