diff --git a/apps/api/app/api/routes/voice_clones.py b/apps/api/app/api/routes/voice_clones.py index 7856aae4e..66ef938fd 100644 --- a/apps/api/app/api/routes/voice_clones.py +++ b/apps/api/app/api/routes/voice_clones.py @@ -27,7 +27,10 @@ from packages.application.voice_clone.use_cases import ( VoiceCloneNotFoundError, VoiceCloneNotRetryableError, ) -from packages.application.voice_clone.workflow import VoiceCloneWorkflowService +from packages.application.voice_clone.workflow import ( + VoiceCloneWorkflowError, + VoiceCloneWorkflowService, +) logger = logging.getLogger(__name__) @@ -104,6 +107,15 @@ def create_voice_clone( logger.info(f"Celery task dispatched for voice clone {profile.id}") except Exception as e: logger.error(f"Failed to dispatch Celery task: {e}") + # P2-3: Celery 调度失败时标记 profile 为 failed，避免永久卡在 processing + try: + workflow.process_clone_failure( + profile.id, f"Celery 任务调度失败: {e}" + ) + except Exception as inner_e: + logger.error( + f"Failed to mark profile as failed after dispatch error: {inner_e}" + ) return _to_response(profile) @@ -232,5 +244,14 @@ def retry_voice_clone( logger.info(f"Celery task dispatched for voice clone retry {profile.id}") except Exception as e: logger.error(f"Failed to dispatch Celery task: {e}") + # P2-3: Celery 调度失败时标记 profile 为 failed，避免永久卡在 processing + try: + workflow.process_clone_failure( + profile.id, f"Celery 任务调度失败: {e}" + ) + except Exception as inner_e: + logger.error( + f"Failed to mark profile as failed after dispatch error: {inner_e}" + ) return _to_response(profile) diff --git a/apps/worker/worker_app/tasks/voice_clone.py b/apps/worker/worker_app/tasks/voice_clone.py index de3138e46..0924f9b74 100644 --- a/apps/worker/worker_app/tasks/voice_clone.py +++ b/apps/worker/worker_app/tasks/voice_clone.py @@ -1,7 +1,6 @@ """Voice clone tasks - process voice clone requests via CosyVoice API.""" import logging -from typing import Optional from celery import Task from celery.exceptions import Retry @@ -16,6 +15,7 @@ from packages.application.cosyvoice_service import ( CosyVoiceService, CosyVoiceTimeoutError, ) +from packages.application.voice_clone.workflow import VoiceCloneWorkflowService logger = logging.getLogger(__name__) @@ -24,7 +24,8 @@ logger = logging.getLogger(__name__) def process_voice_clone(self: Task, profile_id: str) -> dict: """处理音色克隆任务。 - 轮询 CosyVoice 克隆任务状态，更新 VoiceCloneProfile。 + 通过 VoiceCloneWorkflowService.poll_and_process_clone() 轮询 CosyVoice + 克隆任务状态，更新 VoiceCloneProfile。 失败时自动重试（最多 2 次）。 Args: @@ -36,31 +37,22 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: session = SessionLocal() try: repo = SQLAlchemyVoiceCloneProfileRepository(session) - profile = repo.get(profile_id) - if profile is None: - raise ValueError(f"VoiceCloneProfile {profile_id} not found") + workflow = VoiceCloneWorkflowService( + repository=repo, cosyvoice_service=CosyVoiceService(), + ) - # 获取 CosyVoice task_id - task_id = (profile.metadata or {}).get("cosyvoice_task_id", "") - if not task_id: - raise ValueError( - f"VoiceCloneProfile {profile_id} has no cosyvoice_task_id in metadata" - ) - - # 轮询 CosyVoice 任务状态 - service = CosyVoiceService() - result = service._poll_clone_task(task_id, timeout=300) - voice_id = result["voice_id"] - - # 更新 profile 状态为 ready - profile.mark_ready(voice_id) - repo.update(profile) + updated_profile = workflow.poll_and_process_clone(profile_id, timeout=300) session.commit() logger.info( - f"Voice clone completed: profile_id={profile_id}, voice_id={voice_id}" + f"Voice clone completed: profile_id={profile_id}, " + f"voice_id={updated_profile.voice_id}" ) - return {"ok": True, "profile_id": profile_id, "voice_id": voice_id} + return { + "ok": True, + "profile_id": profile_id, + "voice_id": updated_profile.voice_id, + } except Retry: # Celery Retry 异常必须向上传播，不能被后续 except 捕获 @@ -75,9 +67,8 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: except CosyVoiceError as e: logger.error(f"Voice clone failed for {profile_id}: {e}") session.rollback() - # API 错误，标记为 failed + # 标记 profile 为 failed try: - repo = SQLAlchemyVoiceCloneProfileRepository(session) profile = repo.get(profile_id) if profile is not None: profile.mark_failed(str(e)) @@ -91,9 +82,8 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: except Exception as e: logger.error(f"Voice clone unexpected error for {profile_id}: {e}") session.rollback() - # 未知错误，标记为 failed + # 标记 profile 为 failed try: - repo = SQLAlchemyVoiceCloneProfileRepository(session) profile = repo.get(profile_id) if profile is not None: profile.mark_failed(str(e)) diff --git a/packages/application/cosyvoice_service.py b/packages/application/cosyvoice_service.py index b44658d9a..15ce4dce7 100644 --- a/packages/application/cosyvoice_service.py +++ b/packages/application/cosyvoice_service.py @@ -240,6 +240,24 @@ class CosyVoiceService: "message": message, } + def poll_clone_task(self, task_id: str, timeout: float = 300.0) -> dict: + """轮询音色克隆任务状态（公开方法）。 + + 供 Celery 后台任务调用，轮询直到完成或超时。 + + Args: + task_id: CosyVoice 任务 ID + timeout: 超时时间（秒），默认 300 + + Returns: + dict: {"voice_id": str} + + Raises: + CosyVoiceError: 任务失败 + CosyVoiceTimeoutError: 超时 + """ + return self._poll_clone_task(task_id, timeout=timeout) + def clone_voice( self, audio_url: str, diff --git a/packages/application/voice_clone/workflow.py b/packages/application/voice_clone/workflow.py index c9dce3d9a..52dcd52fa 100644 --- a/packages/application/voice_clone/workflow.py +++ b/packages/application/voice_clone/workflow.py @@ -66,10 +66,10 @@ class VoiceCloneWorkflowService: """启动音色克隆流程。 1. 创建 VoiceCloneProfile (pending) - 2. 标记为 processing - 3. 提交 CosyVoice 克隆任务 - 4. 保存 task_id 到 metadata - 5. 返回 profile（Celery task 由调用方触发） + 2. 若有音频 URL：标记 processing → 提交 CosyVoice 克隆任务 + 若无音频 URL：保持 pending，等待用户上传 + 3. 保存 task_id 到 metadata + 4. 返回 profile（Celery task 由调用方触发） Args: user_id: 用户 ID @@ -83,7 +83,7 @@ class VoiceCloneWorkflowService: metadata: 扩展元数据 Returns: - VoiceCloneProfile: 已创建的 profile（状态为 processing） + VoiceCloneProfile: 已创建的 profile（有 URL 时为 processing，无 URL 时为 pending） Raises: VoiceCloneWorkflowError: CosyVoice 提交失败 @@ -102,12 +102,12 @@ class VoiceCloneWorkflowService: metadata=metadata, ) - # 2. 标记为 processing - profile.mark_processing() - profile = self.repository.update(profile) - - # 3. 提交 CosyVoice 克隆任务 + # 2. 提交 CosyVoice 克隆任务（仅有音频 URL 时才标记 processing） if source_audio_url: + # 标记为 processing + profile.mark_processing() + profile = self.repository.update(profile) + try: submit_result = self.cosyvoice_service.submit_clone_task( audio_url=source_audio_url, @@ -152,13 +152,34 @@ class VoiceCloneWorkflowService: logger.error(f"音色克隆参数错误: profile_id={profile.id}, error={e}") return profile else: - # 没有音频 URL，保持 processing 状态等待用户上传 + # 没有音频 URL，保持 pending 状态等待用户上传 logger.info( - f"音色克隆已创建但无音频URL: profile_id={profile.id}" + f"音色克隆已创建但无音频URL，保持pending: profile_id={profile.id}" ) return profile + def poll_and_process_clone( + self, profile_id: str, timeout: float = 300.0 + ) -> VoiceCloneProfile: + """轮询 CosyVoice 克隆任务并处理结果。 + + 从 profile.metadata 获取 task_id，调用 CosyVoiceService.poll_clone_task() + 轮询状态，然后通过 process_clone_result / process_clone_failure 更新 profile。 + + 供 Celery 后台任务调用，避免直接访问私有方法。 + """ + profile = self.repository.get(profile_id) + if profile is None: + raise VoiceCloneNotFoundError(f"Voice clone {profile_id} not found") + task_id = (profile.metadata or {}).get("cosyvoice_task_id", "") + if not task_id: + raise ValueError( + f"VoiceCloneProfile {profile_id} has no cosyvoice_task_id in metadata" + ) + result = self.cosyvoice_service.poll_clone_task(task_id, timeout=timeout) + return self.process_clone_result(profile_id, result["voice_id"]) + def process_clone_result(self, profile_id: str, voice_id: str) -> VoiceCloneProfile: """处理克隆成功结果。 diff --git a/tests/unit/test_voice_clone_task.py b/tests/unit/test_voice_clone_task.py index 77463b821..30020c542 100644 --- a/tests/unit/test_voice_clone_task.py +++ b/tests/unit/test_voice_clone_task.py @@ -77,7 +77,7 @@ class TestProcessVoiceCloneSuccess: mock_repo.update.side_effect = lambda p: p mock_repo_cls.return_value = mock_repo - mock_service._poll_clone_task.return_value = {"voice_id": "voice-xyz"} + mock_service.poll_clone_task.return_value = {"voice_id": "voice-xyz"} mock_service_cls.return_value = mock_service _mock_db_module.SessionLocal.return_value = mock_session @@ -89,7 +89,7 @@ class TestProcessVoiceCloneSuccess: assert result["ok"] is True assert result["voice_id"] == "voice-xyz" - mock_service._poll_clone_task.assert_called_once_with("task-abc", timeout=300) + mock_service.poll_clone_task.assert_called_once_with("task-abc", timeout=300) mock_session.commit.assert_called_once() mock_session.close.assert_called_once() @@ -135,7 +135,7 @@ class TestProcessVoiceCloneTimeout: mock_repo.get.return_value = profile mock_repo_cls.return_value = mock_repo - mock_service._poll_clone_task.side_effect = CosyVoiceTimeoutError("任务超时") + mock_service.poll_clone_task.side_effect = CosyVoiceTimeoutError("任务超时") mock_service_cls.return_value = mock_service _mock_db_module.SessionLocal.return_value = mock_session @@ -172,7 +172,7 @@ class TestProcessVoiceCloneFailure: mock_repo.update.side_effect = lambda p: p mock_repo_cls.return_value = mock_repo - mock_service._poll_clone_task.side_effect = CosyVoiceError("克隆失败") + mock_service.poll_clone_task.side_effect = CosyVoiceError("克隆失败") mock_service_cls.return_value = mock_service _mock_db_module.SessionLocal.return_value = mock_session @@ -200,7 +200,7 @@ class TestProcessVoiceCloneFailure: mock_repo.update.side_effect = lambda p: p mock_repo_cls.return_value = mock_repo - mock_service._poll_clone_task.side_effect = RuntimeError("未知错误") + mock_service.poll_clone_task.side_effect = RuntimeError("未知错误") mock_service_cls.return_value = mock_service _mock_db_module.SessionLocal.return_value = mock_session diff --git a/tests/unit/test_voice_clone_workflow.py b/tests/unit/test_voice_clone_workflow.py index cf76bd85e..f5eb7cc5b 100644 --- a/tests/unit/test_voice_clone_workflow.py +++ b/tests/unit/test_voice_clone_workflow.py @@ -153,7 +153,7 @@ class TestStartClone: assert "认证失败" in profile.error_message def test_start_clone_without_audio_url(self) -> None: - """没有音频 URL 时，profile 保持 processing 状态。""" + """没有音频 URL 时，profile 保持 pending 状态（P2-1 修复后）。""" mock_repo = MagicMock() mock_cosyvoice = MagicMock(spec=CosyVoiceService) @@ -167,8 +167,8 @@ class TestStartClone: source_audio_url="", ) - # 没有音频 URL，保持 processing 状态 - assert profile.status == VoiceCloneStatus.PROCESSING + # P2-1: 没有音频 URL 时不标记 processing，保持 pending + assert profile.status == VoiceCloneStatus.PENDING mock_cosyvoice.submit_clone_task.assert_not_called() @@ -320,3 +320,71 @@ class TestRetryClone: assert result.status == VoiceCloneStatus.FAILED assert "重试失败" in result.error_message + + +# ── poll_and_process_clone ─────────────────────────────── + + +class TestPollAndProcessClone: + """测试 poll_and_process_clone 方法（P2-2 修复）。""" + + def test_poll_and_process_clone_success(self) -> None: + """轮询成功：调用 poll_clone_task → process_clone_result。""" + mock_repo = MagicMock() + mock_cosyvoice = MagicMock(spec=CosyVoiceService) + + profile = _make_profile( + status=VoiceCloneStatus.PROCESSING, + metadata={"cosyvoice_task_id": "task-abc"}, + ) + mock_repo.get.return_value = profile + mock_repo.update.side_effect = lambda p: p + + mock_cosyvoice.poll_clone_task.return_value = {"voice_id": "voice-poll-xyz"} + + service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice) + result = service.poll_and_process_clone(profile.id) + + assert result.status == VoiceCloneStatus.READY + assert result.voice_id == "voice-poll-xyz" + mock_cosyvoice.poll_clone_task.assert_called_once_with("task-abc", timeout=300) + + def test_poll_and_process_clone_no_task_id(self) -> None: + """metadata 中没有 task_id 时抛出 ValueError。""" + mock_repo = MagicMock() + mock_cosyvoice = MagicMock(spec=CosyVoiceService) + + profile = _make_profile(status=VoiceCloneStatus.PROCESSING, metadata={}) + mock_repo.get.return_value = profile + + service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice) + with pytest.raises(ValueError, match="cosyvoice_task_id"): + service.poll_and_process_clone(profile.id) + + def test_poll_and_process_clone_not_found(self) -> None: + """profile 不存在时抛出 VoiceCloneNotFoundError。""" + mock_repo = MagicMock() + mock_repo.get.return_value = None + + service = _make_service(repo=mock_repo) + with pytest.raises(VoiceCloneNotFoundError): + service.poll_and_process_clone("nonexistent") + + def test_poll_and_process_clone_timeout(self) -> None: + """超时时透传 CosyVoiceTimeoutError（由 Celery task 捕获重试）。""" + from packages.application.cosyvoice_service import CosyVoiceTimeoutError + + mock_repo = MagicMock() + mock_cosyvoice = MagicMock(spec=CosyVoiceService) + + profile = _make_profile( + status=VoiceCloneStatus.PROCESSING, + metadata={"cosyvoice_task_id": "task-abc"}, + ) + mock_repo.get.return_value = profile + + mock_cosyvoice.poll_clone_task.side_effect = CosyVoiceTimeoutError("超时") + + service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice) + with pytest.raises(CosyVoiceTimeoutError): + service.poll_and_process_clone(profile.id)