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 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 = ""
    created_by_user_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 = ""
    # ── 视频标题 ──
    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视频, random=随机选取, smart=智能匹配（按质量/时长评分）",
    )
    asset_select_count: int = Field(
        default=0, ge=0, le=100, description="选取数量，0表示全部（仅 random/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 等字段",
    )

    @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 or self.voice_ids)
        if not has_library and not has_assets:
            raise ValueError("asset_library_id 或 asset_ids/title_ids/voice_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)
    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 列表。
    预览为完整时长低清版（480p + 低码率）。
    """

    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)
    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 等字段",
    )

    @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 and not self.voice_ids:
            raise ValueError("asset_ids/title_ids/voice_ids 至少需要提供一个")
        return self


class PreviewGenerationTaskResponse(BaseModel):
    """预览生成任务响应。

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

    task_id: str
    status: str
    progress: float
    is_preview: bool = True
    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 = ""
    created_at: datetime | None = None
    started_at: datetime | None = None
    finished_at: datetime | None = None
    generate_duration: float = 0.0