xiaoxia-saas/apps/worker/worker_app/tasks/voice_clone.py

"""Voice clone tasks - process voice clone requests via CosyVoice API."""

import logging

from celery import Task
from celery.exceptions import Retry
from video_processing.oss_helpers import get_signed_download_url
from worker_app.celery_app import celery_app
from worker_app.db import SessionLocal

from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
    SQLAlchemyVoiceCloneProfileRepository,
)
from packages.application.cosyvoice_service import (
    CosyVoiceError,
    CosyVoiceService,
    CosyVoiceTimeoutError,
)
from packages.application.voice_clone.workflow import VoiceCloneWorkflowService

logger = logging.getLogger(__name__)


@celery_app.task(bind=True, max_retries=2, name="worker.process_voice_clone")
def process_voice_clone(self: Task, profile_id: str) -> dict:
    """处理音色克隆任务。

    通过 VoiceCloneWorkflowService.poll_and_process_clone() 轮询 CosyVoice
    克隆任务状态，更新 VoiceCloneProfile。

    重试策略（Celery 5.x bind=True 模式）：
    - max_retries=2：最多重试 2 次（共执行 3 次），超过后抛出 MaxRetriesExceededError
    - CosyVoiceTimeoutError：网络超时属于临时性故障，使用 countdown=30 延迟 30 秒后重试
    - CosyVoiceError：API 业务错误（如任务失败），属于永久性故障，不重试直接标记 failed
    - Exception：未知错误，不重试直接标记 failed，避免无限重试掩盖 bug
    - Retry 异常：Celery 内部重试信号，必须向上传播不能被捕获

    Args:
        profile_id: VoiceCloneProfile ID

    Returns:
        dict: {"ok": True, "profile_id": str, "voice_id": str}
    """
    # P2-2 修复：session 初始化为 None，避免 SessionLocal() 抛异常时
    # finally 块中 session.close() 触发 UnboundLocalError
    session = None
    try:
        session = SessionLocal()
        repo = SQLAlchemyVoiceCloneProfileRepository(session)
        workflow = VoiceCloneWorkflowService(
            repository=repo,
            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)
        session.commit()

        logger.info(f"Voice clone completed: profile_id={profile_id}, " f"voice_id={updated_profile.voice_id}")
        return {
            "ok": True,
            "profile_id": profile_id,
            "voice_id": updated_profile.voice_id,
        }

    except Retry:
        # Celery Retry 异常必须向上传播，不能被后续 except 捕获
        raise

    except CosyVoiceTimeoutError as e:
        logger.error(f"Voice clone timeout for {profile_id}: {e}")
        if session is not None:
            session.rollback()
        # 超时属于临时性故障，延迟 30 秒后重试
        raise self.retry(exc=e, countdown=30) from e

    except CosyVoiceError as e:
        logger.error(f"Voice clone failed for {profile_id}: {e}")
        if session is not None:
            session.rollback()
        # API 业务错误属于永久性故障，不重试，标记 profile 为 failed
        try:
            if session is not None:
                profile = repo.get(profile_id)
                if profile is not None:
                    profile.mark_failed(str(e))
                    repo.update(profile)
                    session.commit()
        except Exception as inner_e:
            logger.error(f"Failed to mark profile as failed: {inner_e}")
            if session is not None:
                session.rollback()
        return {"ok": False, "profile_id": profile_id, "error": str(e)}

    except Exception as e:
        logger.error(f"Voice clone unexpected error for {profile_id}: {e}")
        if session is not None:
            session.rollback()
        # 未知错误不重试，标记 profile 为 failed，避免无限重试掩盖 bug
        try:
            if session is not None:
                profile = repo.get(profile_id)
                if profile is not None:
                    profile.mark_failed(str(e))
                    repo.update(profile)
                    session.commit()
        except Exception as inner_e:
            logger.error(f"Failed to mark profile as failed: {inner_e}")
            if session is not None:
                session.rollback()
        return {"ok": False, "profile_id": profile_id, "error": str(e)}

    finally:
        if session is not None:
            session.close()