xiaoxia-saas/apps/api/app/api/routes/lipsync.py

"""对口型 API 路由 — #1796 MediaKit 对口型, #1809 参数调整.

接口：
  POST   /api/v1/lipsync/jobs          提交对口型任务
  GET    /api/v1/lipsync/jobs          任务列表
  GET    /api/v1/lipsync/jobs/{id}     任务详情
  POST   /api/v1/lipsync/jobs/{id}/refresh  刷新任务状态
  POST   /api/v1/lipsync/jobs/{id}/cancel   取消任务
"""

from __future__ import annotations

import logging

from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import (
    get_cosyvoice_service,
    get_db_session,
    get_voice_clone_profile_repository,
)
from app.schemas.lipsync import CreateLipsyncJobRequest, LipsyncJobResponse
from app.services.lipsync_service import LipsyncService
from app.services.mediakit_client import MediaKitError
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
from sqlalchemy.orm import Session

from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService

logger = logging.getLogger(__name__)

router = APIRouter()


def _get_service(
    db: Session = Depends(get_db_session),
    voice_clone_repo=Depends(get_voice_clone_profile_repository),
    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
) -> LipsyncService:
    # voice_clone_repo 用于克隆音色 profile 解析；cosyvoice_service 用于 TTS 直生
    # （TTS 合成、音色解析、错误码归一化都在 LipsyncService 内部完成）
    return LipsyncService(
        db,
        cosyvoice_service=cosyvoice_service,
        voice_clone_repo=voice_clone_repo,
    )


# ── POST /jobs — 提交对口型任务 ───────────────────────────────────────────


@router.post("/jobs", response_model=LipsyncJobResponse, status_code=201)
def create_lipsync_job(
    body: CreateLipsyncJobRequest,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: LipsyncService = Depends(_get_service),
):
    """提交对口型任务.

    #1809/#1822: 前端传 {video_url, voice_id, script_text, speed?, emotion?}，
    后端内部解析音色、调 TTS 合成音频、转存 OSS，再提交 MediaKit；
    也支持直接传 {video_url, audio_url}。
    """
    try:
        job = svc.create_job(
            user_id=current_user.user.id,
            video_url=body.video_url,
            audio_url=body.audio_url,
            voice_id=body.voice_id,
            script_text=body.script_text,
            speed=body.speed,
            emotion=body.emotion,
            enable_video_loop=body.enable_video_loop,
            project_id=body.project_id,
        )
    except ValueError as exc:
        # 参数无效（如 voice_id 格式不对、文本过长等）
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except CosyVoiceError as exc:
        # TTS 合成基础设施失败（API/网络/认证）
        raise HTTPException(
            status_code=502,
            detail={"code": "TTSSynthesisFailed", "message": str(exc)},
        ) from exc
    except MediaKitError as exc:
        # TTS 合成失败 / 音色无权访问 → 400/403；MediaKit 提交失败 → 502
        status_code = 502
        if exc.code in ("VoiceForbidden",):
            status_code = 403
        elif exc.code in ("InvalidInput", "TTSInvalidParam", "VoiceNotReady"):
            status_code = 400
        elif exc.code == "TTSSynthesisFailed":
            status_code = 502
        raise HTTPException(
            status_code=status_code,
            detail={
                "code": exc.code,
                "message": str(exc),
                "request_id": getattr(exc, "request_id", ""),
            },
        ) from exc
    except Exception as exc:
        # 兜底：任何未预期的错误返回 400 而非 500
        logger.error("创建对口型任务异常: %s", exc, exc_info=True)
        raise HTTPException(
            status_code=400,
            detail=f"创建对口型任务失败: {exc}",
        ) from exc

    return job


# ── GET /jobs — 任务列表 ─────────────────────────────────────────────────


@router.get("/jobs", response_model=dict)
def list_lipsync_jobs(
    project_id: str = Query("", description="项目 ID 过滤"),
    status: str = Query("", description="状态过滤"),
    offset: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: LipsyncService = Depends(_get_service),
):
    """获取对口型任务列表."""
    items, total = svc.list_jobs(
        user_id=current_user.user.id,
        project_id=project_id,
        status=status,
        offset=offset,
        limit=limit,
    )
    return {
        "items": [LipsyncJobResponse.model_validate(j) for j in items],
        "total": total,
        "offset": offset,
        "limit": limit,
    }


# ── GET /jobs/{job_id} — 任务详情 ────────────────────────────────────────


@router.get("/jobs/{job_id}", response_model=LipsyncJobResponse)
def get_lipsync_job(
    job_id: str,
    background: BackgroundTasks,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: LipsyncService = Depends(_get_service),
):
    """获取对口型任务详情.

    非终态任务：先返回 DB 缓存，挂后台刷新（下次轮询拿到新状态），
    避免 MediaKit 慢响应阻塞前端轮询。
    """
    job = svc.get_job(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="任务不存在")

    if job.status not in ("completed", "failed"):
        background.add_task(svc.refresh_job_status, job_id, current_user.user.id)

    return job


# ── POST /jobs/{job_id}/refresh — 刷新状态 ───────────────────────────────


@router.post("/jobs/{job_id}/refresh", response_model=LipsyncJobResponse)
def refresh_lipsync_job(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: LipsyncService = Depends(_get_service),
):
    """从 MediaKit 拉取最新状态并更新."""
    job = svc.refresh_job_status(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="任务不存在")
    return job


# ── POST /jobs/{job_id}/cancel — 取消任务 ────────────────────────────────


@router.post("/jobs/{job_id}/cancel", response_model=LipsyncJobResponse)
def cancel_lipsync_job(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: LipsyncService = Depends(_get_service),
):
    """取消对口型任务（仅 pending/submitted 状态可取消）."""
    job = svc.cancel_job(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="任务不存在")
    if job.status != "cancelled":
        raise HTTPException(
            status_code=400,
            detail=f"任务状态 {job.status} 不可取消，仅 pending/submitted 可取消",
        )
    return job