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

"""AI数字人渲染合成 API 路由 — #1798.

接口：
  POST   /api/v1/ai-avatar/render              提交渲染任务
  GET    /api/v1/ai-avatar/render/jobs          任务列表
  GET    /api/v1/ai-avatar/render/{job_id}      任务详情
  POST   /api/v1/ai-avatar/render/{job_id}/cancel  取消任务
  POST   /api/v1/ai-avatar/render/{job_id}/retry   重试失败任务
"""

from __future__ import annotations

import logging
from datetime import datetime, timezone

from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session
from app.schemas.ai_avatar_render import (
    AiAvatarRenderJobResponse,
    CreateAiAvatarRenderRequest,
    FinalizeRenderResponse,
    SmartCoverResponse,
)
from app.services.ai_avatar_cover_service import generate_smart_cover
from app.services.ai_avatar_render_service import (
    AiAvatarRenderError,
    AiAvatarRenderService,
)
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

router = APIRouter()


def _get_service(db: Session = Depends(get_db_session)) -> AiAvatarRenderService:
    return AiAvatarRenderService(db)


# ── POST / — 提交渲染任务 ────────────────────────────────────────────────


@router.post("", response_model=AiAvatarRenderJobResponse, status_code=201)
def create_render_job(
    body: CreateAiAvatarRenderRequest,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: AiAvatarRenderService = Depends(_get_service),
):
    """提交 AI 数字人渲染任务.

    将对口型视频 + B-roll 素材 + 标题叠加 + 封面提取合成最终输出视频。
    """
    try:
        job = svc.create_render_job(
            user_id=current_user.user.id,
            lipsync_job_id=body.lipsync_job_id,
            script_id=body.script_id,
            b_roll_segments=[s.model_dump() for s in body.b_roll_segments],
            title_config=body.title_config,
            cover_config=body.cover_config,
            project_id=body.project_id,
        )
    except AiAvatarRenderError as exc:
        status_map = {
            "LipsyncJobNotFound": 404,
            "LipsyncJobNotCompleted": 400,
            "LipsyncJobNoOutput": 400,
            "ScriptNotFound": 404,
        }
        raise HTTPException(
            status_code=status_map.get(exc.code, 400),
            detail={"code": exc.code, "message": str(exc)},
        ) from exc

    # 异步触发渲染
    try:
        from app.tasks.ai_avatar_render import execute_ai_avatar_render

        execute_ai_avatar_render.delay(job.id)
    except Exception as exc:
        logger.exception("Celery 任务投递失败（创建）: job_id=%s err=%s", job.id, exc)
        job.status = "failed"
        job.error_message = f"任务提交失败：{exc}"
        job.updated_at = datetime.now(timezone.utc)
        svc.db.commit()
        svc.db.refresh(job)
        return AiAvatarRenderJobResponse.model_validate(job)

    return AiAvatarRenderJobResponse.model_validate(job)


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


@router.get("/jobs", response_model=dict)
def list_render_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: AiAvatarRenderService = Depends(_get_service),
):
    """获取 AI 数字人渲染任务列表."""
    items, total = svc.list_render_jobs(
        user_id=current_user.user.id,
        project_id=project_id,
        status=status,
        offset=offset,
        limit=limit,
    )
    return {
        "items": [AiAvatarRenderJobResponse.model_validate(j) for j in items],
        "total": total,
        "offset": offset,
        "limit": limit,
    }


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


@router.get("/{job_id}", response_model=AiAvatarRenderJobResponse)
def get_render_job(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: AiAvatarRenderService = Depends(_get_service),
):
    """获取渲染任务详情."""
    job = svc.get_render_job(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="渲染任务不存在")
    return job


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


@router.post("/{job_id}/cancel", response_model=AiAvatarRenderJobResponse)
def cancel_render_job(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: AiAvatarRenderService = Depends(_get_service),
):
    """取消渲染任务（仅 pending 状态可取消）."""
    job = svc.cancel_render_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 可取消",
        )
    return job


# ── POST /{job_id}/retry — 重试失败任务 ──────────────────────────────────


@router.post("/{job_id}/retry", response_model=AiAvatarRenderJobResponse)
def retry_render_job(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    svc: AiAvatarRenderService = Depends(_get_service),
):
    """重试失败的渲染任务."""
    job = svc.retry_render_job(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="渲染任务不存在")
    if job.status != "pending":
        raise HTTPException(
            status_code=400,
            detail=f"仅 failed 状态的任务可重试，当前状态: {job.status}",
        )

    # 重新触发渲染
    try:
        from app.tasks.ai_avatar_render import execute_ai_avatar_render

        execute_ai_avatar_render.delay(job.id)
    except Exception as exc:
        logger.exception("Celery 任务投递失败（重试）: job_id=%s err=%s", job.id, exc)
        job.status = "failed"
        job.error_message = f"任务提交失败：{exc}"
        job.updated_at = datetime.now(timezone.utc)
        svc.db.commit()
        svc.db.refresh(job)
        return AiAvatarRenderJobResponse.model_validate(job)

    return AiAvatarRenderJobResponse.model_validate(job)


# ── POST /{job_id}/smart-cover — 从最终成片智能抽封面（步骤②）────────


@router.post("/{job_id}/smart-cover", response_model=SmartCoverResponse)
def generate_render_smart_cover(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
):
    """从最终渲染成片智能抽帧生成封面（MediaKit 抽帧 + 评分选最佳帧 + 转存 OSS）.

    - 必须等渲染任务 completed 后才可调用（否则返回 400）
    - 生成成功后自动更新 render_job 的 cover_config 与 output_cover_url
    """
    from app.services.ai_avatar_render_service import AiAvatarRenderService

    svc = AiAvatarRenderService(db)
    job = svc.get_render_job(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="渲染任务不存在")
    if job.status != "completed":
        raise HTTPException(status_code=400, detail="请先完成视频生成")
    video_url = (job.output_video_url or "").strip()
    if not video_url:
        raise HTTPException(status_code=400, detail="渲染成片视频 URL 为空")

    try:
        # 从最终成片抽帧，帧本身已含标题/B-roll，直接转存 OSS
        cover_url = generate_smart_cover(video_url, job_id=job_id, max_frames=5)
    except Exception as exc:
        logger.error(
            "渲染成片智能封面生成异常: user=%s render_id=%s video_url=%s err=%s",
            current_user.user.id,
            job_id,
            video_url[:80],
            exc,
            exc_info=True,
        )
        cover_url = ""

    if not cover_url:
        return SmartCoverResponse(
            cover_url="",
            status="fallback_failed",
            message="智能抽帧失败（MediaKit 不可用或抽帧异常），请稍后重试",
        )

    # 更新 render_job 的封面字段（异步写入 DB；失败不影响返回）
    try:
        job.cover_config = {
            **(job.cover_config if isinstance(job.cover_config, dict) else {}),
            "mode": "auto_frame",
            "url": cover_url,
        }
        job.output_cover_url = cover_url
        job.updated_at = datetime.now(timezone.utc)
        db.commit()
    except Exception as exc:
        logger.warning("更新 render_job 封面字段失败（不影响返回）: job_id=%s err=%s", job_id, exc)

    logger.info(
        "渲染成片智能封面生成成功: user=%s render_id=%s cover_url=%s",
        current_user.user.id,
        job_id,
        cover_url[:120],
    )
    return SmartCoverResponse(cover_url=cover_url, status="completed")

# ── POST /{job_id}/finalize — 封面选定后正式入库成片库 ────────────────────


@router.post("/{job_id}/finalize", response_model=FinalizeRenderResponse)
def finalize_render_job(
    job_id: str,
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
):
    """用户完成封面选择后，将视频正式保存到成片库.

    - 必须等渲染任务 completed 后才可调用
    - 如果已通过 smart-cover/custom-cover 设置了封面，会自动带上
    - 返回成片库视频ID
    - 幂等：已 finalize 的任务重复调用会返回 existing 记录
    """
    from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService

    svc = AiAvatarRenderService(db)
    job = svc.get_render_job(job_id, current_user.user.id)
    if job is None:
        raise HTTPException(status_code=404, detail="渲染任务不存在")
    if job.status != "completed":
        raise HTTPException(status_code=400, detail="请先完成视频生成")

    # 幂等检查（通过 generation_task_id=job_id 识别，finalize_job 内部也做了一次，这里提前返回简化）
    from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

    existing = (
        db.query(GeneratedVideoModel)
        .filter(
            GeneratedVideoModel.user_id == current_user.user.id,
            GeneratedVideoModel.generation_task_id == job_id,
        )
        .first()
    )
    if existing is not None:
        return FinalizeRenderResponse(
            video_id=existing.id,
            cover_url=existing.thumbnail_url or "",
            status="already_finalized",
        )

    try:
        video = svc.finalize_job(job_id, current_user.user.id)
        return FinalizeRenderResponse(
            video_id=video.id,
            cover_url=video.thumbnail_url or job.output_cover_url or "",
            status="success",
        )
    except AiAvatarRenderError as exc:
        status_map = {
            "RenderJobNotFound": 404,
            "RenderNotCompleted": 400,
            "OutputVideoMissing": 400,
        }
        raise HTTPException(
            status_code=status_map.get(exc.code, 400),
            detail=str(exc),
        ) from exc
    except Exception as exc:
        logger.error("渲染任务finalize失败: job_id=%s err=%s", job_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail=f"保存到成片库失败: {str(exc)}") from exc