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 调用失败，未知错误")

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