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

"""AI数字人渲染合成管线 API Schema — #1798."""

from __future__ import annotations

from datetime import datetime
from typing import Any, Optional

from pydantic import BaseModel, Field, field_validator


class BRollSegment(BaseModel):
    """B-roll 片段配置."""

    script_segment_index: int = Field(..., ge=0, description="对应文案片段索引")
    asset_url: str = Field(..., description="B-roll 素材 URL")
    mode: str = Field(..., description="插入模式: fullscreen 或 pip")
    start_time: float = Field(..., ge=0.0, description="在对口型视频中的起始时间(秒)")
    end_time: float = Field(..., ge=0.0, description="在对口型视频中的结束时间(秒)")
    pip_position: Optional[str] = Field("bottom_right", description="pip 模式位置")
    pip_scale: Optional[float] = Field(0.3, ge=0.05, le=1.0, description="pip 模式缩放比例")

    @field_validator("mode")
    @classmethod
    def validate_mode(cls, v: str) -> str:
        v = v.strip().lower()
        if v not in ("fullscreen", "pip"):
            raise ValueError("mode 必须为 fullscreen 或 pip")
        return v

    @field_validator("asset_url")
    @classmethod
    def validate_asset_url(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("asset_url 不能为空")
        if not v.startswith(("http://", "https://")):
            raise ValueError("asset_url 必须是 HTTP/HTTPS URL")
        return v

    @field_validator("end_time")
    @classmethod
    def validate_end_time(cls, v: float, info: Any) -> float:
        start = info.data.get("start_time", 0.0)
        if v <= start:
            raise ValueError("end_time 必须大于 start_time")
        return v


class CreateAiAvatarRenderRequest(BaseModel):
    """创建渲染任务请求."""

    lipsync_job_id: str = Field(..., description="对口型任务 ID")
    script_id: str = Field("", description="文案 ID（选自文案库时传；手动输入文案直生场景可留空）")
    b_roll_segments: list[BRollSegment] = Field(default_factory=list, description="B-roll 片段列表")
    title_config: dict[str, Any] = Field(
        default_factory=dict, description="标题配置（可含 title_image_dataurl：前端 Canvas 渲染的标题 PNG dataURL）"
    )
    cover_config: dict[str, Any] = Field(default_factory=dict, description="封面配置")
    project_id: str = Field("", description="项目 ID")

    @field_validator("lipsync_job_id")
    @classmethod
    def validate_lipsync_job_id(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("lipsync_job_id 不能为空")
        return v

    @field_validator("script_id")
    @classmethod
    def validate_script_id(cls, v: str) -> str:
        return (v or "").strip()


class AiAvatarRenderJobResponse(BaseModel):
    """渲染任务响应."""

    id: str
    user_id: str
    project_id: str
    lipsync_job_id: str
    script_id: str = ""
    b_roll_segments: list[dict[str, Any]]
    title_config: dict[str, Any]
    cover_config: dict[str, Any]
    status: str
    progress: int
    output_video_url: str
    output_cover_url: str
    output_duration: float
    error_message: str
    submitted_at: Optional[datetime] = None
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True


class AiAvatarRenderProgressResponse(BaseModel):
    """渲染进度响应."""

    status: str
    progress: int
    output_video_url: str
    output_cover_url: str
    output_duration: float
    error_message: str


class SmartCoverResponse(BaseModel):
    """智能封面响应（封面从最终成片抽帧，不再叠加标题）."""

    cover_url: str = Field("", description="封面图公网 URL（OSS，非临时）；失败为空")
    status: str = Field("completed", description="completed / fallback_failed")
    message: str = Field("", description="失败原因（如有）")


class FinalizeRenderResponse(BaseModel):
    """封面选好后点「完成」，正式入库成片库的响应."""

    video_id: str = Field(..., description="成片库视频ID")
    cover_url: str = Field("", description="封面URL")
    status: str = Field("success", description="success/already_finalized")