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

import json
from datetime import datetime

from pydantic import BaseModel, Field, field_validator, model_validator


class ConfirmGenerationRequest(BaseModel):
    """确认生成请求体 — 基于预览任务创建正式生成任务"""

    output_width: int = Field(default=1080, ge=100, description="输出视频宽度")
    output_height: int = Field(default=1920, ge=100, description="输出视频高度")
    cover_url: str = Field(default="", description="自定义封面图片 URL")
    custom_title: str = Field(default="", description="用户自定义标题文本，非空时同步到任务和编辑计划")


class FinalizeGenerationRequest(BaseModel):
    """Step5 点「完成」请求体：用户选定封面后，正式将视频入成品库。"""

    cover_url: str = Field(
        default="", description="用户选定的封面图片 URL；为空则使用任务默认 cover_url（自动截帧/智能封面）"
    )


class FinalizeGenerationResponse(BaseModel):
    """finalize 响应：返回新创建的成品库视频信息。"""

    video_id: str = Field(description="新创建的成品视频 ID")
    cover_url: str = Field(default="", description="最终绑定的封面 URL")
    file_url: str = Field(default="", description="成品视频 OSS URL")
    status: str = Field(default="success", description="success=新建成功；already_finalized=幂等返回已有记录")
    is_duplicate: bool = Field(default=False, description="是否被判定为与历史成片重复")


class CreateGenerationTaskRequest(BaseModel):
    """创建生成任务请求。

    支持两种模式（至少提供一种）：
    - 项目模式：project_id + asset_library_id（向后兼容）
    - 模板模式：template_id + asset_ids / title_ids / voice_ids
    """

    project_id: str = ""
    asset_library_id: str = ""
    strategy_id: str = ""
    voice_library_id: str = ""
    # ── 多变体独立配音（批量生成）──
    # 长度 1 = 所有变体共用；长度 = count = 每个变体独立配音；空数组 = 回退 voice_library_id
    voice_library_ids: list[str] = Field(
        default_factory=list,
        description="各变体独立配音素材库ID数组：长度1=共用，长度=count=独立。为空时回退 voice_library_id",
    )
    created_by_user_id: str = ""
    # ── 模板模式新增字段 ──
    template_id: str = ""
    asset_ids: list[str] = Field(default_factory=list)
    title_ids: list[str] = Field(default_factory=list)
    # ── 来源剪辑计划 ──
    source_edit_plan_id: str = ""
    # ── variant-plans 轻量选片回传（#1749）：正式生成直接复用，不再重选 ──
    variant_plan_ids: list[str] = Field(
        default_factory=list,
        description="POST /generation/variant-plans 返回的各变体 plan_id（长度须=count）；为空则走服务端选片",
    )
    # ── 标题配置（结构化）──
    title_config: dict | None = Field(
        default=None,
        description="标题样式对象，包含 text/font/font_size/font_color/position/bold/stroke/shadow 等。为空时不影响现有行为。",
    )
    # ── 视频标题 ──
    video_title: str = Field(default="", description="生成视频的标题/名称，为空则使用默认命名")
    # ── 批量生成 ──
    count: int = Field(default=1, ge=1, le=50, description="批量生成数量，默认1，最大50")
    # ── 素材库自动匹配 ──
    asset_select_mode: str = Field(
        default="all",
        description="素材选取模式：all=全部ready视频, smart=智能匹配（按质量/时长评分）",
    )
    asset_select_count: int = Field(default=0, ge=0, le=100, description="选取数量，0表示全部（仅 smart 模式有效）")
    # ── 自动重试 ──
    auto_retry_enabled: bool = Field(
        default=False,
        description="是否开启失败自动重试，默认关闭",
    )
    auto_retry_max: int = Field(
        default=0,
        ge=0,
        le=5,
        description="最大自动重试次数，0表示不自动重试，最大5次",
    )
    # ── 输出分辨率 ──
    resolution: str = Field(
        default="",
        description="输出分辨率，格式为 WIDTHxHEIGHT，如 1280x720、1080x1920。为空使用默认 1280x720",
    )
    # ── 自定义 BGM ──
    bgm_config: dict = Field(
        default_factory=dict,
        description="自定义BGM配置，覆盖模板BGM设置。支持 enabled/source/asset_id/preset_id/audio_url/volume 等字段",
    )
    # ── 预览 / 确认生成 ──
    is_preview: bool = Field(default=False, description="是否为预览任务")
    source_task_id: str = Field(default="", description="来源预览任务 ID（确认生成时传入）")
    output_width: int = Field(default=1280, description="输出视频宽度")
    output_height: int = Field(default=720, description="输出视频高度")
    cover_url: str = Field(default="", description="封面图片 URL")
    # ── 多变体独立封面（批量生成）──
    # 长度 1 = 所有变体共用；长度 = count = 每个变体独立封面；空数组 = 回退 cover_url
    cover_urls: list[str] = Field(
        default_factory=list,
        description="各变体独立封面URL数组：长度1=共用，长度=count=独立。为空时回退 cover_url",
    )
    # ── 多变体独立标题文字（批量生成）──
    # 长度 1 = 所有变体共用；长度 = count = 每个变体独立标题文字；空数组 = 使用 title_config.text
    titles: list[str] = Field(
        default_factory=list,
        description="各变体独立标题文字数组：长度1=共用，长度=count=独立。为空时使用 title_config.text",
    )

    # ── 智能降重开关（#1970）──
    # True（默认）：edge_crop + 片段级微变换（hflip/变速/亮度/对比度/饱和度/BGM偏移）全部生效；
    # False：跳过 edge_crop、不注入微变换，渲染确定性（固定种子）。
    dedup_enabled: bool = Field(default=True, description="智能降重开关，默认开启；关闭后跳过边缘裁切与微变换")

    # ── 剪辑组装模式（#1970 PR3）──
    # random（默认，完全兼容现有随机混剪）/ narrative（叙事剪辑：文案→TTS 配音→标签匹配画面）
    assembly_mode: str = Field(default="random", description="组装模式：random=随机混剪（默认），narrative=叙事剪辑")
    # 叙事模式必填：文案库 scripts.id（后端据此读取 content 合成 TTS）
    script_id: str = Field(default="", description="叙事模式必填：文案库 ID")
    # 叙事模式必填：TTS 音色 ID（preset 为 CosyVoice 音色 id；clone 为克隆档案 id）
    tts_voice_id: str = Field(default="", description="叙事模式必填：TTS 音色 ID（系统音色或克隆档案 ID）")
    tts_voice_source: str = Field(default="preset", description="TTS 音色来源：preset=系统预设（默认），clone=克隆音色")
    # 视频比例：当前前端 9:16/16:9；与 output_width/output_height 并存，传了具体分辨率时以分辨率为准
    video_ratio: str = Field(
        default="", description="视频比例，如 9:16（默认竖屏）/16:9；与显式分辨率冲突时以分辨率为准"
    )

    @model_validator(mode="after")
    def _check_variant_arrays(self) -> "CreateGenerationTaskRequest":
        """变体数组字段长度校验 + #1749 配音严格守卫。

        - cover_urls/titles：空（回退单值）、长度 1（共用）或长度 = count（独立）；
        - voice_library_ids：独立配音长度必须恰好 = count 且逐项非空，禁止静默 fallback
          （长度 1 的"共用"场景请用 voice_library_id 单值字段）；
        - variant_plan_ids：非空时长度必须 = count。
        """
        for name in ("cover_urls", "titles"):
            arr = getattr(self, name)
            if arr and len(arr) != 1 and len(arr) != self.count:
                raise ValueError(f"{name} 长度必须为 1（共用）或 {self.count}（与 count 一致），当前为 {len(arr)}")

        from packages.domain.variant_voice_resolver import VariantVoiceError, resolve_variant_voice_ids

        try:
            resolve_variant_voice_ids(
                count=self.count,
                voice_library_id=self.voice_library_id,
                voice_library_ids=self.voice_library_ids or None,
            )
        except VariantVoiceError as exc:
            raise ValueError(str(exc)) from exc

        if self.variant_plan_ids and len(self.variant_plan_ids) != self.count:
            raise ValueError(f"variant_plan_ids 长度({len(self.variant_plan_ids)})必须与 count({self.count})一致")
        return self

    @model_validator(mode="after")
    def _check_assembly_mode(self) -> "CreateGenerationTaskRequest":
        """#1970 组装模式与叙事模式入参校验。"""
        if self.assembly_mode not in ("random", "narrative"):
            raise ValueError("assembly_mode 仅支持 'random'（默认）或 'narrative'")
        if self.tts_voice_source not in ("preset", "clone"):
            raise ValueError("tts_voice_source 仅支持 'preset' 或 'clone'")
        if self.video_ratio:
            parts = self.video_ratio.split(":")
            if len(parts) != 2 or not all(p.isdigit() and int(p) > 0 for p in parts):
                raise ValueError("video_ratio 格式必须为 '宽:高'，如 9:16 或 16:9")
            if self.video_ratio not in ("9:16", "16:9", "1:1", "3:4", "4:3"):
                raise ValueError("video_ratio 仅支持 9:16 / 16:9 / 1:1 / 3:4 / 4:3")
        if self.assembly_mode == "narrative":
            if not self.script_id.strip():
                raise ValueError("叙事模式（narrative）必须提供 script_id（文案库 ID）")
            if not self.tts_voice_id.strip():
                raise ValueError("叙事模式（narrative）必须提供 tts_voice_id（TTS 音色 ID）")
        return self

    @model_validator(mode="after")
    def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
        has_project = bool(self.project_id.strip())
        has_template = bool(self.template_id.strip())
        if not has_project and not has_template:
            raise ValueError("project_id 或 template_id 至少需要提供一个")
        has_library = bool(self.asset_library_id.strip())
        has_assets = bool(self.asset_ids or self.title_ids)
        if not has_library and not has_assets:
            raise ValueError("asset_library_id 或 asset_ids/title_ids 至少需要提供一个")
        return self


class GenerationTaskResponse(BaseModel):
    id: str
    project_id: str
    asset_library_id: str
    strategy_id: str
    voice_library_id: str
    template_id: str = ""
    asset_ids: list[str] = Field(default_factory=list)
    title_ids: list[str] = Field(default_factory=list)
    voice_ids: list[str] = Field(default_factory=list)
    source_edit_plan_id: str = ""
    asset_select_mode: str = ""
    batch_id: str = ""
    video_title: str = ""
    resolution: str = ""
    bgm_config: dict = Field(default_factory=dict)
    is_preview: bool = False
    source_task_id: str = ""
    output_width: int = 1280
    output_height: int = 720
    cover_url: str = ""
    title_config: dict = Field(default_factory=dict)
    status: str
    progress: float
    result_count: int
    error_message: str
    error_info: dict = Field(default_factory=dict)
    retry_count: int = 0
    auto_retry_enabled: bool = False
    auto_retry_max: int = 0
    logs: list[dict] = Field(default_factory=list)

    @field_validator("logs", mode="before")
    @classmethod
    def _parse_logs(cls, v: object) -> list[dict]:
        """将 JSON 字符串解析为 list[dict]。"""
        if isinstance(v, str):
            try:
                parsed = json.loads(v)
                return parsed if isinstance(parsed, list) else []
            except (json.JSONDecodeError, TypeError):
                return []
        if isinstance(v, list):
            return v
        return []


class BatchGenerationTaskResponse(BaseModel):
    """批量生成任务响应。"""

    items: list[GenerationTaskResponse]
    total: int


class ListGenerationTasksResponse(BaseModel):
    """用户级生成任务列表响应（跨 project）。"""

    items: list[GenerationTaskResponse]


# ── 预览生成（Phase 1） ───────────────────────────────────────────────────────


class CreatePreviewGenerationTaskRequest(BaseModel):
    """创建预览生成任务请求。

    仅支持模板模式：template_id + asset_ids 等素材 ID 列表。
    预览渲染品质与正式生成一致（1080p, CRF 23, medium preset）。
    """

    template_id: str
    asset_ids: list[str] = Field(default_factory=list)
    title_ids: list[str] = Field(default_factory=list)
    voice_library_id: str = Field(
        default="", description="配音素材库ID（用户上传的音频或AI配音），对应配音选择页面选择的配音素材"
    )
    video_title: str = Field(default="", description="生成视频的标题/名称，为空则使用默认命名")
    duration: float = Field(default=0.0, ge=0, description="期望视频时长（秒），0 表示由模板决定")
    video_ratio: str = Field(default="", description="视频比例，如 16:9 / 9:16，为空使用模板默认")
    bgm_config: dict = Field(
        default_factory=dict,
        description="自定义BGM配置，覆盖模板BGM设置。支持 enabled/source/asset_id/preset_id/audio_url/volume 等字段",
    )
    preview_count: int = Field(
        default=1,
        ge=1,
        le=10,
        description="预览视频生成数量，范围 1-10，默认 1",
    )
    source_edit_plan_id: str = Field(
        default="",
        description="关联的编辑计划ID（可选），用于确认生成时复用预览产物",
    )
    title_config: dict = Field(
        default_factory=dict,
        description="标题配置（可选），渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow。N个变体时样式全局共用",
    )
    # ── 多变体独立配置（preview_count > 1）──
    # 长度 1 = 所有变体共用；长度 = preview_count = 每个变体独立；空数组 = 回退单值字段
    titles: list[str] = Field(
        default_factory=list,
        description="各变体独立标题文字数组：长度1=共用，长度=preview_count=独立。为空时使用 title_config.text",
    )
    voice_library_ids: list[str] = Field(
        default_factory=list,
        description="各变体独立配音素材库ID数组：长度1=共用，长度=preview_count=独立。为空时回退 voice_library_id",
    )
    cover_urls: list[str] = Field(
        default_factory=list,
        description="各变体独立封面URL数组：长度1=共用，长度=preview_count=独立（预览阶段通常为空）",
    )

    @model_validator(mode="after")
    def _check_variant_arrays(self) -> "CreatePreviewGenerationTaskRequest":
        """变体数组字段长度校验 + #1749 配音严格守卫。"""
        for name in ("titles", "cover_urls"):
            arr = getattr(self, name)
            if arr and len(arr) != 1 and len(arr) != self.preview_count:
                raise ValueError(
                    f"{name} 长度必须为 1（共用）或 {self.preview_count}（与 preview_count 一致），当前为 {len(arr)}"
                )

        from packages.domain.variant_voice_resolver import VariantVoiceError, resolve_variant_voice_ids

        try:
            resolve_variant_voice_ids(
                count=self.preview_count,
                voice_library_id=self.voice_library_id,
                voice_library_ids=self.voice_library_ids or None,
            )
        except VariantVoiceError as exc:
            raise ValueError(str(exc)) from exc
        return self

    @model_validator(mode="after")
    def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest":
        if not self.template_id.strip():
            raise ValueError("template_id 不能为空")
        return self

    @model_validator(mode="after")
    def _check_asset_ids(self) -> "CreatePreviewGenerationTaskRequest":
        if not self.asset_ids and not self.title_ids:
            raise ValueError("asset_ids/title_ids 至少需要提供一个")
        return self


class PreviewGenerationTaskResponse(BaseModel):
    """单个预览变体任务响应。

    包含任务状态、进度、分辨率、生成结果 URL 等关键字段。
    """

    task_id: str
    status: str
    progress: float
    is_preview: bool = True
    variant_index: int = 0
    resolution: str = ""
    video_url: str = ""
    duration: float = 0.0
    file_size: int = 0
    clip_count: int = 0
    transition_count: int = 0
    material_usage: dict = Field(default_factory=dict)
    error_message: str = ""
    title_text: str = ""
    voice_library_id: str = ""
    created_at: datetime | None = None
    started_at: datetime | None = None
    finished_at: datetime | None = None
    generate_duration: float = 0.0


class BatchPreviewGenerationTaskResponse(BaseModel):
    """批量预览任务响应：preview_count=N 时返回 N 个独立变体任务。

    - items: 变体任务数组，按 variant_index 顺序排列，每个含独立 task_id/状态/预览视频URL
    - total: 变体总数（= preview_count）
    - 前端按 items[i].task_id 分别轮询 GET /preview/{task_id} 获取进度与结果
    """

    items: list[PreviewGenerationTaskResponse]
    total: int