from typing import Annotated, Literal from pydantic import BaseModel, ConfigDict, Field from app.schemas.common import NaiveDatetimeOptional class GenerationAIReference(BaseModel): """AI生成任务参考素材。""" model_config = ConfigDict( json_schema_extra={ "example": { "url": "https://example.com/reference.png", "type": "image", "name": "参考图.png", } } ) url: str = Field( ..., description="参考素材地址，可以是图片地址或视频地址", examples=["https://example.com/reference.png"], ) type: str = Field( ..., description="参考素材类型：image=图片，video=视频", examples=["image"], ) name: str | None = Field( None, description="参考素材名称，前端展示用，可为空", examples=["参考图.png"], ) class GenerationAITaskCreate(BaseModel): """创建AI图片/视频生成任务请求体。""" model_config = ConfigDict( json_schema_extra={ "examples": [ { "project_id": None, "gen_type": "image", "original_prompt": "生成一张赛博朋克风格的城市夜景", "engine_id": None, "media_references": [ { "url": "https://example.com/reference.png", "type": "image", "name": "参考图.png", } ], "idempotency_key": "frontend-submit-uuid-001", "image_size": "2K", "image_proportion": "1:1", "image_px": "2048x2048", "duration": None, "aspect_ratio": None, "resolution": None, }, { "project_id": None, "gen_type": "video", "original_prompt": "生成一段海边日落的电影感视频", "engine_id": None, "media_references": None, "idempotency_key": "frontend-submit-uuid-002", "image_size": None, "image_proportion": None, "image_px": None, "duration": 4, "aspect_ratio": "16:9", "resolution": "480p", }, ] } ) # 新 chat 生成任务不绑定 project_id。为了兼容旧前端误传，保留可选字段但后端不使用。 project_id: str | None = Field( None, description="兼容旧前端字段。当前 /generation-ai 任务不绑定项目，后端不使用该字段，可传 null 或不传", examples=[None], ) gen_type: str = Field( ..., description="生成类型：image=图片生成，video=视频生成", examples=["image"], ) original_prompt: str = Field( ..., min_length=1, max_length=5000, description="用户原始提示词，不能为空，最长5000字符", examples=["生成一张赛博朋克风格的城市夜景"], ) engine_id: str | None = Field( None, description="图片/视频引擎ID；为空则使用当前激活且优先级最高的引擎", examples=[None], ) media_references: list[GenerationAIReference] | None = Field( None, description="参考素材列表。可以传图片/视频参考素材；为空表示不使用参考素材", ) idempotency_key: str | None = Field( None, max_length=64, description=( "幂等键，用于防止前端重复提交、网络重试导致重复创建任务和重复扣费。" "同一用户、同一 idempotency_key、同一 generation_mode 下重复请求会返回已有任务。" "建议前端每次点击生成时生成 UUID；同一次请求失败重试时复用同一个 UUID。" ), examples=["frontend-submit-uuid-001"], ) # image params image_size: str | None = Field( None, description="图片分辨率档位，例如：1K、2K。仅图片生成或视频首帧参数需要使用；为空则使用引擎默认值", examples=["2K"], ) image_proportion: str | None = Field( None, description="图片比例，例如：1:1、16:9、9:16。仅图片生成或视频首帧参数需要使用；为空则使用默认值", examples=["1:1"], ) image_px: str | None = Field( None, description="图片像素尺寸，例如：2048x2048。为空时后端根据 image_size 和 image_proportion 自动匹配", examples=["2048x2048"], ) # video params duration: int | None = Field( None, description="视频时长，单位秒。仅视频生成使用；为空则使用默认时长", examples=[4], ) aspect_ratio: str | None = Field( None, description="视频比例，例如：16:9、9:16、1:1。仅视频生成使用；为空则使用默认比例", examples=["16:9"], ) resolution: str | None = Field( None, description="视频分辨率，例如：480p、720p、1080p。仅视频生成使用；为空则使用默认分辨率", examples=["480p"], ) class GenerationAIImageEngineOptionOut(BaseModel): """AI图片生成可用引擎响应项。""" id: str = Field(..., description="图片引擎ID，创建图片任务时传入 engine_id") name: str = Field(..., description="图片引擎名称，前端展示用") provider: str = Field(..., description="服务商标识，例如 ark") model_name: str | None = Field(None, description="服务商模型名称") supported_models: list[str] = Field(default_factory=list, description="该图片引擎支持的模型名称列表") supported_sizes: dict = Field( default_factory=dict, description="支持的图片尺寸映射。第一层为分辨率档位，例如 2K/4K；第二层为画布比例，例如 1:1/16:9；值为像素尺寸", ) default_size: str | None = Field(None, description="默认图片分辨率档位，例如 2K") priority: int = Field(0, description="引擎优先级，数值越大越优先") class GenerationAIVideoEngineOptionOut(BaseModel): """AI视频生成可用引擎响应项。""" id: str = Field(..., description="视频引擎ID，创建视频任务时传入 engine_id") name: str = Field(..., description="视频引擎名称，前端展示用") provider: str = Field(..., description="服务商标识，例如 ark") model_name: str | None = Field(None, description="服务商模型名称") supported_ratios: list[str] = Field(default_factory=list, description="支持的视频画面比例，例如 16:9、9:16、1:1") supported_resolutions: list[str] = Field(default_factory=list, description="支持的视频分辨率，例如 480p、720p、1080p") supported_durations: list[int] = Field(default_factory=list, description="支持的视频时长列表，单位秒") max_duration: int | None = Field(None, description="最大视频时长，单位秒") priority: int = Field(0, description="引擎优先级，数值越大越优先") class GenerationAIEngineGroupOut(BaseModel): """AI图片/视频引擎分组。""" image: list[GenerationAIImageEngineOptionOut] = Field( default_factory=list, description="当前启用的图片生成引擎列表", ) video: list[GenerationAIVideoEngineOptionOut] = Field( default_factory=list, description="当前启用的视频生成引擎列表", ) class GenerationAIEngineOptionsOut(BaseModel): """AI图片/视频生成引擎列表响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "engine": { "image": [ { "id": "image_engine_xxx", "name": "豆包文生图", "provider": "ark", "model_name": "doubao-seedream-5-0-260128", "supported_models": ["doubao-seedream-5-0-260128"], "supported_sizes": { "2K": { "1:1": "2048x2048", "16:9": "2560x1440", } }, "default_size": "2K", "priority": 10, } ], "video": [ { "id": "video_engine_xxx", "name": "Seedance 2.0", "provider": "ark", "model_name": "doubao-seedance-2-0-260128", "supported_ratios": ["16:9", "4:3", "1:1", "3:4", "9:16", "21:9"], "supported_resolutions": ["480p", "720p", "1080p"], "supported_durations": [4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15], "max_duration": 15, "priority": 10, } ], } } } ) engine: GenerationAIEngineGroupOut = Field(..., description="图片/视频可用引擎分组") class GenerationAITaskOut(BaseModel): """AI生成任务详情响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "id": "0019e0a44895b6d837d", "source_type": "chat_task", "project_id": None, "gen_type": "image", "generation_mode": "chatapi_async", "pipeline_stage": "done", "status": "completed", "original_prompt": "生成一张赛博朋克风格的城市夜景", "optimized_prompt": None, "duration": None, "aspect_ratio": None, "resolution": None, "image_size": "2K", "image_proportion": "1:1", "image_px": "2048x2048", "media_references": None, "provider_task_id": "provider_task_xxx", "seedance_task_id": "provider_task_xxx", "remote_result_url": None, "image_url": "https://example.com/result.png", "video_url": None, "video_cover_url": None, "engine_id": "engine_xxx", "engine_snapshot": { "engine_type": "image", "id": "engine_xxx", "name": "图片生成引擎", "provider": "provider_name", "model_name": "model_name", "supported_models": [], "default_size": "2K", "selected_size": "2K", "selected_proportion": "1:1", "selected_px": "2048x2048", }, "credits_cost": 10.0, "text_credits_cost": 0.0, "text_tokens_used": 0, "image_tokens_used": 0, "video_tokens_used": 0, "retry_count": 0, "poll_count": 3, "error_message": None, "created_at": "2026-05-27T10:12:00", "generated_at": "2026-05-27T10:15:30", } } ) id: str = Field(..., description="生成任务ID") source_type: Literal["chat_task"] = Field( "chat_task", description="历史记录来源。ChatGenerationTask 新任务历史固定为 chat_task", ) project_id: str | None = Field( None, description="项目ID。当前 /generation-ai 任务不绑定项目，通常为 null", ) gen_type: str = Field(..., description="生成类型：image=图片，video=视频") generation_mode: str | None = Field( None, description="生成模式。当前异步Chat生成任务一般为 chatapi_async", ) pipeline_stage: str | None = Field( None, description=( "任务流水线阶段，例如：queued=已入队，creating_provider_task=创建第三方任务中，" "waiting_remote=等待第三方生成，result_ready=远程结果已就绪，downloading=下载中，done=完成" ), ) status: str = Field( ..., description="任务状态，例如：generating=生成中，completed=已完成，failed=失败", ) original_prompt: str = Field(..., description="用户原始提示词") optimized_prompt: str | None = Field(None, description="优化后的提示词，可能为空") duration: int | None = Field(None, description="视频时长，单位秒。图片任务通常为空") aspect_ratio: str | None = Field(None, description="视频比例，例如 16:9。图片任务通常为空") resolution: str | None = Field(None, description="视频分辨率，例如 480p。图片任务通常为空") image_size: str | None = Field(None, description="图片分辨率档位，例如 2K") image_proportion: str | None = Field(None, description="图片比例，例如 1:1") image_px: str | None = Field(None, description="图片像素尺寸，例如 2048x2048") media_references: list[dict] | None = Field( None, description="参考素材列表。由创建任务时传入的 media_references 转换而来", ) provider_task_id: str | None = Field(None, description="第三方服务商任务ID") seedance_task_id: str | None = Field( None, description="兼容旧前端/旧服务命名的第三方任务ID字段", ) remote_result_url: str | None = Field( None, description="第三方远程结果地址。当前接口可能隐藏或为空，最终展示优先使用 image_url/video_url", ) image_url: str | None = Field(None, description="最终图片地址。图片任务完成后通常有值") video_url: str | None = Field(None, description="最终视频地址。视频任务完成后通常有值") video_cover_url: str | None = Field(None, description="视频封面图片地址。视频任务完成且封面截帧成功后通常有值") engine_id: str | None = Field(None, description="本次任务使用的生成引擎ID") engine_snapshot: dict | None = Field( None, description="生成任务创建时的引擎快照，用于前端展示当时使用的模型、比例、尺寸等信息", ) credits_cost: float = Field(0.0, description="本次任务总消耗积分") text_credits_cost: float = Field(0.0, description="文本优化或文本处理消耗积分") text_tokens_used: int = Field(0, description="文本 token 使用量") image_tokens_used: int = Field(0, description="图片 token 使用量") video_tokens_used: int = Field(0, description="视频 token 使用量") retry_count: int = Field(0, description="任务重试次数") poll_count: int = Field(0, description="轮询第三方任务状态次数") error_message: str | None = Field(None, description="错误信息。成功任务一般为 null") created_at: NaiveDatetimeOptional = Field(None, description="任务创建时间") generated_at: NaiveDatetimeOptional = Field(None, description="任务生成完成时间") class GenerationAITaskListOut(BaseModel): """AI生成任务列表响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "total": 1, "items": [ { "id": "0019e0a44895b6d837d", "source_type": "chat_task", "project_id": None, "gen_type": "image", "generation_mode": "chatapi_async", "pipeline_stage": "done", "status": "completed", "original_prompt": "生成一张赛博朋克风格的城市夜景", "optimized_prompt": None, "duration": None, "aspect_ratio": None, "resolution": None, "image_size": "2K", "image_proportion": "1:1", "image_px": "2048x2048", "media_references": None, "provider_task_id": "provider_task_xxx", "seedance_task_id": "provider_task_xxx", "remote_result_url": None, "image_url": "https://example.com/result.png", "video_url": None, "video_cover_url": None, "engine_id": "engine_xxx", "engine_snapshot": {}, "credits_cost": 10.0, "text_credits_cost": 0.0, "text_tokens_used": 0, "image_tokens_used": 0, "video_tokens_used": 0, "retry_count": 0, "poll_count": 3, "error_message": None, "created_at": "2026-05-27T10:12:00", "generated_at": "2026-05-27T10:15:30", } ], } } ) total: int = Field(..., description="符合筛选条件的任务总数") items: list[GenerationAITaskOut] = Field( default_factory=list, description="当前分页的任务列表", ) class GenerationAITaskDeleteOut(BaseModel): """AI生成任务删除响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "message": "任务已删除", "task_id": "0019e0a44895b6d837d", "deleted": True, "freed_size_bytes": 123456, } } ) message: str = Field(..., description="操作结果提示信息") task_id: str = Field(..., description="被软删除的AI生成任务ID") deleted: bool = Field(..., description="是否已完成软删除") freed_size_bytes: int = Field(0, description="本次软删联动释放的有效资源空间字节数") class GenerationAIRetryOut(BaseModel): """AI生成任务重试响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "id": "0019e0a44895b6d837d", "status": "generating", "pipeline_stage": "queued", "message": "任务已重新投递", } } ) id: str = Field(..., description="被重试的任务ID") status: str = Field(..., description="重试后的任务状态") pipeline_stage: str | None = Field(None, description="重试后的任务流水线阶段") message: str = Field(..., description="操作结果提示信息") class GenerationAIRecordHistoryItemOut(BaseModel): """旧 generation_records 历史记录详情响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "id": "0019e0a44895b6d837d", "source_type": "generation_record", "project_id": "project_xxx", "project_name": "默认项目", "gen_type": "image", "generation_mode": "generation_record", "pipeline_stage": None, "status": "completed", "original_prompt": "生成一张赛博朋克风格的城市夜景", "optimized_prompt": "赛博朋克城市夜景，霓虹灯，电影感，高细节", "duration": None, "aspect_ratio": None, "resolution": None, "image_size": "2K", "image_proportion": "1:1", "image_px": "2048x2048", "references": None, "media_references": None, "provider_task_id": "provider_task_xxx", "seedance_task_id": "provider_task_xxx", "remote_result_url": None, "image_url": "https://example.com/result.png", "video_url": None, "video_cover_url": None, "engine_id": None, "engine_snapshot": None, "credits_cost": 10.0, "text_credits_cost": 1.0, "text_tokens_used": 100, "image_tokens_used": 0, "video_tokens_used": 0, "retry_count": 0, "poll_count": 0, "error_message": None, "created_at": "2026-05-27T10:12:00", "generated_at": "2026-05-27T10:15:30", } } ) id: str = Field(..., description="旧生成记录ID") source_type: Literal["generation_record"] = Field( "generation_record", description="历史记录来源固定为 generation_record，用于和 chat_task 历史区分", ) project_id: str | None = Field(None, description="旧项目ID，来源于 generation_records.project_id") project_name: str | None = Field(None, description="旧项目名称，来源于 projects.name；项目不存在时为空") gen_type: str = Field(..., description="生成类型：image=图片，video=视频") generation_mode: str | None = Field( "generation_record", description="兼容新历史结构的生成模式字段。旧表历史固定返回 generation_record", ) pipeline_stage: str | None = Field( None, description="兼容新历史结构的流水线阶段字段。旧 generation_records 无该字段，固定为 null", ) status: str = Field(..., description="记录状态，例如 completed=已完成，failed=失败") original_prompt: str = Field(..., description="用户原始提示词") optimized_prompt: str | None = Field(None, description="优化后的提示词，可能为空") duration: int | None = Field(None, description="视频时长，单位秒。图片记录通常为空") aspect_ratio: str | None = Field(None, description="视频比例，例如 16:9。图片记录通常为空") resolution: str | None = Field(None, description="视频分辨率，例如 480p。图片记录通常为空") image_size: str | None = Field(None, description="图片分辨率档位，例如 2K") image_proportion: str | None = Field(None, description="图片比例，例如 1:1") image_px: str | None = Field(None, description="图片像素尺寸，例如 2048x2048") references: list[dict] | None = Field( None, description="旧接口字段名，来源于 generation_records.media_references 解析后的参考素材列表", ) media_references: list[dict] | None = Field( None, description="兼容新历史结构字段名，和 references 内容一致", ) provider_task_id: str | None = Field(None, description="第三方服务商任务ID，旧表使用 seedance_task_id 兼容填充") seedance_task_id: str | None = Field(None, description="旧服务命名的第三方任务ID字段") remote_result_url: str | None = Field( None, description="第三方远程结果地址。旧 generation_records 未保存该字段，固定为 null", ) image_url: str | None = Field(None, description="最终图片地址。图片记录完成后通常有值") video_url: str | None = Field(None, description="最终视频地址。视频记录完成后通常有值") video_cover_url: str | None = Field(None, description="视频封面图片地址。视频记录完成且封面截帧成功后通常有值") engine_id: str | None = Field( None, description="兼容新历史结构的引擎ID字段。旧 generation_records 未保存该字段，固定为 null", ) engine_snapshot: dict | None = Field( None, description="兼容新历史结构的引擎快照字段。旧 generation_records 未保存该字段，固定为 null", ) credits_cost: float = Field(0.0, description="本次记录总消耗积分") text_credits_cost: float = Field(0.0, description="文本优化或文本处理消耗积分") text_tokens_used: int = Field(0, description="文本 token 使用量") image_tokens_used: int = Field(0, description="图片 token 使用量") video_tokens_used: int = Field(0, description="视频 token 使用量") retry_count: int = Field( 0, description="兼容新历史结构的重试次数字段。旧 generation_records 未保存该字段，固定为 0", ) poll_count: int = Field( 0, description="兼容新历史结构的轮询次数字段。旧 generation_records 未保存该字段，固定为 0", ) error_message: str | None = Field(None, description="错误信息。成功记录一般为 null") created_at: NaiveDatetimeOptional = Field(None, description="记录创建时间") generated_at: NaiveDatetimeOptional = Field(None, description="生成完成时间") GenerationAIHistoryItemOut = Annotated[ GenerationAITaskOut | GenerationAIRecordHistoryItemOut, Field(discriminator="source_type"), ] class GenerationAIHistoryDayGroupOut(BaseModel): """AI生成历史按天分组响应项。""" model_config = ConfigDict( json_schema_extra={ "example": { "generated_date": "2026-05-27", "total": 18, "items": [], } } ) generated_date: str = Field(..., description="生成日期，格式：YYYY-MM-DD") total: int = Field(..., description="当前生成日期下的生成成功记录总数") items: list[GenerationAIHistoryItemOut] = Field( default_factory=list, description=( "当前生成日期下倒序前10条生成记录详情。" "默认 history_source=chat_task 时 item 为 GenerationAITaskOut；" "history_source=generation_record 时 item 为 GenerationAIRecordHistoryItemOut" ), ) class GenerationAIHistoryGroupedOut(BaseModel): """AI生成历史日期分组列表响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "total_days": 2, "page": 1, "page_size": 10, "groups": [ { "generated_date": "2026-05-27", "total": 18, "items": [], }, { "generated_date": "2026-05-26", "total": 6, "items": [], }, ], } } ) total_days: int = Field(..., description="当前生成类型下，用户一共有多少个生成日期分组") page: int = Field(..., description="当前日期分组分页页码") page_size: int = Field(..., description="当前每页返回的日期分组数量，最大10") groups: list[GenerationAIHistoryDayGroupOut] = Field( default_factory=list, description="按生成日期倒序排列的历史记录分组列表", ) class GenerationAIHistoryDayItemsOut(BaseModel): """指定日期下AI生成历史分页响应体。""" model_config = ConfigDict( json_schema_extra={ "example": { "generated_date": "2026-05-27", "total": 18, "page": 2, "page_size": 10, "items": [], } } ) generated_date: str = Field(..., description="当前查询的生成日期，格式：YYYY-MM-DD") total: int = Field(..., description="当前日期下的生成成功记录总数") page: int = Field(..., description="当前日期下的记录分页页码") page_size: int = Field(..., description="当前日期下每页返回的生成记录数量") items: list[GenerationAIHistoryItemOut] = Field( default_factory=list, description=( "当前日期下的生成记录详情列表，按 generated_at 倒序排列。" "默认 history_source=chat_task 时 item 为 GenerationAITaskOut；" "history_source=generation_record 时 item 为 GenerationAIRecordHistoryItemOut" ), )