xiaoxia-saas/packages/application/voice_clone/workflow.py

"""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