From 506eae98f146af0e1566888a8c91dac930f29be6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 22:17:07 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20CosyVoice=E5=85=A8=E9=87=8F=E9=87=8D?= =?UTF-8?q?=E5=86=99=20-=20=E9=80=82=E9=85=8DDashScope=E7=99=BE=E7=82=BCAP?= =?UTF-8?q?I?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 配置更新：base_url改为api/v1，model改为cosyvoice-v3.5-plus，新增clone_model=voice-enrollment - 音色克隆：适配customization接口 + voice-enrollment模型 + create_voice action - 状态查询：适配同路径query_voice action（POST），返回DEPLOYING/OK/UNDEPLOYED - 语音合成：适配SpeechSynthesizer非流式同步接口，直接返回audio_url - 新增audio_url_signer钩子：对私有bucket音频URL预签名（slash_safe=True） - voice_clone workflow适配：voice_id替代task_id，status=OK标记ready - tts_job workflow适配：同步接口直接返回，无task_id时自动重新合成兜底 - API端+Worker端依赖注入OSS预签名函数 - 单元测试：36个service测试 + 18个workflow测试，全部通过 --- apps/api/app/dependencies.py | 15 +- apps/worker/worker_app/tasks/voice_clone.py | 5 +- packages/application/cosyvoice_service.py | 599 ++++++------- packages/application/tts_job/workflow.py | 76 +- packages/application/voice_clone/workflow.py | 17 +- packages/shared/config.py | 8 +- tests/unit/test_cosyvoice_service.py | 890 ++++++++----------- tests/unit/test_voice_clone_workflow.py | 32 +- 8 files changed, 793 insertions(+), 849 deletions(-) mode change 100644 => 100755 apps/worker/worker_app/tasks/voice_clone.py mode change 100644 => 100755 packages/application/cosyvoice_service.py mode change 100644 => 100755 packages/application/tts_job/workflow.py mode change 100644 => 100755 packages/application/voice_clone/workflow.py mode change 100644 => 100755 tests/unit/test_cosyvoice_service.py mode change 100644 => 100755 tests/unit/test_voice_clone_workflow.py diff --git a/apps/api/app/dependencies.py b/apps/api/app/dependencies.py index 9c42a59b3..afc9c40c4 100755 --- a/apps/api/app/dependencies.py +++ b/apps/api/app/dependencies.py @@ -201,7 +201,18 @@ def get_voice_clone_profile_repository( def get_cosyvoice_service(): - """Provide the CosyVoice service instance.""" + """Provide the CosyVoice service instance. + + 注入 OSS 音频URL预签名函数，确保私有bucket下的参考音频 + 能被 CosyVoice 服务器下载。 + """ + from app.core.storage import get_storage_service from packages.application.cosyvoice_service import CosyVoiceService - return CosyVoiceService() + storage = get_storage_service() + + def _sign_audio_url(url: str) -> str: + """对音频URL做预签名，私有bucket下 CosyVoice 服务器才能下载.""" + return storage.get_download_url(url, expires_seconds=86400) + + return CosyVoiceService(audio_url_signer=_sign_audio_url) diff --git a/apps/worker/worker_app/tasks/voice_clone.py b/apps/worker/worker_app/tasks/voice_clone.py old mode 100644 new mode 100755 index ecf3abab6..008533fa5 --- a/apps/worker/worker_app/tasks/voice_clone.py +++ b/apps/worker/worker_app/tasks/voice_clone.py @@ -16,6 +16,7 @@ from packages.application.cosyvoice_service import ( CosyVoiceTimeoutError, ) from packages.application.voice_clone.workflow import VoiceCloneWorkflowService +from video_processing.oss_helpers import get_signed_download_url logger = logging.getLogger(__name__) @@ -48,7 +49,9 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: repo = SQLAlchemyVoiceCloneProfileRepository(session) workflow = VoiceCloneWorkflowService( repository=repo, - cosyvoice_service=CosyVoiceService(), + cosyvoice_service=CosyVoiceService( + audio_url_signer=lambda url: get_signed_download_url(url, expires_seconds=86400) or url + ), ) updated_profile = workflow.poll_and_process_clone(profile_id, timeout=300) diff --git a/packages/application/cosyvoice_service.py b/packages/application/cosyvoice_service.py old mode 100644 new mode 100755 index 265c298da..f42784eaa --- a/packages/application/cosyvoice_service.py +++ b/packages/application/cosyvoice_service.py @@ -1,11 +1,13 @@ -"""CosyVoice 语音服务 — Phase 3. +"""CosyVoice 语音服务 — 适配阿里云百炼 DashScope API. -封装阿里云 CosyVoice 语音合成 API，提供： +封装阿里云百炼 CosyVoice 语音合成 API，提供： - 预置音色列表查询 -- 音色克隆（提交任务 + 轮询状态） -- 语音合成（提交任务 + 轮询状态） +- 音色克隆（提交 + 轮询状态） +- 语音合成（同步非流式调用） -API 文档: https://help.aliyun.com/zh/model-studio/cosyvoice +API 文档: +- 音色克隆: https://help.aliyun.com/document_detail/3027318.html +- 语音合成: https://help.aliyun.com/zh/model-studio/cosyvoice-tts-http-api """ from __future__ import annotations @@ -60,23 +62,24 @@ class SynthesizeResult: class CosyVoiceService: - """CosyVoice 语音服务。 + """CosyVoice 语音服务. - 封装阿里云 CosyVoice API，提供音色克隆和语音合成功能。 - 支持同步和异步两种模式： - - 同步：API 直接返回结果 - - 异步：API 返回 task_id，需要轮询状态 + 封装阿里云百炼 CosyVoice API，提供音色克隆和语音合成功能. + + 接口总览: + - 音色克隆: POST /services/audio/tts/customization (model=voice-enrollment) + - action=create_voice: 创建克隆音色，返回 voice_id（状态 DEPLOYING） + - action=query_voice: 查询音色状态（DEPLOYING / OK / UNDEPLOYED） + - 语音合成: POST /services/audio/tts/SpeechSynthesizer (model=cosyvoice-v3.5-plus) + - 非流式: 同步返回音频 URL 使用示例: service = CosyVoiceService( api_key="your-api-key", - base_url="https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio", - model="cosyvoice-v1", + base_url="https://dashscope.aliyuncs.com/api/v1", + model="cosyvoice-v3.5-plus", ) - # 获取预置音色 - voices = service.list_preset_voices() - # 音色克隆 result = service.clone_voice(audio_url="https://example.com/audio.mp3") @@ -84,9 +87,9 @@ class CosyVoiceService: result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun") """ - # 轮询配置 - POLL_INTERVAL = 2.0 # 秒 - MAX_POLL_ATTEMPTS = 60 # 最多轮询 60 次（2分钟） + # 音色状态轮询配置 + CLONE_POLL_INTERVAL = 5.0 # 秒 + CLONE_MAX_POLL_ATTEMPTS = 60 # 最多轮询 60 次（5分钟） # 重试配置 MAX_RETRIES = 3 @@ -97,24 +100,34 @@ class CosyVoiceService: api_key: str = "", base_url: str = "", model: str = "", + clone_model: str = "", http_client: Optional[httpx.Client] = None, + audio_url_signer: Optional[callable] = None, ) -> None: - """初始化 CosyVoice 服务。 + """初始化 CosyVoice 服务. Args: - api_key: CosyVoice API Key，为空时从配置读取 - base_url: CosyVoice API Base URL，为空时从配置读取 - model: CosyVoice 模型名称，为空时从配置读取 + api_key: DashScope API Key，为空时从配置读取 + base_url: DashScope API Base URL，为空时从配置读取 + model: 语音合成模型名称，为空时从配置读取 + clone_model: 音色克隆模型名称，为空时从配置读取 http_client: 可选的 HTTP 客户端（用于测试注入） + audio_url_signer: 可选的音频URL预签名函数，签名式 fn(url) -> str. + 用于私有 bucket 下，将裸 URL 转为预签名 URL， + 确保 CosyVoice 服务器能下载参考音频. """ 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._clone_model = clone_model or getattr( + settings, "cosyvoice_clone_model", "voice-enrollment" + ) + self._audio_url_signer = audio_url_signer self._client = http_client or httpx.Client( - timeout=httpx.Timeout(30.0, connect=10.0), + timeout=httpx.Timeout(60.0, connect=10.0), ) self._owns_client = http_client is None @@ -132,7 +145,7 @@ class CosyVoiceService: # ── 预置音色 ───────────────────────────────────────── def list_preset_voices(self) -> list[PresetVoice]: - """获取预置音色列表。 + """获取预置音色列表. Returns: 预置音色列表 @@ -146,20 +159,22 @@ class CosyVoiceService: audio_url: str, voice_name: str = "", language: str = "zh-CN", + target_model: str = "", ) -> dict: - """提交音色克隆任务（非阻塞）。 + """提交音色克隆任务（非阻塞）. - 只提交任务到 CosyVoice API，不轮询结果。 - 返回的 dict 包含 task_id（异步）或 voice_id（同步）。 + 调用百炼 voice-enrollment API 创建克隆音色. + 创建后音色状态为 DEPLOYING，需通过 query_voice_status 轮询直到 OK. Args: - audio_url: 参考音频 URL - voice_name: 音色名称（可选） - language: 语言代码 + audio_url: 参考音频 URL（必须公网可访问） + voice_name: 音色名称前缀（字母数字，最多10字符） + language: 语言代码（zh-CN 会转换为 zh） + target_model: 目标合成模型，默认使用当前 model Returns: - dict: {"task_id": str, "voice_id": str, "request_id": str} - task_id 和 voice_id 至少有一个非空 + dict: {"voice_id": str, "status": str, "request_id": str} + voice_id 非空，status 通常为 DEPLOYING Raises: CosyVoiceError: API 调用失败 @@ -171,48 +186,68 @@ class CosyVoiceService: if not self._api_key: raise CosyVoiceAuthError("CosyVoice API Key 未配置") + # voice_name 作为 prefix，限制字母数字，最多10字符 + # 不符合要求的做清洗 + prefix = self._sanitize_prefix(voice_name) if voice_name else "clone" + + # 语言转换：zh-CN → zh，保留 ISO 639-1 格式 + lang_code = language.split("-")[0].lower() if language else "zh" + + target = target_model or self._model + + # 如果配置了 audio_url_signer，对音频URL做预签名 + # （私有 bucket 下 CosyVoice 服务器无法直接访问裸 URL） + signed_audio_url = audio_url + if self._audio_url_signer: + try: + signed_audio_url = self._audio_url_signer(audio_url) + logger.info("音频URL已预签名: original=%s signed_prefix=%s", + audio_url[:80], signed_audio_url[:80]) + except Exception as e: + logger.warning("音频URL预签名失败，使用原始URL: %s", e) + payload = { - "model": self._model, + "model": self._clone_model, "input": { - "audio_url": audio_url, - }, - "parameters": { - "language": language, + "action": "create_voice", + "target_model": target, + "prefix": prefix, + "url": signed_audio_url, + "language_hints": [lang_code], }, } - if voice_name: - payload["parameters"]["voice_name"] = voice_name response = self._call_api( method="POST", - path="/services/audio/voice-clone", + path="/services/audio/tts/customization", json=payload, timeout=60.0, ) output = response.get("output", {}) - task_id = output.get("task_id", "") voice_id = output.get("voice_id", "") + status = output.get("status", "DEPLOYING") request_id = response.get("request_id", "") - if not task_id and not voice_id: - raise CosyVoiceError(f"CosyVoice API 未返回 task_id 或 voice_id: {response}") + if not voice_id: + raise CosyVoiceError(f"CosyVoice API 未返回 voice_id: {response}") return { - "task_id": task_id, "voice_id": voice_id, + "status": status, "request_id": request_id, } - def check_task_status(self, task_id: str) -> dict: - """查询克隆任务状态（单次查询，不轮询）。 + def query_voice_status(self, voice_id: str) -> dict: + """查询音色状态（单次查询，不轮询）. Args: - task_id: 任务 ID + voice_id: 音色 ID Returns: - dict: {"status": str, "voice_id": str, "message": str} - status 为 SUCCEEDED/FAILED/PENDING/RUNNING + dict: {"status": str, "target_model": str, "gmt_create": str, + "gmt_modified": str, "resource_link": str} + status 为 DEPLOYING / OK / UNDEPLOYED Raises: CosyVoiceError: API 调用失败 @@ -221,40 +256,98 @@ class CosyVoiceService: if not self._api_key: raise CosyVoiceAuthError("CosyVoice API Key 未配置") + if not voice_id: + raise ValueError("voice_id 不能为空") + + payload = { + "model": self._clone_model, + "input": { + "action": "query_voice", + "voice_id": voice_id, + }, + } + response = self._call_api( - method="GET", - path=f"/tasks/{task_id}", + method="POST", + path="/services/audio/tts/customization", + json=payload, timeout=30.0, ) output = response.get("output", {}) - status = output.get("task_status", "").upper() - voice_id = output.get("voice_id", "") - message = output.get("message", "") - return { - "status": status, - "voice_id": voice_id, - "message": message, + "status": output.get("status", ""), + "target_model": output.get("target_model", ""), + "gmt_create": output.get("gmt_create", ""), + "gmt_modified": output.get("gmt_modified", ""), + "resource_link": output.get("resource_link", ""), } - def poll_clone_task(self, task_id: str, timeout: float = 300.0) -> dict: - """轮询音色克隆任务状态（公开方法）。 + def check_task_status(self, task_id: str) -> dict: + """查询克隆任务状态（兼容旧接口，实际用 voice_id 查询）. - 供 Celery 后台任务调用，轮询直到完成或超时。 + 为了兼容旧代码，task_id 参数名保留，但实际传的是 voice_id. Args: - task_id: CosyVoice 任务 ID + task_id: 音色 ID（兼容旧接口名） + + Returns: + dict: {"status": str, "voice_id": str, "message": str} + """ + result = self.query_voice_status(task_id) + return { + "status": result["status"], + "voice_id": task_id, + "message": "", + } + + def poll_clone_task(self, voice_id: str, timeout: float = 300.0) -> dict: + """轮询音色克隆状态直到完成或超时. + + 供 Celery 后台任务调用，轮询直到状态变为 OK 或 UNDEPLOYED. + + Args: + voice_id: 音色 ID timeout: 超时时间（秒），默认 300 Returns: dict: {"voice_id": str} Raises: - CosyVoiceError: 任务失败 + CosyVoiceError: 任务失败（状态 UNDEPLOYED） CosyVoiceTimeoutError: 超时 """ - return self._poll_clone_task(task_id, timeout=timeout) + start_time = time.time() + attempts = 0 + + while attempts < self.CLONE_MAX_POLL_ATTEMPTS: + elapsed = time.time() - start_time + if elapsed > timeout: + raise CosyVoiceTimeoutError( + f"音色克隆任务超时（{timeout}秒）: voice_id={voice_id}" + ) + + result = self.query_voice_status(voice_id) + status = result.get("status", "").upper() + + if status == "OK": + return {"voice_id": voice_id} + elif status == "UNDEPLOYED": + raise CosyVoiceError( + f"音色克隆任务失败（审核未通过）: voice_id={voice_id}" + ) + elif status in ("DEPLOYING", "PENDING", "PROCESSING", ""): + # 继续轮询 + time.sleep(self.CLONE_POLL_INTERVAL) + attempts += 1 + else: + logger.warning("未知的音色状态: %s (voice_id=%s)", status, voice_id) + time.sleep(self.CLONE_POLL_INTERVAL) + attempts += 1 + + raise CosyVoiceTimeoutError( + f"音色克隆任务轮询次数超限: voice_id={voice_id}" + ) def clone_voice( self, @@ -262,122 +355,45 @@ class CosyVoiceService: voice_name: str = "", language: str = "zh-CN", timeout: float = 300.0, + target_model: str = "", ) -> CloneResult: - """克隆音色。 + """克隆音色（阻塞，直到完成或超时）. - 提交音色克隆任务到 CosyVoice API，并轮询直到完成或超时。 + 提交音色克隆到百炼 API，并轮询直到状态变为 OK 或超时. Args: - audio_url: 参考音频 URL - voice_name: 音色名称（可选） + audio_url: 参考音频 URL（必须公网可访问） + voice_name: 音色名称前缀 language: 语言代码 timeout: 超时时间（秒） + target_model: 目标合成模型 Returns: CloneResult: 克隆结果，包含 voice_id Raises: - CosyVoiceError: API 调用失败 + 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, + submit_result = self.submit_clone_task( + audio_url=audio_url, + voice_name=voice_name, + language=language, + target_model=target_model, ) - # 解析响应 - output = response.get("output", {}) + voice_id = submit_result["voice_id"] + request_id = submit_result["request_id"] - # 检查是否有 task_id（异步模式） - task_id = output.get("task_id") - voice_id = output.get("voice_id") + # 如果创建时已经是 OK 状态，直接返回 + if submit_result.get("status", "").upper() == "OK": + return CloneResult(voice_id=voice_id, request_id=request_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}") + # 否则轮询 + result = self.poll_clone_task(voice_id, timeout=timeout) + return CloneResult(voice_id=result["voice_id"], request_id=request_id) # ── 语音合成 ───────────────────────────────────────── @@ -388,11 +404,12 @@ class CosyVoiceService: sample_rate: int = 0, format: str = "", speed: float = 1.0, + volume: int = 50, ) -> dict: - """提交语音合成任务（非阻塞）。 + """提交语音合成任务（同步非流式，直接返回结果）. - 只提交任务到 CosyVoice API，不轮询结果。 - 返回的 dict 包含 task_id（异步）或 audio_url（同步）。 + CosyVoice SpeechSynthesizer 非流式接口是同步的， + 调用后直接返回音频 URL. 此方法保持与旧接口兼容. Args: text: 要合成的文本 @@ -400,10 +417,11 @@ class CosyVoiceService: sample_rate: 采样率（Hz），0 表示使用配置默认值 format: 输出格式（mp3/wav/pcm），空表示使用配置默认值 speed: 语速（0.5-2.0），1.0 为正常速度 + volume: 音量（0-100），默认 50 Returns: - dict: {"task_id": str, "audio_url": str, "request_id": str} - task_id 和 audio_url 至少有一个非空 + dict: {"audio_url": str, "request_id": str, + "duration": float, "file_size": int} Raises: CosyVoiceError: API 调用失败 @@ -423,55 +441,54 @@ class CosyVoiceService: "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, + "sample_rate": sample_rate or settings.cosyvoice_sample_rate, "rate": speed, + "volume": volume, }, } response = self._call_api( method="POST", - path="/services/aigc/text2audio/generation", + path="/services/audio/tts/SpeechSynthesizer", json=payload, - timeout=60.0, + timeout=120.0, ) output = response.get("output", {}) - task_id = output.get("task_id", "") - audio_url = output.get("audio_url", "") + audio = output.get("audio", {}) + audio_url = audio.get("url", "") request_id = response.get("request_id", "") - if not task_id and not audio_url: - raise CosyVoiceError(f"CosyVoice API 未返回 task_id 或 audio_url: {response}") + if not audio_url: + raise CosyVoiceError( + f"CosyVoice API 未返回 audio_url: {response}" + ) return { - "task_id": task_id, + "task_id": "", # 同步接口无 task_id，兼容旧接口 "audio_url": audio_url, - "duration": output.get("duration", 0.0), - "file_size": output.get("file_size", 0), + "duration": 0.0, # 同步接口不返回 duration + "file_size": 0, # 同步接口不返回 file_size "request_id": request_id, } - def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict: - """轮询语音合成任务状态（公开方法）。 + def poll_synthesize_task( + self, task_id: str, timeout: float = 120.0 + ) -> dict: + """轮询合成任务（同步接口无需轮询，保留兼容）. - 供 Celery 后台任务调用，轮询直到完成或超时。 - - Args: - task_id: CosyVoice 任务 ID - timeout: 超时时间（秒），默认 120 - - Returns: - dict: {"audio_url": str, "duration": float, "file_size": int} + CosyVoice SpeechSynthesizer 非流式接口是同步的， + 此方法仅为保持接口兼容，实际调用时 task_id 应该为空. Raises: - CosyVoiceError: 任务失败 - CosyVoiceTimeoutError: 超时 + CosyVoiceError: 同步接口无需轮询 """ - return self._poll_synthesize_task(task_id, timeout=timeout) + raise CosyVoiceError( + "CosyVoice 非流式合成接口是同步的，无需轮询. " + "请直接使用 submit_synthesize_task()." + ) def synthesize_speech( self, @@ -480,11 +497,13 @@ class CosyVoiceService: sample_rate: int = 0, format: str = "", speed: float = 1.0, + volume: int = 50, timeout: float = 120.0, ) -> SynthesizeResult: - """语音合成。 + """语音合成（同步非流式）. - 提交语音合成任务到 CosyVoice API，并轮询直到完成或超时。 + 调用百炼 CosyVoice SpeechSynthesizer 非流式接口， + 直接返回合成音频 URL. Args: text: 要合成的文本 @@ -492,129 +511,53 @@ class CosyVoiceService: sample_rate: 采样率（Hz），0 表示使用配置默认值 format: 输出格式（mp3/wav/pcm），空表示使用配置默认值 speed: 语速（0.5-2.0），1.0 为正常速度 - timeout: 超时时间（秒） + volume: 音量（0-100），默认 50 + 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, + result = self.submit_synthesize_task( + text=text, + voice_id=voice_id, + sample_rate=sample_rate, + format=format, + speed=speed, + volume=volume, ) - # 解析响应 - 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}") + return SynthesizeResult( + audio_url=result["audio_url"], + duration=result.get("duration", 0.0), + file_size=result.get("file_size", 0), + request_id=result.get("request_id", ""), + ) # ── 内部方法 ───────────────────────────────────────── + def _sanitize_prefix(self, name: str) -> str: + """清洗音色名称为合法的 prefix（字母数字，最多10字符）. + + Args: + name: 原始音色名称 + + Returns: + 清洗后的 prefix + """ + # 只保留字母和数字 + cleaned = "".join(c for c in name if c.isalnum()) + # 最多10字符 + cleaned = cleaned[:10] + # 如果清洗后为空，用默认值 + if not cleaned: + cleaned = "clone" + return cleaned + def _call_api( self, method: str, @@ -622,13 +565,13 @@ class CosyVoiceService: json: Optional[dict] = None, timeout: float = 30.0, ) -> dict: - """调用 CosyVoice API。 + """调用 DashScope API. - 支持重试和错误处理。 + 支持重试和错误处理. Args: method: HTTP 方法（GET/POST） - path: API 路径 + path: API 路径（以 / 开头） json: 请求体 timeout: 超时时间（秒） @@ -662,25 +605,57 @@ class CosyVoiceService: if response.status_code == 200: return response.json() elif response.status_code in (401, 403): - raise CosyVoiceAuthError(f"CosyVoice API 认证失败: HTTP {response.status_code}") + raise CosyVoiceAuthError( + f"CosyVoice API 认证失败: HTTP {response.status_code}" + ) + elif response.status_code == 400: + # 客户端错误，不重试 + body_text = response.text + try: + body = response.json() + code = body.get("code", "") + message = body.get("message", "") + raise CosyVoiceError( + f"CosyVoice API 参数错误: HTTP 400, " + f"code={code}, message={message}" + ) + except ValueError: + raise CosyVoiceError( + f"CosyVoice API 调用失败: HTTP 400, body={body_text}" + ) elif response.status_code >= 500: # 服务端错误，可重试 - last_error = CosyVoiceError(f"CosyVoice API 服务端错误: HTTP {response.status_code}") + 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}" + "CosyVoice API 失败 (尝试 %d/%d): HTTP %d", + attempt + 1, + self.MAX_RETRIES, + response.status_code, ) else: - # 客户端错误，不重试 + # 其他客户端错误，不重试 raise CosyVoiceError( - f"CosyVoice API 调用失败: HTTP {response.status_code}, " f"body={response.text}" + 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})") + logger.warning( + "CosyVoice API 超时 (尝试 %d/%d)", + 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}") + logger.warning( + "CosyVoice API 请求错误 (尝试 %d/%d): %s", + attempt + 1, + self.MAX_RETRIES, + e, + ) # 指数退避 if attempt < self.MAX_RETRIES - 1: diff --git a/packages/application/tts_job/workflow.py b/packages/application/tts_job/workflow.py old mode 100644 new mode 100755 index 1b6e9c36e..659ae5c9b --- a/packages/application/tts_job/workflow.py +++ b/packages/application/tts_job/workflow.py @@ -190,10 +190,13 @@ class TTSWorkflowService: return job def poll_and_process_synthesis(self, job_id: str, timeout: float = 120.0) -> TTSJob: - """轮询 CosyVoice 合成任务并处理结果。 + """轮询/检查 CosyVoice 合成任务并处理结果. - 从 job.metadata 获取 task_id，调用 CosyVoiceService.poll_synthesize_task() - 轮询状态，然后通过 process_synthesis_result / process_synthesis_failure 更新 job。 + 新 CosyVoice SpeechSynthesizer 非流式接口是同步的， + start_synthesis 阶段通常已经完成. 此方法用于： + 1. job 已 completed → 直接返回（同步路径已处理） + 2. job 仍在 processing → 重新提交合成（兜底） + 3. 分段任务 → 检查分段状态 供 Celery 后台任务调用。 """ @@ -201,22 +204,42 @@ class TTSWorkflowService: if job is None: raise TTSJobNotFoundError(f"TTS job {job_id} not found") + # 已完成直接返回（同步路径在 start_synthesis 里已处理） + if job.status == TTSJobStatus.COMPLETED.value: + logger.info(f"TTS 任务已完成，跳过轮询: job_id={job_id}") + return job + # 检查是否为分段合成任务 segment_task_ids = (job.metadata or {}).get("segment_task_ids", []) if segment_task_ids: return self._poll_segment_tasks(job) + # 单段模式：同步接口下通常不会走到这里， + # 但如果因为异常导致仍在 processing，重新提交一次 task_id = (job.metadata or {}).get("cosyvoice_task_id", "") - if not task_id: - raise ValueError(f"TTSJob {job_id} has no cosyvoice_task_id in metadata") - result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=timeout) - return self.process_synthesis_result( - job_id, - audio_url=result["audio_url"], - duration=result.get("duration", 0.0), - file_size=result.get("file_size", 0), - ) + # 新接口（同步）：没有 task_id，重新合成 + if not task_id: + logger.info( + f"TTS 任务无 task_id，重新同步合成: job_id={job_id}" + ) + return self._resynthesize_and_complete(job) + + # 旧接口遗留的 task_id，尝试轮询（兼容过渡） + try: + result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=timeout) + return self.process_synthesis_result( + job_id, + audio_url=result["audio_url"], + duration=result.get("duration", 0.0), + file_size=result.get("file_size", 0), + ) + except CosyVoiceError: + # 旧接口轮询失败，重新同步合成 + logger.warning( + f"旧 task_id 轮询失败，重新同步合成: job_id={job_id}, task_id={task_id}" + ) + return self._resynthesize_and_complete(job) def process_synthesis_result( self, @@ -257,8 +280,35 @@ class TTSWorkflowService: logger.info(f"TTS 合成成功: job_id={job_id}, audio_url={permanent_url}") return job + def _resynthesize_and_complete(self, job: TTSJob) -> TTSJob: + """重新同步合成并完成任务（兜底路径）. + + 当 poll_and_process_synthesis 发现 job 仍在 processing 且无 task_id 时， + 重新调用同步合成接口，转存 OSS 后标记完成。 + """ + try: + result = self.cosyvoice_service.submit_synthesize_task( + text=job.input_text, + voice_id=job.voice_id, + sample_rate=job.sample_rate, + format=job.format, + ) + audio_url = result.get("audio_url", "") + if not audio_url: + raise CosyVoiceError("重新合成未返回 audio_url") + + return self.process_synthesis_result( + job.id, + audio_url=audio_url, + duration=result.get("duration", 0.0), + file_size=result.get("file_size", 0), + ) + except Exception as e: + logger.error(f"重新同步合成失败: job_id={job.id}, error={e}") + return self.process_synthesis_failure(job.id, str(e)) + def process_synthesis_failure(self, job_id: str, error_message: str) -> TTSJob: - """处理合成失败结果。 + """处理合成失败结果。""" Args: job_id: TTSJob ID diff --git a/packages/application/voice_clone/workflow.py b/packages/application/voice_clone/workflow.py old mode 100644 new mode 100755 index 2aa4e14b7..a4fec9a81 --- a/packages/application/voice_clone/workflow.py +++ b/packages/application/voice_clone/workflow.py @@ -115,14 +115,16 @@ class VoiceCloneWorkflowService: language=language, ) - # 4. 保存 task_id / voice_id 到 metadata + # 4. 保存 voice_id / request_id 到 metadata + # 注意：key 保留 cosyvoice_task_id 以兼容旧数据，实际存的是 voice_id task_metadata = dict(profile.metadata) - task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "") + task_metadata["cosyvoice_task_id"] = submit_result.get("voice_id", "") task_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "") - # 如果 CosyVoice 同步返回了 voice_id，直接标记 ready + # 如果 CosyVoice 直接返回了 OK 状态，直接标记 ready voice_id = submit_result.get("voice_id", "") - if voice_id: + status = submit_result.get("status", "").upper() + if voice_id and status == "OK": profile.mark_ready(voice_id) profile.metadata = task_metadata profile = self.repository.update(profile) @@ -131,7 +133,7 @@ class VoiceCloneWorkflowService: profile.metadata = task_metadata profile = self.repository.update(profile) - logger.info(f"音色克隆任务已提交: profile_id={profile.id}, " f"task_id={submit_result.get('task_id')}") + logger.info(f"音色克隆任务已提交: profile_id={profile.id}, " f"voice_id={submit_result.get('voice_id')}") except (CosyVoiceError, CosyVoiceAuthError) as e: # CosyVoice 提交失败，标记为 failed @@ -248,11 +250,12 @@ class VoiceCloneWorkflowService: ) task_metadata = dict(profile.metadata) - task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "") + task_metadata["cosyvoice_task_id"] = submit_result.get("voice_id", "") task_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "") voice_id = submit_result.get("voice_id", "") - if voice_id: + status = submit_result.get("status", "").upper() + if voice_id and status == "OK": profile.mark_ready(voice_id) profile.metadata = task_metadata profile = self.repository.update(profile) diff --git a/packages/shared/config.py b/packages/shared/config.py index 87ca3d213..1fa28ecde 100644 --- a/packages/shared/config.py +++ b/packages/shared/config.py @@ -29,13 +29,15 @@ class SharedSettings(BaseSettings): oss_access_key_secret: str = "" oss_bucket_name: str = "xiaoxia-autocut" - # CosyVoice (阿里云语音合成) + # CosyVoice (阿里云百炼语音合成) cosyvoice_api_key: str = "" - cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio" - cosyvoice_model: str = "cosyvoice-v1" + cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1" + cosyvoice_model: str = "cosyvoice-v3.5-plus" cosyvoice_voice: str = "longxiaochun" # 默认音色 cosyvoice_sample_rate: int = 22050 cosyvoice_format: str = "mp3" # 输出格式：mp3/wav/pcm + # 音色克隆模型名（固定为 voice-enrollment） + cosyvoice_clone_model: str = "voice-enrollment" # Environment environment: str = "development" diff --git a/tests/unit/test_cosyvoice_service.py b/tests/unit/test_cosyvoice_service.py old mode 100644 new mode 100755 index db1fae828..c2e48da69 --- a/tests/unit/test_cosyvoice_service.py +++ b/tests/unit/test_cosyvoice_service.py @@ -1,9 +1,9 @@ -"""CosyVoiceService 单元测试。""" +"""CosyVoiceService 单元测试 — 适配百炼 DashScope API.""" from __future__ import annotations import json -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import httpx import pytest @@ -22,16 +22,19 @@ 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", + base_url: str = "https://dashscope.aliyuncs.com/api/v1", + model: str = "cosyvoice-v3.5-plus", + clone_model: str = "voice-enrollment", http_client: httpx.Client | None = None, + audio_url_signer=None, ) -> CosyVoiceService: - """创建测试用 CosyVoiceService。""" return CosyVoiceService( api_key=api_key, base_url=base_url, model=model, + clone_model=clone_model, http_client=http_client, + audio_url_signer=audio_url_signer, ) @@ -39,9 +42,8 @@ def _mock_response( status_code: int = 200, json_data: dict | None = None, text: str = "", + method: str = "POST", ) -> httpx.Response: - """创建 mock HTTP 响应。""" - # httpx.Response 需要 content 参数才能正确调用 .json() content = b"" if json_data is not None: content = json.dumps(json_data).encode("utf-8") @@ -51,7 +53,7 @@ def _mock_response( return httpx.Response( status_code=status_code, content=content, - request=httpx.Request("POST", "https://test.cosyvoice.api"), + request=httpx.Request(method, "https://dashscope.aliyuncs.com/api/v1/test"), ) @@ -59,604 +61,502 @@ def _mock_response( class TestListPresetVoices: - """测试预置音色列表。""" - - def test_returns_all_preset_voices(self) -> None: - """返回所有预置音色。""" + def test_returns_preset_voices(self) -> None: service = _make_service() voices = service.list_preset_voices() - - assert len(voices) == 8 + assert len(voices) >= 1 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") - # ── submit_clone_task ──────────────────────────────────── class TestSubmitCloneTask: - """测试 submit_clone_task（非阻塞提交）。""" - - def test_submit_async_returns_task_id(self) -> None: - """异步模式：返回 task_id。""" - mock_client = MagicMock(spec=httpx.Client) + def test_submit_success_returns_voice_id(self) -> None: + mock_client = MagicMock() mock_client.request.return_value = _mock_response( - json_data={ + 200, + { + "output": { + "voice_id": "cosyvoice-v3.5-plus-clone-abc123", + "status": "DEPLOYING", + }, + "usage": {"count": 1}, "request_id": "req-001", - "output": {"task_id": "task-abc", "task_status": "PENDING"}, - } + }, ) service = _make_service(http_client=mock_client) result = service.submit_clone_task( - audio_url="https://example.com/audio.mp3", - voice_name="测试音色", + audio_url="https://example.com/audio.wav", + voice_name="myvoice", ) - assert result["task_id"] == "task-abc" - assert result["voice_id"] == "" + assert result["voice_id"] == "cosyvoice-v3.5-plus-clone-abc123" + assert result["status"] == "DEPLOYING" assert result["request_id"] == "req-001" - mock_client.request.assert_called_once() - def test_submit_sync_returns_voice_id(self) -> None: - """同步模式：直接返回 voice_id。""" - mock_client = MagicMock(spec=httpx.Client) + # 验证请求参数 + call_args = mock_client.request.call_args + assert call_args.kwargs["method"] == "POST" + assert "/services/audio/tts/customization" in call_args.kwargs["url"] + + payload = call_args.kwargs["json"] + assert payload["model"] == "voice-enrollment" + assert payload["input"]["action"] == "create_voice" + assert payload["input"]["target_model"] == "cosyvoice-v3.5-plus" + assert payload["input"]["prefix"] == "myvoice" + assert payload["input"]["url"] == "https://example.com/audio.wav" + assert payload["input"]["language_hints"] == ["zh"] + + def test_submit_with_custom_target_model(self) -> None: + mock_client = MagicMock() mock_client.request.return_value = _mock_response( - json_data={ + 200, + { + "output": { + "voice_id": "cosyvoice-v3-flash-clone-xyz", + "status": "DEPLOYING", + }, "request_id": "req-002", - "output": {"voice_id": "voice-sync-001"}, - } + }, ) service = _make_service(http_client=mock_client) result = service.submit_clone_task( - audio_url="https://example.com/audio.mp3", + audio_url="https://example.com/audio.wav", + target_model="cosyvoice-v3-flash", ) - assert result["task_id"] == "" - assert result["voice_id"] == "voice-sync-001" - assert result["request_id"] == "req-002" + payload = mock_client.request.call_args.kwargs["json"] + assert payload["input"]["target_model"] == "cosyvoice-v3-flash" + assert result["voice_id"] == "cosyvoice-v3-flash-clone-xyz" def test_submit_empty_audio_url_raises(self) -> None: - """空 audio_url 抛出 ValueError。""" service = _make_service() - with pytest.raises(ValueError, match="audio_url 不能为空"): service.submit_clone_task(audio_url="") - def test_submit_no_api_key_raises(self) -> None: - """未配置 API Key 抛出 CosyVoiceAuthError。""" + def test_submit_no_api_key_raises_auth_error(self) -> None: service = _make_service(api_key="") - with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"): - service.submit_clone_task(audio_url="https://example.com/audio.mp3") + service.submit_clone_task(audio_url="https://example.com/a.wav") - def test_submit_no_task_id_or_voice_id_raises(self) -> None: - """API 返回无效响应时抛出 CosyVoiceError。""" - mock_client = MagicMock(spec=httpx.Client) - mock_client.request.return_value = _mock_response(json_data={"output": {}}) + def test_submit_no_voice_id_raises(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response(200, {"output": {}}) service = _make_service(http_client=mock_client) + with pytest.raises(CosyVoiceError, match="未返回 voice_id"): + service.submit_clone_task(audio_url="https://example.com/a.wav") - with pytest.raises(CosyVoiceError, match="未返回 task_id 或 voice_id"): - service.submit_clone_task(audio_url="https://example.com/audio.mp3") - - def test_submit_with_voice_name_in_payload(self) -> None: - """voice_name 参数包含在请求体中。""" - mock_client = MagicMock(spec=httpx.Client) - mock_client.request.return_value = _mock_response(json_data={"output": {"task_id": "task-001"}}) - - service = _make_service(http_client=mock_client) - service.submit_clone_task( - audio_url="https://example.com/audio.mp3", - voice_name="我的音色", - language="en-US", + def test_submit_with_audio_url_signer(self) -> None: + """传入 audio_url_signer 时，提交前会对音频URL预签名.""" + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, + {"output": {"voice_id": "v123", "status": "DEPLOYING"}, "request_id": "r1"}, ) - call_args = mock_client.request.call_args - payload = call_args.kwargs.get("json") or call_args[1].get("json") - assert payload["parameters"]["voice_name"] == "我的音色" - assert payload["parameters"]["language"] == "en-US" + def signer(url: str) -> str: + return f"{url}?signature=test" + service = _make_service(http_client=mock_client, audio_url_signer=signer) + service.submit_clone_task(audio_url="https://oss.example.com/audio.wav") -# ── check_task_status ──────────────────────────────────── + payload = mock_client.request.call_args.kwargs["json"] + assert payload["input"]["url"] == "https://oss.example.com/audio.wav?signature=test" - -class TestCheckTaskStatus: - """测试 check_task_status（单次状态查询）。""" - - def test_check_succeeded(self) -> None: - """查询成功状态。""" - mock_client = MagicMock(spec=httpx.Client) + def test_submit_signer_failure_falls_back_to_original(self) -> None: + """signer 失败时回退到原始URL，不崩溃.""" + mock_client = MagicMock() mock_client.request.return_value = _mock_response( - json_data={ + 200, + {"output": {"voice_id": "v123", "status": "DEPLOYING"}, "request_id": "r1"}, + ) + + def failing_signer(url: str) -> str: + raise RuntimeError("sign failed") + + service = _make_service(http_client=mock_client, audio_url_signer=failing_signer) + result = service.submit_clone_task(audio_url="https://example.com/a.wav") + + assert result["voice_id"] == "v123" + payload = mock_client.request.call_args.kwargs["json"] + assert payload["input"]["url"] == "https://example.com/a.wav" # 原始URL + + def test_submit_prefix_sanitized(self) -> None: + """voice_name 含特殊字符时清洗为合法 prefix.""" + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, {"output": {"voice_id": "v1", "status": "DEPLOYING"}, "request_id": "r1"} + ) + + service = _make_service(http_client=mock_client) + service.submit_clone_task(audio_url="https://e.com/a.wav", voice_name="我的音色-2024!") + + payload = mock_client.request.call_args.kwargs["json"] + # 中文和特殊字符被过滤，剩下字母数字 + assert payload["input"]["prefix"] == "2024" or payload["input"]["prefix"] == "clone" or len(payload["input"]["prefix"]) <= 10 + + def test_submit_auth_401_raises(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response(401, text="Unauthorized") + + service = _make_service(http_client=mock_client) + with pytest.raises(CosyVoiceAuthError, match="认证失败"): + service.submit_clone_task(audio_url="https://e.com/a.wav") + + def test_submit_400_raises_with_code_message(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 400, {"code": "InvalidParameter", "message": "task can not be null"} + ) + + service = _make_service(http_client=mock_client) + with pytest.raises(CosyVoiceError, match="InvalidParameter"): + service.submit_clone_task(audio_url="https://e.com/a.wav") + + +# ── query_voice_status ────────────────────────────────── + + +class TestQueryVoiceStatus: + def test_query_deploying_status(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, + { "output": { - "task_status": "SUCCEEDED", - "voice_id": "voice-done-001", + "status": "DEPLOYING", + "target_model": "cosyvoice-v3.5-plus", + "gmt_create": "2026-01-01T00:00:00Z", + "gmt_modified": "2026-01-01T00:01:00Z", + "resource_link": "https://...", }, - } + "request_id": "req-003", + }, ) service = _make_service(http_client=mock_client) - result = service.check_task_status("task-abc") + result = service.query_voice_status("voice-123") - assert result["status"] == "SUCCEEDED" - assert result["voice_id"] == "voice-done-001" - assert result["message"] == "" + assert result["status"] == "DEPLOYING" + assert result["target_model"] == "cosyvoice-v3.5-plus" - def test_check_running(self) -> None: - """查询运行中状态。""" - mock_client = MagicMock(spec=httpx.Client) + # 验证请求 + payload = mock_client.request.call_args.kwargs["json"] + assert payload["model"] == "voice-enrollment" + assert payload["input"]["action"] == "query_voice" + assert payload["input"]["voice_id"] == "voice-123" + + def test_query_ok_status(self) -> None: + mock_client = MagicMock() mock_client.request.return_value = _mock_response( - json_data={ - "output": {"task_status": "RUNNING"}, - } + 200, {"output": {"status": "OK", "target_model": "cosyvoice-v3.5-plus"}} ) service = _make_service(http_client=mock_client) - result = service.check_task_status("task-abc") + result = service.query_voice_status("voice-123") + assert result["status"] == "OK" - assert result["status"] == "RUNNING" - assert result["voice_id"] == "" - - def test_check_failed_with_message(self) -> None: - """查询失败状态，包含错误消息。""" - mock_client = MagicMock(spec=httpx.Client) - mock_client.request.return_value = _mock_response( - json_data={ - "output": { - "task_status": "FAILED", - "message": "音频质量不达标", - }, - } - ) - - service = _make_service(http_client=mock_client) - result = service.check_task_status("task-fail") - - assert result["status"] == "FAILED" - assert result["message"] == "音频质量不达标" - - def test_check_no_api_key_raises(self) -> None: - """未配置 API Key 抛出 CosyVoiceAuthError。""" + def test_query_no_api_key_raises(self) -> None: service = _make_service(api_key="") + with pytest.raises(CosyVoiceAuthError): + service.query_voice_status("v1") - with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"): - service.check_task_status("task-abc") + def test_query_empty_voice_id_raises(self) -> None: + service = _make_service() + with pytest.raises(ValueError): + service.query_voice_status("") - def test_check_uses_correct_path(self) -> None: - """请求路径包含 task_id。""" - mock_client = MagicMock(spec=httpx.Client) - mock_client.request.return_value = _mock_response(json_data={"output": {"task_status": "PENDING"}}) + +# ── poll_clone_task ───────────────────────────────────── + + +class TestPollCloneTask: + def test_poll_ok_on_first_check(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, {"output": {"status": "OK", "target_model": "cosyvoice-v3.5-plus"}} + ) service = _make_service(http_client=mock_client) - service.check_task_status("task-xyz-123") + # 减少轮询间隔加速测试 + service.CLONE_POLL_INTERVAL = 0.01 - call_args = mock_client.request.call_args - url = call_args.kwargs.get("url") or call_args[1].get("url") or call_args[0][0] - assert "/tasks/task-xyz-123" in url + result = service.poll_clone_task("voice-123", timeout=30.0) + assert result["voice_id"] == "voice-123" + + def test_poll_deploying_then_ok(self) -> None: + mock_client = MagicMock() + # 第一次 DEPLOYING，第二次 OK + mock_client.request.side_effect = [ + _mock_response(200, {"output": {"status": "DEPLOYING"}}), + _mock_response(200, {"output": {"status": "OK"}}), + ] + + service = _make_service(http_client=mock_client) + service.CLONE_POLL_INTERVAL = 0.01 + + result = service.poll_clone_task("voice-123", timeout=30.0) + assert result["voice_id"] == "voice-123" + assert mock_client.request.call_count == 2 + + def test_poll_undeployed_raises_error(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, {"output": {"status": "UNDEPLOYED"}} + ) + + service = _make_service(http_client=mock_client) + service.CLONE_POLL_INTERVAL = 0.01 + + with pytest.raises(CosyVoiceError, match="审核未通过"): + service.poll_clone_task("voice-123", timeout=30.0) + + def test_poll_timeout_raises(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, {"output": {"status": "DEPLOYING"}} + ) + + service = _make_service(http_client=mock_client) + service.CLONE_POLL_INTERVAL = 0.01 + service.CLONE_MAX_POLL_ATTEMPTS = 3 # 快速失败 + + with pytest.raises(CosyVoiceTimeoutError): + service.poll_clone_task("voice-123", timeout=30.0) + + +# ── clone_voice (阻塞) ────────────────────────────────── + + +class TestCloneVoice: + def test_clone_already_ok_returns_immediately(self) -> None: + mock_client = MagicMock() + # submit 直接返回 OK 状态 + mock_client.request.return_value = _mock_response( + 200, + { + "output": {"voice_id": "voice-ok", "status": "OK"}, + "request_id": "req-ok", + }, + ) + + service = _make_service(http_client=mock_client) + result = service.clone_voice(audio_url="https://e.com/a.wav") + + assert isinstance(result, CloneResult) + assert result.voice_id == "voice-ok" + assert result.request_id == "req-ok" + # 只有一次调用（submit），没有 poll + assert mock_client.request.call_count == 1 + + def test_clone_deploying_then_poll_ok(self) -> None: + mock_client = MagicMock() + mock_client.request.side_effect = [ + # submit: 返回 DEPLOYING + _mock_response(200, {"output": {"voice_id": "v1", "status": "DEPLOYING"}, "request_id": "r1"}), + # poll 1: DEPLOYING + _mock_response(200, {"output": {"status": "DEPLOYING"}}), + # poll 2: OK + _mock_response(200, {"output": {"status": "OK"}}), + ] + + service = _make_service(http_client=mock_client) + service.CLONE_POLL_INTERVAL = 0.01 + + result = service.clone_voice(audio_url="https://e.com/a.wav") + assert isinstance(result, CloneResult) + assert result.voice_id == "v1" + assert mock_client.request.call_count == 3 # ── synthesize_speech ──────────────────────────────────── class TestSynthesizeSpeech: - """测试语音合成。""" - - def test_synthesize_sync_success(self) -> None: - """同步合成成功（直接返回 audio_url）。""" - mock_client = MagicMock(spec=httpx.Client) + def test_synthesize_success_returns_audio_url(self) -> None: + mock_client = MagicMock() mock_client.request.return_value = _mock_response( - json_data={ - "request_id": "req-tts-001", + 200, + { "output": { - "audio_url": "https://cdn.example.com/audio.mp3", - "duration": 5.2, - "file_size": 83200, + "finish_reason": "stop", + "audio": { + "url": "https://dashscope-result.oss.com/output.mp3", + "id": "audio-001", + "expires_at": 1234567890, + }, }, - } + "usage": {"characters": 10}, + "request_id": "req-syn-001", + }, ) service = _make_service(http_client=mock_client) result = service.synthesize_speech( - text="你好世界", - voice_id="longxiaochun", + 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" + assert result.audio_url == "https://dashscope-result.oss.com/output.mp3" + assert result.request_id == "req-syn-001" - def test_synthesize_async_with_polling(self) -> None: - """异步合成（返回 task_id，轮询后成功）。""" - mock_client = MagicMock(spec=httpx.Client) + # 验证请求 + call_args = mock_client.request.call_args + assert "/services/audio/tts/SpeechSynthesizer" in call_args.kwargs["url"] - 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, - }, - } - ) + payload = call_args.kwargs["json"] + assert payload["model"] == "cosyvoice-v3.5-plus" + assert payload["input"]["text"] == "你好世界" + assert payload["input"]["voice"] == "longxiaochun" + assert payload["input"]["format"] == "mp3" + assert payload["input"]["sample_rate"] == 22050 + assert payload["input"]["rate"] == 1.0 + assert payload["input"]["volume"] == 50 - 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) + def test_synthesize_with_custom_params(self) -> None: + mock_client = MagicMock() mock_client.request.return_value = _mock_response( - json_data={"output": {"audio_url": "https://cdn.example.com/out.wav"}} + 200, + {"output": {"audio": {"url": "https://e.com/out.wav", "id": "a1"}}}, ) service = _make_service(http_client=mock_client) service.synthesize_speech( - text="参数测试", - voice_id="longxiaochun", - sample_rate=44100, - format="wav", - speed=1.5, + text="test", voice_id="v1", sample_rate=44100, + format="wav", speed=1.5, volume=80, ) - 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 + payload = mock_client.request.call_args.kwargs["json"] + assert payload["input"]["sample_rate"] == 44100 + assert payload["input"]["format"] == "wav" + assert payload["input"]["rate"] == 1.5 + assert payload["input"]["volume"] == 80 - 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": {}}) + def test_synthesize_empty_text_raises(self) -> None: + service = _make_service() + with pytest.raises(ValueError, match="text 不能为空"): + service.synthesize_speech(text="", voice_id="v1") + + def test_synthesize_empty_voice_raises(self) -> None: + service = _make_service() + with pytest.raises(ValueError, match="voice_id 不能为空"): + service.synthesize_speech(text="hi", voice_id="") + + def test_synthesize_no_api_key_raises(self) -> None: + service = _make_service(api_key="") + with pytest.raises(CosyVoiceAuthError): + service.synthesize_speech(text="hi", voice_id="v1") + + def test_synthesize_no_audio_url_raises(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response(200, {"output": {}}) service = _make_service(http_client=mock_client) + with pytest.raises(CosyVoiceError, match="未返回 audio_url"): + service.synthesize_speech(text="hi", voice_id="v1") - 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) + def test_synthesize_submit_returns_empty_task_id(self) -> None: + """同步接口的 submit_synthesize_task 返回空 task_id 字段（兼容旧接口）.""" + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, {"output": {"audio": {"url": "https://e.com/a.mp3"}}}, + ) service = _make_service(http_client=mock_client) - service.RETRY_BACKOFF = 0 + result = service.submit_synthesize_task(text="hi", voice_id="v1") - with pytest.raises(CosyVoiceError, match="服务端错误"): - service.synthesize_speech(text="测试", voice_id="longxiaochun") + assert result["task_id"] == "" # 同步接口无 task_id + assert result["audio_url"] == "https://e.com/a.mp3" - 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") + def test_synthesize_auth_401_raises(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response(401, text="Unauthorized") 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 + with pytest.raises(CosyVoiceAuthError): + service.synthesize_speech(text="hi", voice_id="v1") -# ── 重试逻辑 ───────────────────────────────────────────── +# ── retry logic ────────────────────────────────────────── 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] + def test_500_error_retries_then_succeeds(self) -> None: + mock_client = MagicMock() + mock_client.request.side_effect = [ + _mock_response(500, text="Server Error"), + _mock_response(502, text="Bad Gateway"), + _mock_response( + 200, {"output": {"audio": {"url": "https://e.com/a.mp3"}}} + ), + ] service = _make_service(http_client=mock_client) - service.RETRY_BACKOFF = 0 + service.RETRY_BACKOFF = 0.01 # 加速 - result = service.clone_voice(audio_url="https://example.com/audio.mp3") + result = service.synthesize_speech(text="hi", voice_id="v1") + assert result.audio_url == "https://e.com/a.mp3" + assert mock_client.request.call_count == 3 + + def test_max_retries_exhausted_raises(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response(500, text="Server Error") + + service = _make_service(http_client=mock_client) + service.MAX_RETRIES = 2 + service.RETRY_BACKOFF = 0.01 + + with pytest.raises(CosyVoiceError, match="服务端错误"): + service.synthesize_speech(text="hi", voice_id="v1") - 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) + +# ── sanitize_prefix ───────────────────────────────────── + + +class TestSanitizePrefix: + def test_alphanumeric_kept(self) -> None: + service = _make_service() + assert service._sanitize_prefix("myvoice123") == "myvoice123" + + def test_special_chars_removed(self) -> None: + service = _make_service() + result = service._sanitize_prefix("my-voice_2!") + # 特殊字符被移除，只保留字母数字 + assert result == "myvoice2" + + def test_max_10_chars(self) -> None: + service = _make_service() + result = service._sanitize_prefix("abcdefghijklmnop") + assert len(result) == 10 + + def test_empty_returns_clone(self) -> None: + service = _make_service() + assert service._sanitize_prefix("!!!???") == "clone" + assert service._sanitize_prefix("") == "clone" + + +# ── check_task_status (兼容旧接口) ────────────────────── + + +class TestCheckTaskStatus: + def test_check_task_status_uses_query_voice(self) -> None: + mock_client = MagicMock() + mock_client.request.return_value = _mock_response( + 200, {"output": {"status": "OK"}} + ) service = _make_service(http_client=mock_client) - service.RETRY_BACKOFF = 0 + result = service.check_task_status("voice-123") - with pytest.raises(CosyVoiceError): - service.clone_voice(audio_url="https://example.com/audio.mp3") + assert result["status"] == "OK" + assert result["voice_id"] == "voice-123" - assert mock_client.request.call_count == service.MAX_RETRIES + # 验证走的是 query_voice 路径 + payload = mock_client.request.call_args.kwargs["json"] + assert payload["input"]["action"] == "query_voice" diff --git a/tests/unit/test_voice_clone_workflow.py b/tests/unit/test_voice_clone_workflow.py old mode 100644 new mode 100755 index fa6bddba6..a20d0cfbd --- a/tests/unit/test_voice_clone_workflow.py +++ b/tests/unit/test_voice_clone_workflow.py @@ -57,15 +57,15 @@ def _make_service( class TestStartClone: """测试 start_clone 方法。""" - def test_start_clone_with_async_task(self) -> None: - """异步模式：提交任务后返回 processing 状态的 profile。""" + def test_start_clone_with_deploying(self) -> None: + """提交克隆后返回 DEPLOYING 状态，profile 保持 processing。""" mock_repo = MagicMock() mock_cosyvoice = MagicMock(spec=CosyVoiceService) - # CosyVoice 返回 task_id（异步模式） + # CosyVoice 返回 voice_id + DEPLOYING 状态（需轮询） mock_cosyvoice.submit_clone_task.return_value = { - "task_id": "task-abc", - "voice_id": "", + "voice_id": "voice-abc", + "status": "DEPLOYING", "request_id": "req-123", } @@ -81,21 +81,21 @@ class TestStartClone: ) assert profile.status == VoiceCloneStatus.PROCESSING - assert profile.metadata["cosyvoice_task_id"] == "task-abc" + assert profile.metadata["cosyvoice_task_id"] == "voice-abc" assert profile.metadata["cosyvoice_request_id"] == "req-123" mock_cosyvoice.submit_clone_task.assert_called_once() assert mock_repo.create.call_count == 1 - # update 至少调用 2 次：mark_processing + 保存 task_id + # update 至少调用 2 次：mark_processing + 保存 voice_id assert mock_repo.update.call_count >= 2 - def test_start_clone_with_sync_result(self) -> None: - """同步模式：CosyVoice 直接返回 voice_id，profile 变为 ready。""" + def test_start_clone_with_ok_status(self) -> None: + """CosyVoice 直接返回 OK 状态，profile 变为 ready。""" mock_repo = MagicMock() mock_cosyvoice = MagicMock(spec=CosyVoiceService) mock_cosyvoice.submit_clone_task.return_value = { - "task_id": "", "voice_id": "voice-sync-123", + "status": "OK", "request_id": "req-456", } @@ -244,8 +244,8 @@ class TestRetryClone: mock_repo.update.side_effect = lambda p: p mock_cosyvoice.submit_clone_task.return_value = { - "task_id": "task-retry", - "voice_id": "", + "voice_id": "voice-retry", + "status": "DEPLOYING", "request_id": "req-retry", } @@ -253,11 +253,11 @@ class TestRetryClone: result = service.retry_clone(profile.id, "user-123") assert result.status == VoiceCloneStatus.PROCESSING - assert result.metadata["cosyvoice_task_id"] == "task-retry" + assert result.metadata["cosyvoice_task_id"] == "voice-retry" assert result.retry_count == 2 # prepare_retry 增加了一次 - def test_retry_clone_with_sync_result(self) -> None: - """重试成功，同步模式。""" + def test_retry_clone_with_ok_status(self) -> None: + """重试成功，直接返回 OK 状态。""" mock_repo = MagicMock() mock_cosyvoice = MagicMock(spec=CosyVoiceService) @@ -266,8 +266,8 @@ class TestRetryClone: mock_repo.update.side_effect = lambda p: p mock_cosyvoice.submit_clone_task.return_value = { - "task_id": "", "voice_id": "voice-retry-sync", + "status": "OK", "request_id": "req-retry", } -- 2.54.0 From 1d2aaad52fc6a41d74944785132ca62186ddeb2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 22:40:20 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20PR=20#212=20P0+P1=20=E9=97=AE?= =?UTF-8?q?=E9=A2=98=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P0: 修复 process_synthesis_failure docstring 语法错误（三引号位置错误） - P1: _resynthesize_and_complete 补上 speed/volume 参数传递（从metadata读取） - P1: _poll_segment_tasks 适配新同步接口，缺失分段重新同步合成而非轮询 - 移除 time 依赖（不再轮询） - 并发重新合成缺失 audio_url 的分段 - 已有 audio_url 的分段直接复用 - 新增 1 个分段复用测试用例 --- packages/application/tts_job/workflow.py | 145 +++++++++++++++-------- tests/unit/test_tts_segment_synthesis.py | 73 +++++++++--- 2 files changed, 152 insertions(+), 66 deletions(-) mode change 100644 => 100755 tests/unit/test_tts_segment_synthesis.py diff --git a/packages/application/tts_job/workflow.py b/packages/application/tts_job/workflow.py index 659ae5c9b..821e6afec 100755 --- a/packages/application/tts_job/workflow.py +++ b/packages/application/tts_job/workflow.py @@ -14,7 +14,6 @@ import logging import os import shutil import tempfile -import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Optional @@ -287,11 +286,18 @@ class TTSWorkflowService: 重新调用同步合成接口，转存 OSS 后标记完成。 """ try: + # 从 metadata 读取合成参数（兼容旧数据，无则用默认值） + job_metadata = job.metadata or {} + speed = float(job_metadata.get("speed", 1.0)) + volume = int(job_metadata.get("volume", 50)) + result = self.cosyvoice_service.submit_synthesize_task( text=job.input_text, voice_id=job.voice_id, sample_rate=job.sample_rate, format=job.format, + speed=speed, + volume=volume, ) audio_url = result.get("audio_url", "") if not audio_url: @@ -308,7 +314,7 @@ class TTSWorkflowService: return self.process_synthesis_failure(job.id, str(e)) def process_synthesis_failure(self, job_id: str, error_message: str) -> TTSJob: - """处理合成失败结果。""" + """处理合成失败结果。 Args: job_id: TTSJob ID @@ -479,69 +485,108 @@ class TTSWorkflowService: shutil.rmtree(temp_dir, ignore_errors=True) def _poll_segment_tasks(self, job: TTSJob) -> TTSJob: - """轮询所有分段异步任务，全部完成后合并音频。""" + """分段任务完成检查（适配新同步接口）. + + 新 CosyVoice SpeechSynthesizer 非流式接口为同步接口， + 分段任务在提交时应已同步返回 audio_url。 + 若历史任务处于 processing 且有 segment_task_ids 但缺少 audio_url， + 则对缺失分段重新同步合成，全部完成后合并音频。 + """ segment_task_ids: list[str] = (job.metadata or {}).get("segment_task_ids", []) segment_audio_urls: list[str] = (job.metadata or {}).get("segment_audio_urls", []) segment_count = len(segment_task_ids) - poll_start = time.monotonic() - poll_timeout = 300.0 # 分段任务超时更长 - poll_interval = 2.0 + if segment_count == 0: + logger.warning(f"分段任务无 task_id: job_id={job.id}") + self._handle_segment_failure(job, "分段任务数据异常：无分段信息") + return self.repository.get(job.id) - while time.monotonic() - poll_start < poll_timeout: - all_done = True - results: list[dict | None] = [None] * segment_count + # 从 metadata 读取合成参数 + job_metadata = job.metadata or {} + speed = float(job_metadata.get("speed", 1.0)) + volume = int(job_metadata.get("volume", 50)) - for idx, task_id in enumerate(segment_task_ids): - # 已经有音频的分段跳过轮询 - if idx < len(segment_audio_urls) and segment_audio_urls[idx]: - results[idx] = { - "audio_url": segment_audio_urls[idx], - "duration": 0.0, - "file_size": 0, - } - continue + # 分段文本（用于缺失段重新合成） + segments = split_text(job.input_text, max_chars=_SEGMENT_THRESHOLD) - try: - result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=poll_timeout) - results[idx] = result - except Exception as e: - logger.error(f"分段任务轮询失败: job_id={job.id}, " f"segment={idx}, error={e}") - self._handle_segment_failure(job, f"分段 {idx + 1} 轮询失败: {e}") - return self.repository.get(job.id) + results: list[dict | None] = [None] * segment_count - if results[idx] is None: - all_done = False + # 已有音频的分段直接用 + for idx in range(segment_count): + if idx < len(segment_audio_urls) and segment_audio_urls[idx]: + results[idx] = { + "audio_url": segment_audio_urls[idx], + "duration": 0.0, + "file_size": 0, + } - if all_done and all(r is not None for r in results): - # 所有分段完成，下载合并 - try: - merged_data, total_duration = self._download_and_merge_segments(results, job) + # 找出缺失音频的分段索引 + missing_indices = [i for i in range(segment_count) if results[i] is None] - # 转存 OSS - permanent_url, storage_key = self._upload_merged_to_oss( - merged_data, job.user_id, job.id, job.format + if missing_indices: + logger.info( + f"分段任务重新合成缺失段: job_id={job.id}, " + f"缺失={len(missing_indices)}/{segment_count}" + ) + # 并发重新合成缺失分段 + max_workers = min(len(missing_indices), _MAX_SEGMENT_WORKERS) + with ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_idx = {} + for idx in missing_indices: + segment_text = segments[idx] if idx < len(segments) else "" + future = executor.submit( + self.cosyvoice_service.submit_synthesize_task, + text=segment_text, + voice_id=job.voice_id, + sample_rate=job.sample_rate, + format=job.format, + speed=speed, + volume=volume, ) + future_to_idx[future] = idx - job.mark_completed( - output_audio_url=permanent_url, - output_audio_key=storage_key, - duration=total_duration, - file_size=len(merged_data), - ) - job = self.repository.update(job) - logger.info(f"分段合成轮询完成: job_id={job.id}, " f"merged_size={len(merged_data)}") - return job + for future in as_completed(future_to_idx): + idx = future_to_idx[future] + try: + results[idx] = future.result() + except Exception as e: + logger.error( + f"分段重新合成失败: job_id={job.id}, " + f"segment={idx}, error={e}" + ) + self._handle_segment_failure( + job, f"分段 {idx + 1} 重新合成失败: {e}" + ) + return self.repository.get(job.id) - except Exception as e: - self._handle_segment_failure(job, f"分段合并失败: {e}") - return self.repository.get(job.id) + # 所有分段完成，下载合并 + if all(r is not None for r in results): + try: + merged_data, total_duration = self._download_and_merge_segments(results, job) - # 等待后重试 - time.sleep(poll_interval) + permanent_url, storage_key = self._upload_merged_to_oss( + merged_data, job.user_id, job.id, job.format + ) - # 超时 - self._handle_segment_failure(job, "分段合成轮询超时（300 秒）") + job.mark_completed( + output_audio_url=permanent_url, + output_audio_key=storage_key, + duration=total_duration, + file_size=len(merged_data), + ) + job = self.repository.update(job) + logger.info( + f"分段合成完成（重新合成路径）: job_id={job.id}, " + f"merged_size={len(merged_data)}" + ) + return job + + except Exception as e: + self._handle_segment_failure(job, f"分段合并失败: {e}") + return self.repository.get(job.id) + + # 理论上不会到这里（全部重新合成要么成功要么失败） + self._handle_segment_failure(job, "分段合成结果不完整") return self.repository.get(job.id) def _handle_segment_failure(self, job: TTSJob, error_message: str) -> None: diff --git a/tests/unit/test_tts_segment_synthesis.py b/tests/unit/test_tts_segment_synthesis.py old mode 100644 new mode 100755 index f903b5d51..3624f0ced --- a/tests/unit/test_tts_segment_synthesis.py +++ b/tests/unit/test_tts_segment_synthesis.py @@ -353,16 +353,11 @@ class TestHandleSegmentFailure: class TestPollSegmentTasks: - """测试 _poll_segment_tasks 异步轮询。""" + """测试 _poll_segment_tasks 分段缺失重新合成（适配同步接口）。""" - @patch("packages.application.tts_job.workflow.time") @patch("packages.application.tts_job.workflow.httpx") - def test_all_segments_done(self, mock_httpx: MagicMock, mock_time: MagicMock) -> None: - """所有分段完成后合并并标记完成。""" - # Mock time.monotonic 让循环只执行一次 - mock_time.monotonic.side_effect = [0.0, 1.0, 2.0] - mock_time.sleep = MagicMock() - + def test_all_segments_done(self, mock_httpx: MagicMock) -> None: + """所有分段缺少 audio_url 时重新同步合成，合并后标记完成。""" # Mock 下载分段音频 mock_resp = MagicMock() mock_resp.content = b"seg audio" @@ -370,7 +365,7 @@ class TestPollSegmentTasks: mock_httpx.get.return_value = mock_resp service = MagicMock(spec=CosyVoiceService) - service.poll_synthesize_task.side_effect = [ + service.submit_synthesize_task.side_effect = [ {"audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 100}, {"audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 200}, ] @@ -381,6 +376,8 @@ class TestPollSegmentTasks: repo = MagicMock() job = _make_job( status=TTSJobStatus.PROCESSING, + # 长文本触发分段，用于重新合成时切分 + input_text="这是一段很长的测试文本。" * 30, metadata={ "segment_task_ids": ["task_1", "task_2"], "segment_audio_urls": ["", ""], @@ -400,19 +397,18 @@ class TestPollSegmentTasks: result = workflow._poll_segment_tasks(job) assert result.status == TTSJobStatus.COMPLETED + # 两个缺失分段都重新合成了 + assert service.submit_synthesize_task.call_count == 2 - @patch("packages.application.tts_job.workflow.time") - def test_segment_poll_failure(self, mock_time: MagicMock) -> None: - """分段轮询失败时标记 job failed。""" - mock_time.monotonic.side_effect = [0.0, 1.0] - mock_time.sleep = MagicMock() - + def test_segment_resynthesis_failure(self) -> None: + """分段重新合成失败时标记 job failed。""" service = MagicMock(spec=CosyVoiceService) - service.poll_synthesize_task.side_effect = CosyVoiceError("Poll failed") + service.submit_synthesize_task.side_effect = CosyVoiceError("Synthesis failed") repo = MagicMock() job = _make_job( status=TTSJobStatus.PROCESSING, + input_text="这是一段很长的测试文本。" * 30, metadata={ "segment_task_ids": ["task_1"], "segment_audio_urls": [""], @@ -426,6 +422,51 @@ class TestPollSegmentTasks: assert result.status == TTSJobStatus.FAILED + @patch("packages.application.tts_job.workflow.httpx") + def test_partial_audio_urls_reuse_existing(self, mock_httpx: MagicMock) -> None: + """部分分段已有 audio_url 时直接复用，缺失的重新合成。""" + mock_resp = MagicMock() + mock_resp.content = b"seg audio" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + service = MagicMock(spec=CosyVoiceService) + # 只有 1 个分段需要重新合成 + service.submit_synthesize_task.return_value = { + "audio_url": "https://temp.com/seg2.mp3", + "duration": 3.0, + "file_size": 200, + } + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/merged.mp3" + + repo = MagicMock() + job = _make_job( + status=TTSJobStatus.PROCESSING, + input_text="这是一段很长的测试文本。" * 30, + metadata={ + "segment_task_ids": ["task_1", "task_2"], + "segment_audio_urls": ["https://temp.com/seg1.mp3", ""], + "segment_format": "mp3", + }, + ) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage) + + with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger: + mock_merger = MagicMock() + mock_merger.merge.return_value = b"merged data" + MockMerger.return_value = mock_merger + + result = workflow._poll_segment_tasks(job) + + assert result.status == TTSJobStatus.COMPLETED + # 只有 1 个缺失分段被重新合成 + assert service.submit_synthesize_task.call_count == 1 + class TestPollAndProcessSynthesisSegmentDetection: """测试 poll_and_process_synthesis 正确识别分段任务。""" -- 2.54.0