"""Voice clone workflow orchestration — Phase 3. 编排音色克隆的完整流程： 1. 创建 VoiceCloneProfile 2. 提交 CosyVoice 克隆任务 3. 处理克隆结果（成功/失败） 4. 重试失败的克隆 """ from __future__ import annotations import logging from typing import Optional from packages.application.cosyvoice_service import ( CosyVoiceAuthError, CosyVoiceError, CosyVoiceService, ) from packages.application.voice_clone.use_cases import ( CreateVoiceCloneUseCase, RetryVoiceCloneUseCase, VoiceCloneNotFoundError, ) from packages.domain.voice_clone_profile import VoiceCloneProfile from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository from packages.shared.url_security import UrlSecurityError, validate_url_safety logger = logging.getLogger(__name__) class VoiceCloneWorkflowError(Exception): """音色克隆工作流异常。""" pass class VoiceCloneWorkflowService: """音色克隆工作流编排服务。 协调 CreateVoiceCloneUseCase + CosyVoiceService， 实现完整的克隆生命周期管理。 """ def __init__( self, repository: VoiceCloneProfileRepository, cosyvoice_service: CosyVoiceService, ) -> None: self.repository = repository self.cosyvoice_service = cosyvoice_service def start_clone( self, user_id: str, name: str, *, description: str = "", source_audio_url: str = "", voice_model: str = "", language: str = "zh-CN", gender: str = "unknown", max_retries: int = 3, metadata: Optional[dict] = None, ) -> VoiceCloneProfile: """启动音色克隆流程。 1. 创建 VoiceCloneProfile (pending) 2. 若有音频 URL：标记 processing → 提交 CosyVoice 克隆任务 若无音频 URL：保持 pending，等待用户上传 3. 保存 task_id 到 metadata 4. 返回 profile（Celery task 由调用方触发） Args: user_id: 用户 ID name: 音色名称 description: 描述 source_audio_url: 参考音频 URL voice_model: 模型名称 language: 语言 gender: 性别 max_retries: 最大重试次数 metadata: 扩展元数据 Returns: VoiceCloneProfile: 已创建的 profile（有 URL 时为 processing，无 URL 时为 pending） Raises: VoiceCloneWorkflowError: CosyVoice 提交失败 """ # 1. 创建 profile create_use_case = CreateVoiceCloneUseCase(self.repository) profile = create_use_case.execute( user_id=user_id, name=name, description=description, source_audio_url=source_audio_url, voice_model=voice_model, language=language, gender=gender, max_retries=max_retries, metadata=metadata, ) # 2. 提交 CosyVoice 克隆任务（仅有音频 URL 时才标记 processing） if source_audio_url: # SSRF 防护：校验音频 URL 安全性 try: source_audio_url = validate_url_safety(source_audio_url, purpose="download") except UrlSecurityError as e: profile.mark_failed(f"音频URL安全校验失败: {e}") profile = self.repository.update(profile) logger.warning(f"音色克隆音频URL安全校验失败: profile_id={profile.id}, error={e}") return profile # 标记为 processing profile.mark_processing() profile = self.repository.update(profile) try: submit_result = self.cosyvoice_service.submit_clone_task( audio_url=source_audio_url, voice_name=name, language=language, ) # 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("voice_id", "") task_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "") # 如果 CosyVoice 直接返回了 OK 状态，直接标记 ready voice_id = submit_result.get("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) logger.info(f"音色克隆同步完成: profile_id={profile.id}, voice_id={voice_id}") return profile profile.metadata = task_metadata profile = self.repository.update(profile) logger.info( f"音色克隆任务已提交: profile_id={profile.id}, " f"voice_id={submit_result.get('voice_id')}" ) except (CosyVoiceError, CosyVoiceAuthError) as e: # CosyVoice 提交失败，标记为 failed profile.mark_failed(str(e)) profile = self.repository.update(profile) logger.error(f"音色克隆提交失败: profile_id={profile.id}, error={e}") return profile except ValueError as e: profile.mark_failed(str(e)) profile = self.repository.update(profile) logger.error(f"音色克隆参数错误: profile_id={profile.id}, error={e}") return profile else: # 没有音频 URL，保持 pending 状态等待用户上传 logger.info(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: """处理克隆成功结果。 Args: profile_id: Profile ID voice_id: CosyVoice 返回的音色 ID Returns: VoiceCloneProfile: 更新后的 profile Raises: VoiceCloneNotFoundError: profile 不存在 """ profile = self.repository.get(profile_id) if profile is None: raise VoiceCloneNotFoundError(f"Voice clone {profile_id} not found") profile.mark_ready(voice_id) profile = self.repository.update(profile) logger.info(f"音色克隆成功: profile_id={profile_id}, voice_id={voice_id}") return profile def process_clone_failure(self, profile_id: str, error_message: str) -> VoiceCloneProfile: """处理克隆失败结果。 Args: profile_id: Profile ID error_message: 错误信息 Returns: VoiceCloneProfile: 更新后的 profile Raises: VoiceCloneNotFoundError: profile 不存在 """ profile = self.repository.get(profile_id) if profile is None: raise VoiceCloneNotFoundError(f"Voice clone {profile_id} not found") profile.mark_failed(error_message) profile = self.repository.update(profile) logger.error(f"音色克隆失败: profile_id={profile_id}, error={error_message}") return profile def retry_clone(self, clone_id: str, user_id: str) -> VoiceCloneProfile: """重试失败的音色克隆。 1. 调用 RetryVoiceCloneUseCase 重置状态为 pending 2. 标记为 processing 3. 重新提交 CosyVoice 克隆任务 Args: clone_id: Profile ID user_id: 用户 ID Returns: VoiceCloneProfile: 更新后的 profile Raises: VoiceCloneNotFoundError: profile 不存在 VoiceCloneNotRetryableError: 不可重试 VoiceCloneWorkflowError: CosyVoice 提交失败 """ # 1. 重置状态 retry_use_case = RetryVoiceCloneUseCase(self.repository) profile = retry_use_case.execute(clone_id, user_id) # 2. 标记为 processing profile.mark_processing() profile = self.repository.update(profile) # 3. 重新提交 CosyVoice if profile.source_audio_url: # SSRF 防护：重新校验音频 URL 安全性 try: validate_url_safety(profile.source_audio_url, purpose="download") except UrlSecurityError as e: profile.mark_failed(f"音频URL安全校验失败: {e}") profile = self.repository.update(profile) logger.warning(f"音色克隆重试音频URL安全校验失败: profile_id={clone_id}, error={e}") return profile try: submit_result = self.cosyvoice_service.submit_clone_task( audio_url=profile.source_audio_url, voice_name=profile.name, language=profile.language, ) task_metadata = dict(profile.metadata) 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", "") 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) return profile profile.metadata = task_metadata profile = self.repository.update(profile) except (CosyVoiceError, CosyVoiceAuthError, ValueError) as e: profile.mark_failed(str(e)) profile = self.repository.update(profile) logger.error(f"音色克隆重试提交失败: profile_id={clone_id}, error={e}") return profile