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

"""对口型 API Schema 定义 — #1796 / #1809 / #1822.

支持两种输入模式（二选一）：
1. TTS 直生模式（推荐）：传 voice_id + script_text（+ speed/emotion），
   后端内部先调 CosyVoice 合成音频，再提交 MediaKit 对口型。
2. 直接音频模式：传 video_url + audio_url（音频已由调用方准备好）。
"""

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 必填（+ 可选 speed/emotion）；audio_url 留空。
    - 直接音频：video_url + audio_url 必填。
    """

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

    # 模式 2：直接音频
    audio_url: str = Field("", description="驱动音频 URL（mp3/aac/wav/m4a/flac）；直生模式留空")

    # 模式 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(False, description="音频长于视频时是否循环画面")
    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]
        if not lower.endswith(".mp4"):
            raise ValueError("video_url 仅支持 MP4 格式")

        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