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 Any, Optional

from packages.application.cosyvoice_service import (
    CosyVoiceAuthError,
    CosyVoiceError,
    CosyVoiceService,
)
from packages.application.voice_clone.use_cases import (
    CreateVoiceCloneUseCase,
    RetryVoiceCloneUseCase,
    VoiceCloneNotFoundError,
    VoiceCloneNotRetryableError,
)
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository

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. 标记为 processing
        3. 提交 CosyVoice 克隆任务
        4. 保存 task_id 到 metadata
        5. 返回 profile（Celery task 由调用方触发）

        Args:
            user_id: 用户 ID
            name: 音色名称
            description: 描述
            source_audio_url: 参考音频 URL
            voice_model: 模型名称
            language: 语言
            gender: 性别
            max_retries: 最大重试次数
            metadata: 扩展元数据

        Returns:
            VoiceCloneProfile: 已创建的 profile（状态为 processing）

        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. 标记为 processing
        profile.mark_processing()
        profile = self.repository.update(profile)

        # 3. 提交 CosyVoice 克隆任务
        if source_audio_url:
            try:
                submit_result = self.cosyvoice_service.submit_clone_task(
                    audio_url=source_audio_url,
                    voice_name=name,
                    language=language,
                )

                # 4. 保存 task_id / voice_id 到 metadata
                task_metadata = dict(profile.metadata)
                task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "")
                task_metadata["cosyvoice_request_id"] = submit_result.get(
                    "request_id", ""
                )

                # 如果 CosyVoice 同步返回了 voice_id，直接标记 ready
                voice_id = submit_result.get("voice_id", "")
                if voice_id:
                    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"task_id={submit_result.get('task_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，保持 processing 状态等待用户上传
            logger.info(
                f"音色克隆已创建但无音频URL: profile_id={profile.id}"
            )

        return profile

    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:
            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("task_id", "")
                task_metadata["cosyvoice_request_id"] = submit_result.get(
                    "request_id", ""
                )

                voice_id = submit_result.get("voice_id", "")
                if voice_id:
                    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