Files
video-gen/video-gen-api/app/api/v1/generation_ai.py
T
2026-07-11 12:51:48 +08:00

738 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from datetime import datetime, timezone
from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query
from sqlalchemy import and_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_current_user, get_db
from app.models.chat_generation_task import ChatGenerationTask
from app.models.user import User
from app.schemas.generation_ai import (
GenerationAIEngineOptionsOut,
GenerationAIHistoryBatchDeleteOut,
GenerationAIHistoryBatchDeleteRequest,
GenerationAIHistoryDayItemsOut,
GenerationAIHistoryGroupedOut,
GenerationAIRetryOut,
GenerationAITaskDeleteOut,
GenerationAITaskCreate,
GenerationAITaskListOut,
GenerationAITaskOut,
)
from app.services.generation_ai_service import (
create_async_generation_task,
list_generation_ai_engine_options,
list_async_generation_tasks,
list_generation_history_day_items,
list_generation_history_grouped_days,
record_to_out,
soft_delete_chat_generation_task,
)
from app.services.generation_billing_service import (
OWNER_CHAT_GENERATION_TASK,
charge_generation_media_by_params,
get_next_credit_attempt_no,
)
from app.services.generation_history_delete_service import batch_delete_generation_history_items
from app.services.generation_log_service import log_task_event
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
from app.services.private_portrait.reference_resolver import batch_resolve_private_portrait_reference_display_urls, resolve_private_portrait_reference_display_urls
from app.services.resource_capacity_service import assert_user_resource_capacity_available
from app.tasks.celery_app import celery_app
router = APIRouter(
prefix="/generation-ai",
tags=["generation-ai"],
)
@router.get(
"/engines",
response_model=GenerationAIEngineOptionsOut,
summary="获取AI图片/视频可用引擎列表",
description=(
"获取当前启用状态的图片生成引擎和视频生成引擎。"
"返回格式为 engine.image 和 engine.video 两个数组。"
"前端创建 /generation-ai/tasks 任务时,可以把对应引擎 id 作为 engine_id 传入。"
"该接口只返回前端需要展示和选择的模型能力信息,不返回 api_key 等敏感配置。"
),
responses={
200: {
"description": "查询成功,返回当前启用的图片/视频生成引擎列表",
"content": {
"application/json": {
"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,
"supports_first_last_frame": False,
"supports_universal_reference": False,
}
],
}
}
}
},
},
401: {
"description": "未登录或 Token 无效",
},
},
)
async def list_engines(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
return await list_generation_ai_engine_options(db)
@router.post(
"/tasks",
response_model=GenerationAITaskOut,
summary="创建AI图片/视频生成任务",
description=(
"创建一个项目无关的AI生成任务。"
"该接口用于 Chat 风格的图片/视频生成,不再绑定 project_id。"
"创建成功后会写入 chat_generation_tasks 表,并投递 Celery 异步任务。"
"支持 image 图片生成和 video 视频生成。"
"media_references 支持 image/video/audiosource 可为 upload_resource=历史上传素材、"
"private_portrait_asset=私域真人/虚拟素材、空=本次普通上传素材。"
"视频/音频参考素材必须携带 duration,并按 AI 创作原逻辑校验单段 2~15 秒、总时长不超过 15 秒。"
"建议前端传入 idempotency_key,用于防止按钮连点、网络重试导致重复创建任务和重复扣费。"
),
responses={
200: {
"description": "任务创建成功,返回任务详情",
},
400: {
"description": "请求参数错误,例如 gen_type 不支持、图片尺寸不支持、视频参数不支持等",
},
401: {
"description": "未登录或 Token 无效",
},
503: {
"description": "Celery 未启用或消息队列未配置",
},
},
)
async def create_task(
req: GenerationAITaskCreate = Body(
...,
description=(
"AI生成任务创建参数。gen_type=image 时使用图片参数;gen_type=video 时使用视频参数。"
"枚举:gen_type=image/videomedia_references[].type=image/video/audio"
"media_references[].source=upload_resource/private_portrait_asset/空;"
"media_references[].role=first_frame/last_frame/reference_image/reference_video/reference_audio。"
),
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
if celery_app is None:
raise HTTPException(status_code=503, detail="Celery未启用:请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker")
task = await create_async_generation_task(db, current_user, req)
await db.commit()
await log_task_event(
task,
event_type="TASK_CREATED",
to_status="generating",
to_stage="queued",
detail={"gen_type": task.gen_type},
)
from app.tasks.generation_create_tasks import chatapi_create_generation_task
try:
chatapi_create_generation_task.delay(task.id)
except Exception as exc:
await mark_chat_generation_task_failed_and_refund_once(
db,
task_id=task.id,
error_message=f"任务队列投递失败: {exc}",
pipeline_stage="failed",
)
await db.commit()
raise HTTPException(status_code=503, detail="任务队列投递失败,请稍后重试")
refs = await resolve_private_portrait_reference_display_urls(db, record_to_out(task).media_references, user_id=current_user.id)
return record_to_out(task, media_references=refs)
@router.get(
"/tasks",
response_model=GenerationAITaskListOut,
summary="获取AI生成任务列表",
description=(
"分页获取当前登录用户的AI生成任务列表。"
"可按生成类型 gen_type 和任务状态 status 过滤。"
"该接口返回的是普通任务列表,不按日期分组。"
"如果前端需要按生成日期分组展示生成历史记录,请使用 /generation-ai/history 接口。"
"上传素材历史不是生成历史,请使用 /upload-resources/history。"
),
responses={
200: {
"description": "查询成功,返回任务总数和当前分页任务列表",
},
401: {
"description": "未登录或 Token 无效",
},
},
)
async def list_tasks(
gen_type: str | None = Query(
None,
description="生成类型筛选:image=图片任务,video=视频任务;为空表示不过滤生成类型",
examples=["image"],
),
status: str | None = Query(
None,
description=(
"任务状态筛选。常见值:generating=生成中,completed=已完成,failed=失败;"
"为空表示不过滤状态"
),
examples=["completed"],
),
page: int = Query(
1,
ge=1,
description="分页页码,从1开始",
examples=[1],
),
page_size: int = Query(
20,
ge=1,
le=100,
description="每页返回数量,范围 1~100",
examples=[20],
),
user_id: str | None = Query(
None,
description="查询相关用户ID的对应记录[管理后台]",
examples=["0019e0a448a23114888"],
),
user_name: str | None = Query(
None,
description="查询相关用户名的对应记录[管理后台]",
examples=["demo"],
),
engine_id: str | None = Query(
None,
description="按引擎ID筛选[管理后台]",
examples=["0019e1697667d0eff39"],
),
created_start: datetime | None = Query(
None,
description="创建时间起始(含),ISO 格式,例如 2026-07-03T00:00:00",
examples=["2026-07-03T00:00:00"],
),
created_end: datetime | None = Query(
None,
description="创建时间截止(含),ISO 格式,例如 2026-07-03T23:59:59",
examples=["2026-07-03T23:59:59"],
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
is_admin = False
if current_user.user_type == 'admin':
is_admin = True
else:
user_id = current_user.id
total, items = await list_async_generation_tasks(
db,
user_id,
user_name,
gen_type,
status,
page,
page_size,
is_admin,
engine_id=engine_id,
created_start=created_start,
created_end=created_end,
)
# ====================== 在这里加排序(最新在前)======================
if not is_admin:
# 按 created_at 降序(没有则用 id 降序)
items_sorted = sorted(
items,
key=lambda x: x.created_at if x.created_at is not None else x.id,
reverse=False # 升序
)
else:
items_sorted = items
refs_map = await batch_resolve_private_portrait_reference_display_urls(
db,
{item.id: record_to_out(task=item, is_admin=is_admin).media_references for item in items_sorted},
user_id=None if is_admin else current_user.id,
)
return GenerationAITaskListOut(total=total, items=[record_to_out(task=i, is_admin=is_admin, media_references=refs_map.get(i.id)) for i in items_sorted])
@router.get(
"/history",
response_model=GenerationAIHistoryGroupedOut,
summary="获取AI生成历史日期分组",
description=(
"按生成完成日期倒序返回当前用户的AI生成历史记录。"
"默认查询 chat_generation_tasks / ChatGenerationTask 的 AI创作历史。"
"history_source=chat_task 时查询 AI创作;"
"history_source=hot_opening_replicate 时查询爆款开头复刻生成素材;"
"history_source=shot_replicate 时查询拆镜复刻生成素材;"
"history_source=generation_record 时查询 generation_records / GenerationRecord 旧项目生成历史。"
"该接口只返回生成成功的任务,即 status=completed 且 generated_at 不为空的数据。"
"必须通过 gen_type 区分图片和视频。"
"分页对象是生成日期,不是单条记录。"
"每页最多返回10个生成日期分组,每个日期分组内最多返回该日期下倒序前10条生成记录。"
"每条历史记录会返回 generated_resource_id,表示 generated_resources.id;历史脏数据可能为空。"
"如果某一天 total 大于10,前端可调用 /generation-ai/history/{generated_date} 加载该日期下的后续分页数据。"
),
responses={
200: {
"description": "查询成功,返回按生成日期分组的历史记录",
},
400: {
"description": "参数错误,例如 gen_type 不是 image 或 video,或 history_source 不支持",
},
401: {
"description": "未登录或 Token 无效",
},
},
)
async def list_history_grouped_days(
gen_type: str = Query(
...,
description="生成类型:image=图片历史,video=视频历史",
examples=["image"],
),
page: int = Query(
1,
ge=1,
description="日期分组分页页码,从1开始。注意:这里分页的是生成日期,不是单条生成记录",
examples=[1],
),
page_size: int = Query(
10,
ge=1,
le=10,
description="每页返回的生成日期数量,范围 1~10,最大只能获取10天",
examples=[10],
),
history_source: str | None = Query(
None,
description=(
"历史数据来源。默认不传或传 chat_task 查询 AI创作;"
"传 generation_record 查询旧项目生成;"
"传 hot_opening_replicate 查询爆款开头复刻素材;"
"传 shot_replicate 查询拆镜复刻素材"
),
examples=["shot_replicate"],
),
keyword: str | None = Query(
None,
description="提示词搜索关键词,模糊匹配 original_prompt 字段",
examples=["猫咪"],
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
return await list_generation_history_grouped_days(
db=db,
user_id=current_user.id,
gen_type=gen_type,
page=page,
page_size=page_size,
history_source=history_source,
keyword=keyword,
)
@router.delete(
"/history/batch",
response_model=GenerationAIHistoryBatchDeleteOut,
summary="批量删除素材云历史记录",
description=(
"按 history_source 批量软删除素材云历史记录,单次最多30条。"
"generation_record 和 chat_task 入参 ids 为对应记录ID,且只有生成完成后才能删除;"
"hot_opening_replicate 入参 ids 为 module_project_id"
"shot_replicate 入参 ids 为 shot_segment_id。"
"爆款开头复刻和拆镜复刻会联动软删 ModuleGenerationProject、ModuleGenerationStep、关联 ChatGenerationTask 和 generated_resources"
"拆镜复刻还会联动软删 ShotReplicateSegment。"
"如果存在生成中、轮询中、下载中等任务,接口直接拦截,不做失败标记、不退款。"
),
responses={
200: {
"description": "批量软删除成功",
},
400: {
"description": "参数错误,例如 history_source 不支持、ids 为空、超过30条或重复",
},
401: {
"description": "未登录或 Token 无效",
},
404: {
"description": "部分 ID 不存在、不属于当前用户或已删除",
},
409: {
"description": "存在未完成或生成中的记录,当前不能删除",
},
},
)
async def batch_delete_history_items(
req: GenerationAIHistoryBatchDeleteRequest = Body(
...,
description="素材云历史批量删除参数。不同 history_source 对应不同 ID 语义,详见字段说明",
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
return await batch_delete_generation_history_items(
db=db,
current_user=current_user,
history_source=req.history_source,
ids=req.ids,
)
@router.get(
"/history/{generated_date}",
response_model=GenerationAIHistoryDayItemsOut,
summary="获取指定日期下的AI生成历史分页",
description=(
"获取某一个生成日期下的生成成功记录分页。"
"默认查询 chat_generation_tasks / ChatGenerationTask 的 AI创作历史。"
"history_source=chat_task 时查询 AI创作;"
"history_source=hot_opening_replicate 时查询爆款开头复刻生成素材;"
"history_source=shot_replicate 时查询拆镜复刻生成素材;"
"history_source=generation_record 时查询 generation_records / GenerationRecord 旧项目生成历史。"
"该接口用于前端在历史分组列表中继续加载某一天的后续记录。"
"例如 /history 接口中某一天 total=18,但 items 只返回前10条,"
"则前端可以调用本接口 page=2&page_size=10 获取该日期下剩余记录。"
"每条历史记录会返回 generated_resource_id,表示 generated_resources.id;历史脏数据可能为空。"
"该接口同样必须通过 gen_type 区分图片和视频。"
),
responses={
200: {
"description": "查询成功,返回指定日期下的历史记录分页",
},
400: {
"description": "参数错误,例如 generated_date 格式不是 YYYY-MM-DDgen_type 不合法,或 history_source 不支持",
},
401: {
"description": "未登录或 Token 无效",
},
},
)
async def list_history_day_items(
generated_date: str = Path(
...,
description="生成日期,格式:YYYY-MM-DD,例如:2026-05-27",
examples=["2026-05-27"],
),
gen_type: str = Query(
...,
description="生成类型:image=图片历史,video=视频历史",
examples=["image"],
),
page: int = Query(
1,
ge=1,
description="当前日期下的记录分页页码,从1开始",
examples=[1],
),
page_size: int = Query(
10,
ge=1,
le=100,
description="当前日期下每页返回的生成记录数量,范围 1~100",
examples=[10],
),
history_source: str | None = Query(
None,
description=(
"历史数据来源。默认不传或传 chat_task 查询 AI创作;"
"传 generation_record 查询旧项目生成;"
"传 hot_opening_replicate 查询爆款开头复刻素材;"
"传 shot_replicate 查询拆镜复刻素材"
),
examples=["hot_opening_replicate"],
),
keyword: str | None = Query(
None,
description="提示词搜索关键词,模糊匹配 original_prompt 字段",
examples=["猫咪"],
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
return await list_generation_history_day_items(
db=db,
user_id=current_user.id,
gen_type=gen_type,
generated_date=generated_date,
page=page,
page_size=page_size,
history_source=history_source,
keyword=keyword,
)
@router.get(
"/tasks/{task_id}",
response_model=GenerationAITaskOut,
summary="获取AI生成任务详情",
description=(
"根据任务ID获取当前登录用户的AI生成任务详情。"
"只能查询当前用户自己的任务,且只查询 generation_mode=chatapi_async 的任务。"
"如果任务不存在或不属于当前用户,返回404。"
),
responses={
200: {
"description": "查询成功,返回任务详情",
},
401: {
"description": "未登录或 Token 无效",
},
404: {
"description": "任务不存在,或任务不属于当前用户",
},
},
)
async def get_task(
task_id: str = Path(
...,
description="AI生成任务ID",
examples=["0019e0a44895b6d837d"],
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
result = await db.execute(
select(ChatGenerationTask).where(
ChatGenerationTask.id == task_id,
ChatGenerationTask.user_id == current_user.id,
ChatGenerationTask.generation_mode == "chatapi_async",
ChatGenerationTask.deleted_at.is_(None),
)
.limit(1)
)
task = result.scalar_one_or_none()
if not task:
raise HTTPException(status_code=404, detail="任务不存在")
refs = await resolve_private_portrait_reference_display_urls(db, record_to_out(task).media_references, user_id=current_user.id)
return record_to_out(task, media_references=refs)
@router.delete(
"/tasks/{task_id}",
response_model=GenerationAITaskDeleteOut,
summary="删除AI生成任务",
description=(
"软删除当前登录用户自己的AI生成任务。"
"该接口不会物理删除数据库记录和本地文件,只会设置 deleted_at,后续列表、详情、历史统计默认不再返回。"
"删除已完成任务时会联动软删 generated_resources 资源账本,并重新扣减用户有效资源空间统计。"
"如果任务仍处于 generating 生成中状态,接口会直接拦截,不允许删除。"
),
responses={
200: {
"description": "软删除成功,返回任务ID和本次释放的资源空间字节数",
},
400: {
"description": "任务正在生成中,暂不能删除",
},
401: {
"description": "未登录或 Token 无效",
},
404: {
"description": "任务不存在,或任务不属于当前用户,或任务已经被删除",
},
},
)
async def delete_task(
task_id: str = Path(
...,
description="需要删除的AI生成任务ID",
examples=["0019e0a44895b6d837d"],
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
result = await db.execute(
select(ChatGenerationTask).where(
ChatGenerationTask.id == task_id,
ChatGenerationTask.user_id == current_user.id,
ChatGenerationTask.generation_mode == "chatapi_async",
ChatGenerationTask.deleted_at.is_(None),
)
.limit(1)
)
task = result.scalar_one_or_none()
if not task:
raise HTTPException(status_code=404, detail="任务不存在")
if task.status == "generating":
raise HTTPException(status_code=400, detail="当前任务正在生成中,暂不能删除")
deleted_at = datetime.now(timezone.utc)
freed_size_bytes = await soft_delete_chat_generation_task(
db,
task=task,
deleted_at=deleted_at,
)
await db.flush()
return GenerationAITaskDeleteOut(
message="任务已删除",
task_id=task.id,
deleted=True,
freed_size_bytes=freed_size_bytes,
)
@router.post(
"/tasks/{task_id}/retry",
response_model=GenerationAIRetryOut,
summary="重试失败的AI生成任务",
description=(
"重新投递一个失败的AI生成任务。"
"只有 status=failed 的任务允许重试。"
"重试次数超过3次后不允许继续重试。"
"后端会根据任务当前数据自动判断应从哪个流水线阶段恢复,"
"例如重新优化提示词、重新创建第三方任务、继续轮询远程结果或重新下载结果。"
),
responses={
200: {
"description": "任务重新投递成功",
},
400: {
"description": "任务状态不允许重试,或已超过最大重试次数",
},
401: {
"description": "未登录或 Token 无效",
},
404: {
"description": "任务不存在,或任务不属于当前用户",
},
503: {
"description": "Celery 未启用或消息队列未配置",
},
},
)
async def retry_task(
task_id: str = Path(
...,
description="需要重试的AI生成任务ID",
examples=["0019e0a44895b6d837d"],
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
if celery_app is None:
raise HTTPException(status_code=503, detail="Celery未启用:请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker")
result = await db.execute(
select(ChatGenerationTask).where(
ChatGenerationTask.id == task_id,
ChatGenerationTask.user_id == current_user.id,
ChatGenerationTask.generation_mode == "chatapi_async",
ChatGenerationTask.deleted_at.is_(None),
)
.with_for_update()
.limit(1)
)
task = result.scalar_one_or_none()
if not task:
raise HTTPException(status_code=404, detail="任务不存在")
if task.status != "failed":
raise HTTPException(status_code=400, detail="只有失败任务可以重试")
await assert_user_resource_capacity_available(db, current_user.id)
attempt_no = await get_next_credit_attempt_no(
db,
owner_type=OWNER_CHAT_GENERATION_TASK,
owner_id=task.id,
)
media_billing = await charge_generation_media_by_params(
db,
user_id=task.user_id,
record_id=task.id,
gen_type=task.gen_type,
image_size=task.image_size,
duration=task.duration,
resolution=task.resolution,
engine_id=task.engine_id,
project_name="AI生成任务",
description_prefix="Chat任务重试",
owner_type=OWNER_CHAT_GENERATION_TASK,
attempt_no=attempt_no,
)
task.status = "generating"
task.pipeline_stage = "queued"
task.error_message = None
task.poll_count = 0
task.last_poll_at = None
task.provider_task_id = None
task.seedance_task_id = None
task.remote_result_url = None
task.provider_response_json = None
task.image_url = None
task.video_url = None
task.video_cover_url = None
task.generated_at = None
task.credits_cost = round(float(task.credits_cost or 0) + media_billing.total_charged, 2)
await db.commit()
from app.tasks.generation_create_tasks import chatapi_create_generation_task
try:
chatapi_create_generation_task.delay(task.id)
except Exception as exc:
await mark_chat_generation_task_failed_and_refund_once(
db,
task_id=task.id,
error_message=f"任务队列投递失败: {exc}",
pipeline_stage="failed",
)
await db.commit()
raise HTTPException(status_code=503, detail="任务队列投递失败,请稍后重试")
return GenerationAIRetryOut(
id=task.id,
status=task.status,
pipeline_stage=task.pipeline_stage,
message="任务已重新扣费并重新投递",
)