From 099bc31cf225cddfc4a33e1eff1a3740e4e357a4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Thu, 2 Jul 2026 11:36:02 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20CosyVoice=20=E8=AF=AD=E9=9F=B3=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E5=B1=82=20(Task=203.07)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 创建 CosyVoiceService 封装阿里云 CosyVoice API - list_preset_voices(): 返回预置音色列表 - clone_voice(): 音色克隆（提交任务 + 轮询状态） - synthesize_speech(): 语音合成（提交任务 + 轮询状态） - 支持同步/异步两种 API 模式 - 指数退避重试（最多 3 次）+ 超时控制 - 完整错误处理：AuthError / TimeoutError / 通用 Error - 26 个单元测试全部通过（mock CosyVoice API 响应） --- packages/application/cosyvoice_service.py | 512 ++++++++++++++++++++++ tests/unit/test_cosyvoice_service.py | 508 +++++++++++++++++++++ 2 files changed, 1020 insertions(+) create mode 100644 packages/application/cosyvoice_service.py create mode 100644 tests/unit/test_cosyvoice_service.py diff --git a/packages/application/cosyvoice_service.py b/packages/application/cosyvoice_service.py new file mode 100644 index 000000000..4bb5c83b7 --- /dev/null +++ b/packages/application/cosyvoice_service.py @@ -0,0 +1,512 @@ +"""CosyVoice 语音服务 — Phase 3. + +封装阿里云 CosyVoice 语音合成 API，提供： +- 预置音色列表查询 +- 音色克隆（提交任务 + 轮询状态） +- 语音合成（提交任务 + 轮询状态） + +API 文档: https://help.aliyun.com/zh/model-studio/cosyvoice +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Any, Optional + +import httpx + +from packages.domain.preset_voices import PresetVoice, get_preset_voices +from packages.shared.config import get_shared_settings + +logger = logging.getLogger(__name__) + + +class CosyVoiceError(Exception): + """CosyVoice API 调用异常。""" + + pass + + +class CosyVoiceTimeoutError(CosyVoiceError): + """CosyVoice API 超时。""" + + pass + + +class CosyVoiceAuthError(CosyVoiceError): + """CosyVoice API 认证失败。""" + + pass + + +@dataclass +class CloneResult: + """音色克隆结果。""" + + voice_id: str + request_id: str = "" + + +@dataclass +class SynthesizeResult: + """语音合成结果。""" + + audio_url: str + duration: float = 0.0 + file_size: int = 0 + request_id: str = "" + + +class CosyVoiceService: + """CosyVoice 语音服务。 + + 封装阿里云 CosyVoice API，提供音色克隆和语音合成功能。 + 支持同步和异步两种模式： + - 同步：API 直接返回结果 + - 异步：API 返回 task_id，需要轮询状态 + + 使用示例: + service = CosyVoiceService( + api_key="your-api-key", + base_url="https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio", + model="cosyvoice-v1", + ) + + # 获取预置音色 + voices = service.list_preset_voices() + + # 音色克隆 + result = service.clone_voice(audio_url="https://example.com/audio.mp3") + + # 语音合成 + result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun") + """ + + # 轮询配置 + POLL_INTERVAL = 2.0 # 秒 + MAX_POLL_ATTEMPTS = 60 # 最多轮询 60 次（2分钟） + + # 重试配置 + MAX_RETRIES = 3 + RETRY_BACKOFF = 1.0 # 秒，指数退避基数 + + def __init__( + self, + api_key: str = "", + base_url: str = "", + model: str = "", + http_client: Optional[httpx.Client] = None, + ) -> None: + """初始化 CosyVoice 服务。 + + Args: + api_key: CosyVoice API Key，为空时从配置读取 + base_url: CosyVoice API Base URL，为空时从配置读取 + model: CosyVoice 模型名称，为空时从配置读取 + http_client: 可选的 HTTP 客户端（用于测试注入） + """ + settings = get_shared_settings() + + self._api_key = api_key or settings.cosyvoice_api_key + self._base_url = base_url or settings.cosyvoice_base_url + self._model = model or settings.cosyvoice_model + + self._client = http_client or httpx.Client( + timeout=httpx.Timeout(30.0, connect=10.0), + ) + self._owns_client = http_client is None + + def __enter__(self) -> CosyVoiceService: + return self + + def __exit__(self, *args: Any) -> None: + self.close() + + def close(self) -> None: + """关闭 HTTP 客户端。""" + if self._owns_client and self._client: + self._client.close() + + # ── 预置音色 ───────────────────────────────────────── + + def list_preset_voices(self) -> list[PresetVoice]: + """获取预置音色列表。 + + Returns: + 预置音色列表 + """ + return get_preset_voices() + + # ── 音色克隆 ───────────────────────────────────────── + + def clone_voice( + self, + audio_url: str, + voice_name: str = "", + language: str = "zh-CN", + timeout: float = 300.0, + ) -> CloneResult: + """克隆音色。 + + 提交音色克隆任务到 CosyVoice API，并轮询直到完成或超时。 + + Args: + audio_url: 参考音频 URL + voice_name: 音色名称（可选） + language: 语言代码 + timeout: 超时时间（秒） + + Returns: + CloneResult: 克隆结果，包含 voice_id + + Raises: + CosyVoiceError: API 调用失败 + CosyVoiceTimeoutError: 超时 + CosyVoiceAuthError: 认证失败 + ValueError: 参数无效 + """ + if not audio_url: + raise ValueError("audio_url 不能为空") + if not self._api_key: + raise CosyVoiceAuthError("CosyVoice API Key 未配置") + + # 构建请求 + payload = { + "model": self._model, + "input": { + "audio_url": audio_url, + }, + "parameters": { + "language": language, + }, + } + if voice_name: + payload["parameters"]["voice_name"] = voice_name + + # 调用 API + response = self._call_api( + method="POST", + path="/services/audio/voice-clone", + json=payload, + timeout=timeout, + ) + + # 解析响应 + output = response.get("output", {}) + + # 检查是否有 task_id（异步模式） + task_id = output.get("task_id") + voice_id = output.get("voice_id") + + if task_id: + # 异步模式：轮询任务状态 + result = self._poll_clone_task(task_id, timeout) + return CloneResult( + voice_id=result["voice_id"], + request_id=response.get("request_id", ""), + ) + elif voice_id: + # 同步模式：直接返回结果 + return CloneResult( + voice_id=voice_id, + request_id=response.get("request_id", ""), + ) + else: + raise CosyVoiceError( + f"CosyVoice API 未返回 task_id 或 voice_id: {response}" + ) + + def _poll_clone_task(self, task_id: str, timeout: float) -> dict: + """轮询音色克隆任务状态。 + + Args: + task_id: 任务 ID + timeout: 超时时间（秒） + + Returns: + 任务结果字典 + + Raises: + CosyVoiceError: 任务失败 + CosyVoiceTimeoutError: 超时 + """ + start_time = time.time() + attempts = 0 + + while attempts < self.MAX_POLL_ATTEMPTS: + elapsed = time.time() - start_time + if elapsed > timeout: + raise CosyVoiceTimeoutError( + f"音色克隆任务超时（{timeout}秒）: task_id={task_id}" + ) + + response = self._call_api( + method="GET", + path=f"/tasks/{task_id}", + timeout=30.0, + ) + + output = response.get("output", {}) + status = output.get("task_status", "").upper() + + if status == "SUCCEEDED": + voice_id = output.get("voice_id", "") + if not voice_id: + raise CosyVoiceError( + f"音色克隆任务成功但未返回 voice_id: {response}" + ) + return {"voice_id": voice_id} + elif status == "FAILED": + error_msg = output.get("message", "未知错误") + raise CosyVoiceError(f"音色克隆任务失败: {error_msg}") + elif status in ("PENDING", "RUNNING"): + # 继续轮询 + time.sleep(self.POLL_INTERVAL) + attempts += 1 + else: + raise CosyVoiceError(f"未知的任务状态: {status}") + + raise CosyVoiceTimeoutError( + f"音色克隆任务轮询次数超限: task_id={task_id}" + ) + + # ── 语音合成 ───────────────────────────────────────── + + def synthesize_speech( + self, + text: str, + voice_id: str = "", + sample_rate: int = 0, + format: str = "", + speed: float = 1.0, + timeout: float = 120.0, + ) -> SynthesizeResult: + """语音合成。 + + 提交语音合成任务到 CosyVoice API，并轮询直到完成或超时。 + + Args: + text: 要合成的文本 + voice_id: 音色 ID（预置音色或克隆音色） + sample_rate: 采样率（Hz），0 表示使用配置默认值 + format: 输出格式（mp3/wav/pcm），空表示使用配置默认值 + speed: 语速（0.5-2.0），1.0 为正常速度 + timeout: 超时时间（秒） + + Returns: + SynthesizeResult: 合成结果，包含 audio_url + + Raises: + CosyVoiceError: API 调用失败 + CosyVoiceTimeoutError: 超时 + CosyVoiceAuthError: 认证失败 + ValueError: 参数无效 + """ + if not text: + raise ValueError("text 不能为空") + if not voice_id: + raise ValueError("voice_id 不能为空") + if not self._api_key: + raise CosyVoiceAuthError("CosyVoice API Key 未配置") + + settings = get_shared_settings() + + # 构建请求 + payload = { + "model": self._model, + "input": { + "text": text, + }, + "parameters": { + "voice": voice_id, + "sample_rate": sample_rate or settings.cosyvoice_sample_rate, + "format": format or settings.cosyvoice_format, + "rate": speed, + }, + } + + # 调用 API + response = self._call_api( + method="POST", + path="/services/aigc/text2audio/generation", + json=payload, + timeout=timeout, + ) + + # 解析响应 + output = response.get("output", {}) + + # 检查是否有 task_id（异步模式） + task_id = output.get("task_id") + audio_url = output.get("audio_url") + + if task_id: + # 异步模式：轮询任务状态 + result = self._poll_synthesize_task(task_id, timeout) + return SynthesizeResult( + audio_url=result["audio_url"], + duration=result.get("duration", 0.0), + file_size=result.get("file_size", 0), + request_id=response.get("request_id", ""), + ) + elif audio_url: + # 同步模式：直接返回结果 + return SynthesizeResult( + audio_url=audio_url, + duration=output.get("duration", 0.0), + file_size=output.get("file_size", 0), + request_id=response.get("request_id", ""), + ) + else: + raise CosyVoiceError( + f"CosyVoice API 未返回 audio_url 或 task_id: {response}" + ) + + def _poll_synthesize_task(self, task_id: str, timeout: float) -> dict: + """轮询语音合成任务状态。 + + Args: + task_id: 任务 ID + timeout: 超时时间（秒） + + Returns: + 任务结果字典 + + Raises: + CosyVoiceError: 任务失败 + CosyVoiceTimeoutError: 超时 + """ + start_time = time.time() + attempts = 0 + + while attempts < self.MAX_POLL_ATTEMPTS: + elapsed = time.time() - start_time + if elapsed > timeout: + raise CosyVoiceTimeoutError( + f"语音合成任务超时（{timeout}秒）: task_id={task_id}" + ) + + response = self._call_api( + method="GET", + path=f"/tasks/{task_id}", + timeout=30.0, + ) + + output = response.get("output", {}) + status = output.get("task_status", "").upper() + + if status == "SUCCEEDED": + audio_url = output.get("audio_url", "") + if not audio_url: + raise CosyVoiceError( + f"语音合成任务成功但未返回 audio_url: {response}" + ) + return { + "audio_url": audio_url, + "duration": output.get("duration", 0.0), + "file_size": output.get("file_size", 0), + } + elif status == "FAILED": + error_msg = output.get("message", "未知错误") + raise CosyVoiceError(f"语音合成任务失败: {error_msg}") + elif status in ("PENDING", "RUNNING"): + # 继续轮询 + time.sleep(self.POLL_INTERVAL) + attempts += 1 + else: + raise CosyVoiceError(f"未知的任务状态: {status}") + + raise CosyVoiceTimeoutError( + f"语音合成任务轮询次数超限: task_id={task_id}" + ) + + # ── 内部方法 ───────────────────────────────────────── + + def _call_api( + self, + method: str, + path: str, + json: Optional[dict] = None, + timeout: float = 30.0, + ) -> dict: + """调用 CosyVoice API。 + + 支持重试和错误处理。 + + Args: + method: HTTP 方法（GET/POST） + path: API 路径 + json: 请求体 + timeout: 超时时间（秒） + + Returns: + API 响应字典 + + Raises: + CosyVoiceError: API 调用失败 + CosyVoiceAuthError: 认证失败 + CosyVoiceTimeoutError: 超时 + """ + url = f"{self._base_url}{path}" + headers = { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + last_error: Optional[Exception] = None + + for attempt in range(self.MAX_RETRIES): + try: + response = self._client.request( + method=method, + url=url, + headers=headers, + json=json, + timeout=timeout, + ) + + # 处理响应 + if response.status_code == 200: + return response.json() + elif response.status_code in (401, 403): + raise CosyVoiceAuthError( + f"CosyVoice API 认证失败: HTTP {response.status_code}" + ) + elif response.status_code >= 500: + # 服务端错误，可重试 + last_error = CosyVoiceError( + f"CosyVoice API 服务端错误: HTTP {response.status_code}" + ) + logger.warning( + f"CosyVoice API 失败 (尝试 {attempt + 1}/{self.MAX_RETRIES}): " + f"HTTP {response.status_code}" + ) + else: + # 客户端错误，不重试 + raise CosyVoiceError( + f"CosyVoice API 调用失败: HTTP {response.status_code}, " + f"body={response.text}" + ) + + except httpx.TimeoutException as e: + last_error = CosyVoiceTimeoutError(f"请求超时: {e}") + logger.warning( + f"CosyVoice API 超时 (尝试 {attempt + 1}/{self.MAX_RETRIES})" + ) + except httpx.RequestError as e: + last_error = CosyVoiceError(f"请求错误: {e}") + logger.warning( + f"CosyVoice API 请求错误 (尝试 {attempt + 1}/{self.MAX_RETRIES}): {e}" + ) + + # 指数退避 + if attempt < self.MAX_RETRIES - 1: + sleep_time = self.RETRY_BACKOFF * (2**attempt) + time.sleep(sleep_time) + + # 所有重试都失败 + if last_error: + raise last_error + raise CosyVoiceError("CosyVoice API 调用失败，未知错误") diff --git a/tests/unit/test_cosyvoice_service.py b/tests/unit/test_cosyvoice_service.py new file mode 100644 index 000000000..b32020954 --- /dev/null +++ b/tests/unit/test_cosyvoice_service.py @@ -0,0 +1,508 @@ +"""CosyVoiceService 单元测试。""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock + +import httpx +import pytest + +from packages.application.cosyvoice_service import ( + CloneResult, + CosyVoiceAuthError, + CosyVoiceError, + CosyVoiceService, + CosyVoiceTimeoutError, + SynthesizeResult, +) +from packages.domain.preset_voices import PresetVoice + + +def _make_service( + *, + api_key: str = "test-api-key", + base_url: str = "https://test.cosyvoice.api", + model: str = "cosyvoice-v1", + http_client: httpx.Client | None = None, +) -> CosyVoiceService: + """创建测试用 CosyVoiceService。""" + return CosyVoiceService( + api_key=api_key, + base_url=base_url, + model=model, + http_client=http_client, + ) + + +def _mock_response( + status_code: int = 200, + json_data: dict | None = None, + text: str = "", +) -> httpx.Response: + """创建 mock HTTP 响应。""" + # httpx.Response 需要 content 参数才能正确调用 .json() + content = b"" + if json_data is not None: + content = json.dumps(json_data).encode("utf-8") + elif text: + content = text.encode("utf-8") + + return httpx.Response( + status_code=status_code, + content=content, + request=httpx.Request("POST", "https://test.cosyvoice.api"), + ) + + +# ── list_preset_voices ─────────────────────────────────── + + +class TestListPresetVoices: + """测试预置音色列表。""" + + def test_returns_all_preset_voices(self) -> None: + """返回所有预置音色。""" + service = _make_service() + voices = service.list_preset_voices() + + assert len(voices) == 8 + assert all(isinstance(v, PresetVoice) for v in voices) + + def test_preset_voice_ids(self) -> None: + """预置音色 ID 正确。""" + service = _make_service() + voices = service.list_preset_voices() + voice_ids = [v.voice_id for v in voices] + + assert "longxiaochun" in voice_ids + assert "longxiaoxia" in voice_ids + assert "longxiaochen" in voice_ids + assert "longyue" in voice_ids + assert "longshu" in voice_ids + assert "longjing" in voice_ids + assert "longbo" in voice_ids + assert "longtian" in voice_ids + + def test_preset_voice_has_required_fields(self) -> None: + """预置音色包含必要字段。""" + service = _make_service() + voices = service.list_preset_voices() + + for voice in voices: + assert voice.voice_id + assert voice.name + assert voice.gender in ("male", "female") + assert voice.language == "zh-CN" + + +# ── clone_voice ────────────────────────────────────────── + + +class TestCloneVoice: + """测试音色克隆。""" + + def test_clone_sync_success(self) -> None: + """同步克隆成功（直接返回 voice_id）。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + json_data={ + "request_id": "req-001", + "output": {"voice_id": "clone-voice-001"}, + } + ) + + service = _make_service(http_client=mock_client) + result = service.clone_voice( + audio_url="https://example.com/audio.mp3", + voice_name="我的音色", + ) + + assert isinstance(result, CloneResult) + assert result.voice_id == "clone-voice-001" + assert result.request_id == "req-001" + mock_client.request.assert_called_once() + + def test_clone_async_with_polling(self) -> None: + """异步克隆（返回 task_id，轮询后成功）。""" + mock_client = MagicMock(spec=httpx.Client) + + # 第一次调用：提交任务，返回 task_id + submit_response = _mock_response( + json_data={ + "request_id": "req-001", + "output": {"task_id": "task-abc123", "task_status": "PENDING"}, + } + ) + + # 第二次调用：查询状态 → RUNNING + running_response = _mock_response( + json_data={ + "request_id": "req-002", + "output": {"task_status": "RUNNING"}, + } + ) + + # 第三次调用：查询状态 → SUCCEEDED + success_response = _mock_response( + json_data={ + "request_id": "req-003", + "output": { + "task_status": "SUCCEEDED", + "voice_id": "clone-voice-async-001", + }, + } + ) + + mock_client.request.side_effect = [ + submit_response, + running_response, + success_response, + ] + + service = _make_service(http_client=mock_client) + service.POLL_INTERVAL = 0 # 测试中不等待 + + result = service.clone_voice( + audio_url="https://example.com/audio.mp3", + ) + + assert result.voice_id == "clone-voice-async-001" + assert mock_client.request.call_count == 3 + + def test_clone_async_task_failed(self) -> None: + """异步克隆任务失败。""" + mock_client = MagicMock(spec=httpx.Client) + + submit_response = _mock_response( + json_data={ + "output": {"task_id": "task-fail"}, + } + ) + failed_response = _mock_response( + json_data={ + "output": { + "task_status": "FAILED", + "message": "音频质量不达标", + }, + } + ) + + mock_client.request.side_effect = [submit_response, failed_response] + + service = _make_service(http_client=mock_client) + service.POLL_INTERVAL = 0 + + with pytest.raises(CosyVoiceError, match="音频质量不达标"): + service.clone_voice(audio_url="https://example.com/bad.mp3") + + def test_clone_empty_audio_url_raises(self) -> None: + """空 audio_url 抛出 ValueError。""" + service = _make_service() + + with pytest.raises(ValueError, match="audio_url 不能为空"): + service.clone_voice(audio_url="") + + def test_clone_no_api_key_raises_auth_error(self) -> None: + """未配置 API Key 抛出 CosyVoiceAuthError。""" + service = _make_service(api_key="") + + with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + def test_clone_auth_failure(self) -> None: + """API 认证失败（401）。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response(status_code=401) + + service = _make_service(http_client=mock_client) + + with pytest.raises(CosyVoiceAuthError, match="认证失败"): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + def test_clone_client_error_no_retry(self) -> None: + """客户端错误（400）不重试。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + status_code=400, text="Bad Request" + ) + + service = _make_service(http_client=mock_client) + + with pytest.raises(CosyVoiceError, match="HTTP 400"): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + # 客户端错误不重试，只调用一次 + assert mock_client.request.call_count == 1 + + def test_clone_server_error_retries(self) -> None: + """服务端错误（500）重试。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response(status_code=500) + + service = _make_service(http_client=mock_client) + service.RETRY_BACKOFF = 0 # 测试中不等待 + + with pytest.raises(CosyVoiceError, match="服务端错误"): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + # 服务端错误重试 MAX_RETRIES 次 + assert mock_client.request.call_count == service.MAX_RETRIES + + def test_clone_timeout_retries(self) -> None: + """超时重试。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.side_effect = httpx.TimeoutException("timeout") + + service = _make_service(http_client=mock_client) + service.RETRY_BACKOFF = 0 + + with pytest.raises(CosyVoiceTimeoutError): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + assert mock_client.request.call_count == service.MAX_RETRIES + + def test_clone_with_voice_name(self) -> None: + """带 voice_name 参数。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + json_data={"output": {"voice_id": "v-001"}} + ) + + service = _make_service(http_client=mock_client) + service.clone_voice( + audio_url="https://example.com/audio.mp3", + voice_name="测试音色", + ) + + call_args = mock_client.request.call_args + payload = call_args.kwargs.get("json") or call_args[1].get("json") + assert payload["parameters"]["voice_name"] == "测试音色" + + def test_clone_no_task_id_or_voice_id_raises(self) -> None: + """API 返回无效响应（无 task_id 也无 voice_id）。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + json_data={"output": {}} + ) + + service = _make_service(http_client=mock_client) + + with pytest.raises(CosyVoiceError, match="未返回 task_id 或 voice_id"): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + +# ── synthesize_speech ──────────────────────────────────── + + +class TestSynthesizeSpeech: + """测试语音合成。""" + + def test_synthesize_sync_success(self) -> None: + """同步合成成功（直接返回 audio_url）。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + json_data={ + "request_id": "req-tts-001", + "output": { + "audio_url": "https://cdn.example.com/audio.mp3", + "duration": 5.2, + "file_size": 83200, + }, + } + ) + + service = _make_service(http_client=mock_client) + result = service.synthesize_speech( + text="你好世界", + voice_id="longxiaochun", + ) + + assert isinstance(result, SynthesizeResult) + assert result.audio_url == "https://cdn.example.com/audio.mp3" + assert result.duration == 5.2 + assert result.file_size == 83200 + assert result.request_id == "req-tts-001" + + def test_synthesize_async_with_polling(self) -> None: + """异步合成（返回 task_id，轮询后成功）。""" + mock_client = MagicMock(spec=httpx.Client) + + submit_response = _mock_response( + json_data={ + "output": {"task_id": "task-tts-001", "task_status": "PENDING"}, + } + ) + success_response = _mock_response( + json_data={ + "output": { + "task_status": "SUCCEEDED", + "audio_url": "https://cdn.example.com/async.mp3", + "duration": 3.0, + }, + } + ) + + mock_client.request.side_effect = [submit_response, success_response] + + service = _make_service(http_client=mock_client) + service.POLL_INTERVAL = 0 + + result = service.synthesize_speech( + text="异步合成测试", + voice_id="longxiaoxia", + ) + + assert result.audio_url == "https://cdn.example.com/async.mp3" + assert result.duration == 3.0 + assert mock_client.request.call_count == 2 + + def test_synthesize_async_task_failed(self) -> None: + """异步合成任务失败。""" + mock_client = MagicMock(spec=httpx.Client) + + submit_response = _mock_response( + json_data={"output": {"task_id": "task-tts-fail"}} + ) + failed_response = _mock_response( + json_data={ + "output": { + "task_status": "FAILED", + "message": "文本过长", + }, + } + ) + + mock_client.request.side_effect = [submit_response, failed_response] + + service = _make_service(http_client=mock_client) + service.POLL_INTERVAL = 0 + + with pytest.raises(CosyVoiceError, match="文本过长"): + service.synthesize_speech( + text="超长文本" * 10000, + voice_id="longxiaochun", + ) + + def test_synthesize_empty_text_raises(self) -> None: + """空 text 抛出 ValueError。""" + service = _make_service() + + with pytest.raises(ValueError, match="text 不能为空"): + service.synthesize_speech(text="", voice_id="longxiaochun") + + def test_synthesize_empty_voice_id_raises(self) -> None: + """空 voice_id 抛出 ValueError。""" + service = _make_service() + + with pytest.raises(ValueError, match="voice_id 不能为空"): + service.synthesize_speech(text="测试", voice_id="") + + def test_synthesize_no_api_key_raises(self) -> None: + """未配置 API Key 抛出 CosyVoiceAuthError。""" + service = _make_service(api_key="") + + with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"): + service.synthesize_speech(text="测试", voice_id="longxiaochun") + + def test_synthesize_with_parameters(self) -> None: + """带采样率、格式、语速参数。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + json_data={"output": {"audio_url": "https://cdn.example.com/out.wav"}} + ) + + service = _make_service(http_client=mock_client) + service.synthesize_speech( + text="参数测试", + voice_id="longxiaochun", + sample_rate=44100, + format="wav", + speed=1.5, + ) + + call_args = mock_client.request.call_args + payload = call_args.kwargs.get("json") or call_args[1].get("json") + params = payload["parameters"] + assert params["sample_rate"] == 44100 + assert params["format"] == "wav" + assert params["rate"] == 1.5 + + def test_synthesize_no_url_or_task_id_raises(self) -> None: + """API 返回无效响应（无 audio_url 也无 task_id）。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response( + json_data={"output": {}} + ) + + service = _make_service(http_client=mock_client) + + with pytest.raises(CosyVoiceError, match="未返回 audio_url 或 task_id"): + service.synthesize_speech(text="测试", voice_id="longxiaochun") + + def test_synthesize_server_error_retries(self) -> None: + """服务端错误重试。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response(status_code=502) + + service = _make_service(http_client=mock_client) + service.RETRY_BACKOFF = 0 + + with pytest.raises(CosyVoiceError, match="服务端错误"): + service.synthesize_speech(text="测试", voice_id="longxiaochun") + + assert mock_client.request.call_count == service.MAX_RETRIES + + def test_synthesize_timeout_retries(self) -> None: + """超时重试。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.side_effect = httpx.TimeoutException("timeout") + + service = _make_service(http_client=mock_client) + service.RETRY_BACKOFF = 0 + + with pytest.raises(CosyVoiceTimeoutError): + service.synthesize_speech(text="测试", voice_id="longxiaochun") + + assert mock_client.request.call_count == service.MAX_RETRIES + + +# ── 重试逻辑 ───────────────────────────────────────────── + + +class TestRetryLogic: + """测试重试逻辑。""" + + def test_retry_then_success(self) -> None: + """第一次失败，第二次成功。""" + mock_client = MagicMock(spec=httpx.Client) + + # 第一次：服务端错误 + error_response = _mock_response(status_code=500) + # 第二次：成功 + success_response = _mock_response( + json_data={"output": {"voice_id": "v-retry-ok"}} + ) + + mock_client.request.side_effect = [error_response, success_response] + + service = _make_service(http_client=mock_client) + service.RETRY_BACKOFF = 0 + + result = service.clone_voice(audio_url="https://example.com/audio.mp3") + + assert result.voice_id == "v-retry-ok" + assert mock_client.request.call_count == 2 + + def test_max_retries_exhausted(self) -> None: + """达到最大重试次数后抛出异常。""" + mock_client = MagicMock(spec=httpx.Client) + mock_client.request.return_value = _mock_response(status_code=503) + + service = _make_service(http_client=mock_client) + service.RETRY_BACKOFF = 0 + + with pytest.raises(CosyVoiceError): + service.clone_voice(audio_url="https://example.com/audio.mp3") + + assert mock_client.request.call_count == service.MAX_RETRIES -- 2.54.0