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

"""对口型 API Schema 定义 — #1796 / #1809 / #1822 / #1845（配音前置）.

支持三种输入模式：
1. TTS 直生模式（兼容旧版前端）：传 voice_id + script_text（+ speed/emotion），
   后端 Celery 异步做 TTS 合成 + MediaKit 提交。
2. 直接音频模式：传 video_url + audio_url（音频已由调用方准备好）。
3. 预合成音频模式（#1845 配音前置新主路径）：前端先调 POST /lipsync/tts-preview
   拿到 audio_url + sentence_timings，再在 create_job 时传 audio_url + audio_duration
   + sentence_timings，后端跳过 TTS 和时间戳计算，直接 ffprobe 校验后提交 MediaKit。
"""

from __future__ import annotations

from datetime import datetime
from typing import Optional

from pydantic import BaseModel, Field, model_validator


class LipsyncJobResponse(BaseModel):
    """对口型任务响应."""

    id: str
    user_id: str
    project_id: str
    video_url: str
    audio_url: str
    enable_video_loop: bool
    voice_id: str = ""
    script_text: str = ""
    speed: float = 1.0
    emotion: str = ""
    mediakit_task_id: str
    status: str
    output_video_url: str
    output_duration: float
    error_message: str
    error_code: str
    sentence_timings: Optional[list] = None
    submitted_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True


class CreateLipsyncJobRequest(BaseModel):
    """创建对口型任务请求.

    三种模式（三选一）：
    - TTS 直生（旧版/降级）：voice_id + script_text 必填；audio_url 留空。
    - 直接音频：video_url + audio_url 必填。
    - 预合成音频（#1845 新主路径）：audio_url 必填 + 可选 audio_duration/sentence_timings；
      后端同步 ffprobe 校验时长、写入 timings，直接提交 MediaKit。
    """

    video_url: str = Field(..., description="人物视频 URL（MP4，≤30min，单人真人）")

    # 模式 2/3：直接/预合成音频
    audio_url: str = Field("", description="驱动音频 URL（mp3/aac/wav/m4a/flac）；直生模式留空")
    audio_duration: Optional[float] = Field(None, ge=0, description="预合成音频时长（秒），可选；后端会 ffprobe 校验")
    sentence_timings: Optional[list] = Field(None, description="预合成接口返回的句子时间戳，可选；若传入则直接写入 job")

    # 模式 1：TTS 直生
    voice_id: str = Field("", description="音色 ID（预置音色或克隆音色 profile UUID）")
    script_text: str = Field("", description="要合成的文案（直生模式必填，最长 5000 字符）")
    speed: float = Field(1.0, ge=0.5, le=2.0, description="语速（0.5-2.0），默认 1.0")
    emotion: str = Field("", description="情绪（natural/excited/calm/friendly 或中文 自然/兴奋/沉稳/亲切）")

    enable_video_loop: bool = Field(
        True, description="音频长于视频时是否循环画面（AI数字人默认开启，防止音频长于视频被截断）"
    )
    project_id: str = Field("", description="项目 ID（可选）")

    @model_validator(mode="after")
    def _validate_input_mode(self) -> "CreateLipsyncJobRequest":
        video = (self.video_url or "").strip()
        if not video:
            raise ValueError("video_url 不能为空")
        if not video.startswith(("http://", "https://")):
            raise ValueError("video_url 必须是 HTTP/HTTPS URL")
        lower = video.lower().split("?")[0]
        allowed_video_exts = (".mp4", ".mov", ".m4v", ".webm", ".avi", ".mkv", ".3gp")
        if not any(lower.endswith(ext) for ext in allowed_video_exts):
            raise ValueError("video_url 格式不支持，仅支持: " + ", ".join(allowed_video_exts))

        has_audio = bool((self.audio_url or "").strip())
        has_tts = bool((self.voice_id or "").strip()) and bool((self.script_text or "").strip())

        if not has_audio and not has_tts:
            raise ValueError(
                "必须提供驱动音频：要么传 audio_url（直接/预合成音频模式），"
                "要么同时传 voice_id + script_text（TTS 直生模式）"
            )

        if has_tts and len(self.script_text) > 5000:
            raise ValueError("script_text 最长 5000 字符")

        if has_audio:
            au = self.audio_url.strip()
            if not au.startswith(("http://", "https://")):
                raise ValueError("audio_url 必须是 HTTP/HTTPS URL")
            au_lower = au.lower().split("?")[0]
            allowed = (".mp3", ".aac", ".wav", ".m4a", ".flac")
            if not any(au_lower.endswith(ext) for ext in allowed):
                raise ValueError(f"audio_url 格式不支持，仅支持: {', '.join(allowed)}")
            self.audio_url = au

        return self


# ── #1845 TTS 预合成接口 ────────────────────────────────────────────────


class AiAvatarTtsPreviewRequest(BaseModel):
    """步骤1「生成配音」预合成请求（同步 HTTP，~2-3s）."""

    voice_id: str = Field(..., min_length=1, max_length=128, description="音色 ID")
    script_text: str = Field(..., min_length=1, max_length=5000, description="要合成的文案")
    speed: float = Field(1.0, ge=0.5, le=2.0, description="语速（0.5-2.0），默认 1.0")
    emotion: str = Field("natural", max_length=32, description="情绪")


class AiAvatarTtsPreviewResponse(BaseModel):
    """TTS 预合成响应（临时 URL，24h 内有效，足够当前会话使用）."""

    audio_url: str = Field(..., description="CosyVoice 临时音频 URL")
    duration: float = Field(..., ge=0, description="音频总时长（秒），ffprobe 测得")
    sentence_timings: list[dict] = Field(..., description="句子级精确时间戳")