851 lines
34 KiB
Python
851 lines
34 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from types import SimpleNamespace
|
||
|
||
from fastapi import APIRouter, Body, Depends, File, HTTPException, Path, Query, UploadFile
|
||
from sqlalchemy import inspect as sa_inspect
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.dependencies import get_current_user, get_db
|
||
from app.models.user import User
|
||
from app.enums.common import ModuleProjectStatusEnum, ModuleEventTypeEnum
|
||
from app.enums.generation_task import GenerationOwnerType
|
||
from app.enums.hot_opening_replicate import HotOpeningLogEventEnum, HotOpeningStepCodeEnum, ModuleCodeEnum
|
||
from app.schemas.hot_opening_replicate import (
|
||
HotOpeningActionOut,
|
||
HotOpeningDeleteOut,
|
||
HotOpeningGenerateImagePromptRequest,
|
||
HotOpeningGenerateImageRequest,
|
||
HotOpeningGenerateVideoPromptRequest,
|
||
HotOpeningGenerateVideoRequest,
|
||
HotOpeningImagePromptUpdateRequest,
|
||
HotOpeningMaterialUpdateRequest,
|
||
HotOpeningSpecOut,
|
||
HotOpeningTaskCreate,
|
||
HotOpeningTaskDetailOut,
|
||
HotOpeningTaskListOut,
|
||
HotOpeningVideoPromptSchemaUpdateRequest,
|
||
)
|
||
from app.services.hot_opening_replicate_service import (
|
||
_get_project_for_user,
|
||
create_hot_opening_project,
|
||
delete_hot_opening_project,
|
||
generate_image_from_prompt,
|
||
generate_video_from_prompt,
|
||
list_hot_opening_projects,
|
||
mark_hot_opening_step_dispatch_failed,
|
||
project_to_detail_out,
|
||
submit_image_prompt_optimize,
|
||
submit_video_prompt_optimize,
|
||
update_hot_opening_image_prompt,
|
||
update_hot_opening_material_input,
|
||
update_hot_opening_video_prompt_schema,
|
||
)
|
||
from app.services.module_generation_log_service import log_module_error, log_module_event_file
|
||
from app.services.module_async_recovery_service import (
|
||
TASK_HOT_IMAGE_PROMPT,
|
||
TASK_HOT_VIDEO_PROMPT,
|
||
register_module_step_task,
|
||
)
|
||
from app.tasks.celery_app import celery_app
|
||
from app.enums.upload_resource import UploadResourceEventEnum, UploadResourceModuleEnum, UploadResourceSourceModelEnum, UploadResourceTypeEnum
|
||
from app.services.upload_resource import upload_reference_file, bind_upload_resources, cleanup_upload_resource_files_after_commit
|
||
from app.services.upload_resource.log_service import log_upload_resource_exception, safe_rollback_with_log
|
||
|
||
MODULE = ModuleCodeEnum.HOT_OPENING_REPLICATE.value
|
||
|
||
|
||
|
||
def _safe_user_id(user: object | None) -> str | None:
|
||
"""从 ORM 对象中安全取用户ID，避免 rollback/commit 后访问过期属性触发 MissingGreenlet。"""
|
||
if user is None:
|
||
return None
|
||
try:
|
||
value = getattr(user, "__dict__", {}).get("id")
|
||
if value is not None:
|
||
return str(value)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
identity = sa_inspect(user).identity
|
||
if identity:
|
||
return str(identity[0])
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _safe_user_is_admin(user: object | None) -> bool:
|
||
"""安全判断管理员身份；如果对象属性已过期，保守按普通用户处理。"""
|
||
if user is None:
|
||
return False
|
||
try:
|
||
data = getattr(user, "__dict__", {})
|
||
if "is_admin" in data:
|
||
return bool(data.get("is_admin"))
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
def _user_context(user: object | None) -> SimpleNamespace:
|
||
return SimpleNamespace(id=_safe_user_id(user), is_admin=_safe_user_is_admin(user))
|
||
|
||
router = APIRouter(
|
||
prefix="/hot-opening-replications",
|
||
tags=["hot-opening-replications"],
|
||
)
|
||
|
||
|
||
|
||
|
||
def _log_api_error(
|
||
*,
|
||
event_type: str,
|
||
current_user: User | None = None,
|
||
project_id: str | None = None,
|
||
step_id: str | None = None,
|
||
message: str | None = None,
|
||
exc: BaseException | None = None,
|
||
detail: dict | None = None,
|
||
) -> None:
|
||
log_module_error(
|
||
module=MODULE,
|
||
event_type=event_type,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
user_id=_safe_user_id(current_user),
|
||
message=message,
|
||
detail=detail,
|
||
exc=exc,
|
||
)
|
||
|
||
|
||
def _log_api_exception_from_locals(exc: BaseException, local_values: dict, message: str) -> None:
|
||
current_user = local_values.get("current_user")
|
||
project_id = local_values.get("project_id_value") or local_values.get("project_id")
|
||
step_id = local_values.get("step_id_value") or local_values.get("step_id")
|
||
req = local_values.get("req")
|
||
detail = {"request": req.model_dump() if hasattr(req, "model_dump") else str(req) if req is not None else None}
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.API_REQUEST_FAILED.value,
|
||
current_user=current_user if isinstance(current_user, User) else None,
|
||
project_id=str(project_id) if project_id else None,
|
||
step_id=str(step_id) if step_id else None,
|
||
message=message,
|
||
detail=detail,
|
||
exc=exc,
|
||
)
|
||
|
||
async def _reload_project_detail(
|
||
db: AsyncSession,
|
||
current_user: User,
|
||
project_id: str,
|
||
) -> HotOpeningTaskDetailOut:
|
||
"""提交事务后统一重新查询详情，避免继续访问 commit 前 ORM 对象。"""
|
||
project = await _get_project_for_user(
|
||
db,
|
||
project_id=project_id,
|
||
user=_user_context(current_user),
|
||
for_update=False,
|
||
populate_existing=True,
|
||
)
|
||
return await project_to_detail_out(db, project)
|
||
|
||
|
||
async def _mark_dispatch_failed_and_raise(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
step_id: str | None,
|
||
message: str,
|
||
) -> None:
|
||
"""Celery 投递失败后，数据库事务已提交，单独标记步骤失败，避免一直 processing。"""
|
||
if step_id:
|
||
try:
|
||
await mark_hot_opening_step_dispatch_failed(
|
||
db,
|
||
current_user=_user_context(current_user),
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
error_message=message,
|
||
)
|
||
await db.commit()
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="Celery 投递失败后标记步骤失败也失败",
|
||
detail={"dispatch_error": message},
|
||
exc=exc,
|
||
)
|
||
log_module_error(
|
||
module=MODULE,
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISPATCH_FAILED.value,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
user_id=_safe_user_id(current_user),
|
||
message=message,
|
||
detail={"reason": "celery_dispatch_failed"},
|
||
error=message,
|
||
)
|
||
raise HTTPException(status_code=503, detail=message)
|
||
|
||
|
||
@router.get(
|
||
"/spec",
|
||
response_model=HotOpeningSpecOut,
|
||
summary="查询爆款开头复刻模块状态枚举和步骤 JSON 结构说明",
|
||
description="保留给调试和前端兜底读取。原业务接口已经在 Path、Query、Body 和响应模型字段上直接展示参数说明与枚举值。",
|
||
)
|
||
async def get_spec():
|
||
return HotOpeningSpecOut()
|
||
|
||
|
||
@router.post(
|
||
"/upload-image",
|
||
summary="上传爆款开头复刻图片素材",
|
||
description="上传后先记录为 pending UploadResource；创建项目成功后绑定 ModuleGenerationProject。",
|
||
)
|
||
async def upload_hot_opening_image(
|
||
file: UploadFile = File(...),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await upload_reference_file(
|
||
db,
|
||
file=file,
|
||
current_user=current_user,
|
||
module=UploadResourceModuleEnum.HOT_OPENING_REPLICATE.value,
|
||
resource_type=UploadResourceTypeEnum.IMAGE.value,
|
||
gen_type="video",
|
||
)
|
||
await db.commit()
|
||
return {
|
||
"url": result.url,
|
||
"filename": result.filename,
|
||
"type": "image",
|
||
"module": result.module,
|
||
"resource_id": result.resource_id,
|
||
"file_size_bytes": result.file_size_bytes,
|
||
}
|
||
|
||
|
||
@router.post(
|
||
"/upload-video",
|
||
summary="上传爆款开头复刻视频素材",
|
||
description="上传后先记录为 pending UploadResource；创建项目成功后绑定 ModuleGenerationProject。",
|
||
)
|
||
async def upload_hot_opening_video(
|
||
file: UploadFile = File(...),
|
||
duration_seconds: float | None = Query(None, description="前端识别的视频时长秒数，可选"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await upload_reference_file(
|
||
db,
|
||
file=file,
|
||
current_user=current_user,
|
||
module=UploadResourceModuleEnum.HOT_OPENING_REPLICATE.value,
|
||
resource_type=UploadResourceTypeEnum.VIDEO.value,
|
||
duration_seconds=duration_seconds,
|
||
)
|
||
await db.commit()
|
||
return {
|
||
"url": result.url,
|
||
"filename": result.filename,
|
||
"type": "video",
|
||
"module": result.module,
|
||
"resource_id": result.resource_id,
|
||
"file_size_bytes": result.file_size_bytes,
|
||
"duration_seconds": result.duration_seconds,
|
||
}
|
||
|
||
|
||
@router.post(
|
||
"/tasks",
|
||
response_model=HotOpeningTaskDetailOut,
|
||
summary="创建爆款开头复刻总任务项目",
|
||
description=(
|
||
"创建爆款开头复刻总任务项目。总任务表 id 就是项目ID，不再传 project_id。"
|
||
"接口只同步创建第1个素材输入子任务，保存素材视频链接、素材图片链接、视频素材内容项目名称、生成项目名称和50字核心内容点。"
|
||
"后端不开发上传接口，也不校验素材文件时长、大小、格式，直接使用前端已有上传接口返回的链接。"
|
||
"创建后不会自动生成第2步图片 AI 提词，需要前端手动调用 generate-image-prompt。"
|
||
),
|
||
)
|
||
async def create_task(
|
||
req: HotOpeningTaskCreate = Body(..., description="爆款开头复刻创建参数，只包含素材链接和项目描述，不包含图片/视频引擎参数"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
log_module_event_file(
|
||
module=MODULE,
|
||
event_type=ModuleEventTypeEnum.V1_CREATE_BLOCKED.value,
|
||
user_id=current_user.id,
|
||
message="拦截爆款开头复刻 V1 创建请求",
|
||
detail={"api_version": "v1", "flow_version": "v1"},
|
||
)
|
||
raise HTTPException(
|
||
status_code=410,
|
||
detail="V1 创建流程已停止，请使用 V2 API",
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/tasks",
|
||
response_model=HotOpeningTaskListOut,
|
||
summary="查询爆款开头复刻总任务项目列表",
|
||
description="分页查询爆款开头复刻总任务项目列表。普通用户只能查看自己的项目，管理员可查看全部。",
|
||
)
|
||
async def list_tasks(
|
||
status: ModuleProjectStatusEnum | None = Query(None, description="总任务状态筛选：pending=已创建，waiting_user=等待用户操作，processing=处理中，completed=完成，failed=失败，cancelled=取消；为空不过滤"),
|
||
keyword: str | None = Query(None, description="关键词搜索：项目ID、标题、生成项目名称、素材项目名称、核心内容点；普通用户只在自己的数据内搜索"),
|
||
user_id: str | None = Query(None, description="管理员专用：按用户ID筛选；普通用户不可使用"),
|
||
user_name: str | None = Query(None, description="管理员专用：按用户名模糊筛选；普通用户不可使用"),
|
||
created_start: datetime | None = Query(None, description="管理员专用：创建时间开始，ISO datetime；普通用户不可使用"),
|
||
created_end: datetime | None = Query(None, description="管理员专用：创建时间结束，ISO datetime；普通用户不可使用"),
|
||
page: int = Query(1, ge=1, description="分页页码，从1开始"),
|
||
page_size: int = Query(20, ge=1, le=100, description="每页数量，范围1-100"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
admin_only_params = {
|
||
"user_id": user_id,
|
||
"user_name": user_name,
|
||
"created_start": created_start,
|
||
"created_end": created_end,
|
||
}
|
||
if not _safe_user_is_admin(current_user) and any(value is not None and str(value).strip() != "" for value in admin_only_params.values()):
|
||
raise HTTPException(status_code=403, detail="当前搜索条件仅管理员可用")
|
||
|
||
return await list_hot_opening_projects(
|
||
db,
|
||
current_user=current_user,
|
||
status=status.value if status else None,
|
||
keyword=keyword,
|
||
user_id=user_id,
|
||
user_name=user_name,
|
||
created_start=created_start,
|
||
created_end=created_end,
|
||
page=page,
|
||
page_size=page_size,
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/tasks/{project_id}",
|
||
response_model=HotOpeningTaskDetailOut,
|
||
summary="获取爆款开头复刻总任务项目详情",
|
||
description=(
|
||
"获取爆款开头复刻总任务详情。详情会聚合返回第1步素材信息、第2步图片提词、第3步图片引擎和参数、"
|
||
"第4步视频提词 JSON schema、第5步视频引擎和参数、最终图片、最终视频和完整子任务列表。"
|
||
),
|
||
)
|
||
async def get_task(
|
||
project_id: str = Path(..., description="总任务项目ID，即 module_generation_projects.id"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
return await _reload_project_detail(db, current_user, project_id)
|
||
|
||
|
||
@router.put(
|
||
"/tasks/{project_id}/material",
|
||
response_model=HotOpeningActionOut,
|
||
summary="修改第1步素材输入并重建第1步新版本",
|
||
description=(
|
||
"反复修改爆款开头复刻第1步素材输入。"
|
||
"接口会软删除旧第1步以及第2、3、4、5步当前有效子任务，联动软删除关联 ChatGenerationTask，"
|
||
"然后新建第1步 material_input 的 version+1，项目回到 waiting_user 状态。"
|
||
),
|
||
)
|
||
async def update_material(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
req: HotOpeningMaterialUpdateRequest = Body(..., description="第1步素材输入修改参数，至少传一个字段"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
try:
|
||
project_id_value, step_id_value = await update_hot_opening_material_input(
|
||
db,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
req=req,
|
||
)
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_exception_from_locals(exc, locals(), f"修改素材输入失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=f"修改素材输入失败: {exc}")
|
||
|
||
return HotOpeningActionOut(
|
||
message="素材输入已修改，旧步骤已软删除，请重新生成图片 AI 提词",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.put(
|
||
"/tasks/{project_id}/steps/{step_id}/image-prompt",
|
||
response_model=HotOpeningActionOut,
|
||
summary="直接修改第2步图片 AI 优化提词",
|
||
description=(
|
||
"直接修改第2步图片 AI 优化提词，不调用 AI、不扣积分。"
|
||
"保存后会软删除第3、4、5步当前有效任务和关联 ChatGenerationTask，"
|
||
"清空旧图片/视频结果，让用户从图片生成开始重新执行。"
|
||
),
|
||
)
|
||
async def update_image_prompt(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
step_id: str = Path(..., description="第2步图片 AI 提词子任务ID"),
|
||
req: HotOpeningImagePromptUpdateRequest = Body(..., description="图片 AI 优化提词修改参数"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
try:
|
||
project, step = await update_hot_opening_image_prompt(
|
||
db,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
req=req,
|
||
)
|
||
project_id_value = str(project.id)
|
||
step_id_value = str(step.id)
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_exception_from_locals(exc, locals(), f"修改图片 AI 提词失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=f"修改图片 AI 提词失败: {exc}")
|
||
|
||
return HotOpeningActionOut(
|
||
message="图片 AI 提词已修改，后续步骤已软删除，请重新生成图片",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.put(
|
||
"/tasks/{project_id}/steps/{step_id}/video-prompt-schema",
|
||
response_model=HotOpeningActionOut,
|
||
summary="修改第4步视频 AI 提词 JSON schema",
|
||
description=(
|
||
"修改第4步视频 AI 提词 JSON schema，不调用 AI、不扣积分。"
|
||
"前端提交的 schema 只作为 patch，服务端会锁定视频时长、比例、清晰度、帧率、推荐分辨率、"
|
||
"动作/镜头/动态时间规划数组长度和时间段、输出规格、质量控制、合规控制、schema_version、schema_usage。"
|
||
"最终提示词允许修改，但会清洗秒数、比例、分辨率、帧率等视频参数。保存后软删除第5步视频生成任务。"
|
||
),
|
||
)
|
||
async def update_video_prompt_schema(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
step_id: str = Path(..., description="第4步视频 AI 提词子任务ID"),
|
||
req: HotOpeningVideoPromptSchemaUpdateRequest = Body(..., description="视频 AI 提词 schema 修改参数"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
try:
|
||
project, step = await update_hot_opening_video_prompt_schema(
|
||
db,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
req=req,
|
||
)
|
||
project_id_value = str(project.id)
|
||
step_id_value = str(step.id)
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
raise HTTPException(status_code=500, detail=f"修改视频 AI 提词 schema 失败: {exc}")
|
||
|
||
return HotOpeningActionOut(
|
||
message="视频 AI 提词 schema 已修改，第5步视频生成任务已软删除，请重新生成视频",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/tasks/{project_id}/steps/{step_id}/generate-image-prompt",
|
||
response_model=HotOpeningActionOut,
|
||
summary="基于素材输入手动生成图片 AI 提词",
|
||
description=(
|
||
"基于第1步素材输入子任务手动生成第2步图片 AI 提词。"
|
||
"如果已存在旧的第2、3、4、5步，会先软删除旧步骤，再创建新的第2步。"
|
||
),
|
||
)
|
||
async def generate_image_prompt(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
step_id: str = Path(..., description="第1步素材输入子任务ID"),
|
||
req: HotOpeningGenerateImagePromptRequest = Body(default_factory=HotOpeningGenerateImagePromptRequest, description="图片提词生成参数，当前无需传参，额外字段会忽略"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
_ = req
|
||
if celery_app is None:
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker",
|
||
detail={"api": "hot_opening_replicate"},
|
||
)
|
||
raise HTTPException(status_code=503, detail="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker")
|
||
|
||
try:
|
||
project, step = await submit_image_prompt_optimize(db, current_user=current_user, project_id=project_id, material_step_id=step_id)
|
||
project_id_value = str(project.id)
|
||
step_id_value = str(step.id)
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_exception_from_locals(exc, locals(), f"图片提词任务创建失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=f"图片提词任务创建失败: {exc}")
|
||
|
||
from app.tasks.hot_opening_replicate_tasks import start_image_prompt_optimize
|
||
|
||
await register_module_step_task(
|
||
module=MODULE,
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
||
task_name=TASK_HOT_IMAGE_PROMPT,
|
||
)
|
||
try:
|
||
start_image_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue="gen_chatapi_create", countdown=0)
|
||
except Exception as exc:
|
||
await _mark_dispatch_failed_and_raise(
|
||
db,
|
||
current_user=current_user,
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
message=f"图片提词任务投递失败: {exc}",
|
||
)
|
||
|
||
return HotOpeningActionOut(
|
||
message="图片 AI 提词任务已提交",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/tasks/{project_id}/steps/{step_id}/generate-image",
|
||
response_model=HotOpeningActionOut,
|
||
summary="基于图片 AI 提词生成新项目图片",
|
||
description=(
|
||
"基于第2步图片 AI 提词生成新项目图片。调用时传入图片生成引擎和图片生成参数。"
|
||
"后端会创建第3步图片生成子任务，ChatGenerationTask 幂等键由后端按任务ID自动生成，不再使用前端幂等键。"
|
||
"图片生成媒体积分在创建 ChatGenerationTask 时扣除，生成失败走媒体积分退款。"
|
||
"如果已存在旧的第3、4、5步，会先软删除旧步骤，再创建新的第3步。"
|
||
),
|
||
)
|
||
async def generate_image(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
step_id: str = Path(..., description="第2步图片 AI 提词子任务ID"),
|
||
req: HotOpeningGenerateImageRequest = Body(..., description="图片生成引擎和参数"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
if celery_app is None:
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker",
|
||
detail={"api": "hot_opening_replicate"},
|
||
)
|
||
raise HTTPException(status_code=503, detail="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker")
|
||
|
||
try:
|
||
project, step = await generate_image_from_prompt(db, current_user=current_user, project_id=project_id, prompt_step_id=step_id, req=req)
|
||
project_id_value = str(project.id)
|
||
step_id_value = str(step.id)
|
||
chat_task_id_value = step.chat_task_id
|
||
if not chat_task_id_value:
|
||
raise HTTPException(status_code=500, detail="图片生成任务创建失败：chat_task_id为空")
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_exception_from_locals(exc, locals(), f"图片生成任务创建失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=f"图片生成任务创建失败: {exc}")
|
||
|
||
from app.tasks.generation_create_tasks import chatapi_create_generation_task
|
||
|
||
try:
|
||
chatapi_create_generation_task.apply_async(
|
||
args=[chat_task_id_value],
|
||
kwargs={"owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": 1},
|
||
queue="gen_chatapi_create",
|
||
countdown=0,
|
||
)
|
||
except Exception as exc:
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISPATCH_FAILED.value,
|
||
current_user=current_user,
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
message=f"图片生成任务投递失败，等待生成恢复任务补投: {exc}",
|
||
detail={"recoverable": True, "chat_task_id": chat_task_id_value},
|
||
exc=exc,
|
||
)
|
||
|
||
return HotOpeningActionOut(
|
||
message="图片生成任务已提交",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/tasks/{project_id}/steps/{step_id}/generate-video-prompt",
|
||
response_model=HotOpeningActionOut,
|
||
summary="基于图片结果手动生成视频 AI 提词",
|
||
description=(
|
||
"基于第3步图片生成子任务手动生成第4步视频 AI 提词 JSON schema。"
|
||
"视频时长、比例、分辨率集中在本步骤确定并写入 step.output_json.payload.params_used_for_prompt。"
|
||
"视频 AI 提词成功后按文本 token 扣积分；该文本积分不参与后续视频生成失败退款。"
|
||
"如果已存在旧的第4、5步，会先软删除旧步骤，再创建新的第4步。"
|
||
),
|
||
)
|
||
async def generate_video_prompt(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
step_id: str = Path(..., description="第3步图片生成子任务ID"),
|
||
req: HotOpeningGenerateVideoPromptRequest = Body(..., description="视频提词生成参数，用于读取接口配置并规划视频时长、比例、分辨率"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
if celery_app is None:
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker",
|
||
detail={"api": "hot_opening_replicate"},
|
||
)
|
||
raise HTTPException(status_code=503, detail="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker")
|
||
|
||
try:
|
||
project, step = await submit_video_prompt_optimize(db, current_user=current_user, project_id=project_id, image_step_id=step_id, req=req)
|
||
project_id_value = str(project.id)
|
||
step_id_value = str(step.id)
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_exception_from_locals(exc, locals(), f"视频提词任务创建失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=f"视频提词任务创建失败: {exc}")
|
||
|
||
from app.tasks.hot_opening_replicate_tasks import start_video_prompt_optimize
|
||
|
||
await register_module_step_task(
|
||
module=MODULE,
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
step_code=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
|
||
task_name=TASK_HOT_VIDEO_PROMPT,
|
||
)
|
||
try:
|
||
start_video_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue="gen_chatapi_create", countdown=0)
|
||
except Exception as exc:
|
||
await _mark_dispatch_failed_and_raise(
|
||
db,
|
||
current_user=current_user,
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
message=f"视频提词任务投递失败: {exc}",
|
||
)
|
||
|
||
return HotOpeningActionOut(
|
||
message="视频 AI 提词任务已提交",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/tasks/{project_id}/steps/{step_id}/generate-video",
|
||
response_model=HotOpeningActionOut,
|
||
summary="基于视频 AI 提词生成最终视频",
|
||
description=(
|
||
"基于第4步视频 AI 提词生成最终视频。请求体只需要选择视频生成引擎 engine_id。"
|
||
"视频时长、比例、分辨率从第4步视频提词优化结果读取，不再由本接口动态传入。"
|
||
"关联 ChatGenerationTask 的 original_prompt 和 optimized_prompt 都使用第4步生成的 prompt_schema JSON 字符串。"
|
||
"ChatGenerationTask 幂等键由后端按任务ID自动生成；视频生成媒体积分失败时走退款。"
|
||
"如果已存在旧的第5步，会先软删除旧步骤，再创建新的第5步。"
|
||
),
|
||
)
|
||
async def generate_video(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
step_id: str = Path(..., description="第4步视频 AI 提词子任务ID"),
|
||
req: HotOpeningGenerateVideoRequest = Body(..., description="视频生成参数：只传 engine_id，其它视频参数继承第4步视频提词结果"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
if celery_app is None:
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker",
|
||
detail={"api": "hot_opening_replicate"},
|
||
)
|
||
raise HTTPException(status_code=503, detail="Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker")
|
||
|
||
try:
|
||
project, step = await generate_video_from_prompt(db, current_user=current_user, project_id=project_id, prompt_step_id=step_id, req=req)
|
||
project_id_value = str(project.id)
|
||
step_id_value = str(step.id)
|
||
chat_task_id_value = step.chat_task_id
|
||
if not chat_task_id_value:
|
||
raise HTTPException(status_code=500, detail="视频生成任务创建失败：chat_task_id为空")
|
||
await db.commit()
|
||
except HTTPException:
|
||
await db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
_log_api_exception_from_locals(exc, locals(), f"视频生成任务创建失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=f"视频生成任务创建失败: {exc}")
|
||
|
||
from app.tasks.generation_create_tasks import chatapi_create_generation_task
|
||
|
||
try:
|
||
chatapi_create_generation_task.apply_async(
|
||
args=[chat_task_id_value],
|
||
kwargs={"owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": 1},
|
||
queue="gen_chatapi_create",
|
||
countdown=0,
|
||
)
|
||
except Exception as exc:
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.CELERY_DISPATCH_FAILED.value,
|
||
current_user=current_user,
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
message=f"视频生成任务投递失败，等待生成恢复任务补投: {exc}",
|
||
detail={"recoverable": True, "chat_task_id": chat_task_id_value},
|
||
exc=exc,
|
||
)
|
||
|
||
return HotOpeningActionOut(
|
||
message="视频生成任务已提交",
|
||
project_id=project_id_value,
|
||
step_id=step_id_value,
|
||
detail=await _reload_project_detail(db, current_user, project_id_value),
|
||
)
|
||
|
||
|
||
@router.delete(
|
||
"/tasks/{project_id}",
|
||
response_model=HotOpeningDeleteOut,
|
||
summary="删除爆款开头复刻总任务项目",
|
||
description="软删除爆款开头复刻总任务项目，并联动软删除当前有效子任务和关联的 ChatGenerationTask。",
|
||
)
|
||
async def delete_task(
|
||
project_id: str = Path(..., description="总任务项目ID"),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
user_id = _safe_user_id(current_user)
|
||
pending_ids: list[str] = []
|
||
try:
|
||
result = await delete_hot_opening_project(db, current_user=current_user, project_id=project_id)
|
||
pending_ids = list(result.pending_delete_resource_ids or [])
|
||
await db.commit()
|
||
except HTTPException as exc:
|
||
await safe_rollback_with_log(
|
||
db,
|
||
event_type=UploadResourceEventEnum.HOT_OPENING_DELETE_PROJECT_ROLLBACK_FAILED.value,
|
||
message="删除爆款开头复刻项目 HTTPException 回滚失败",
|
||
user_id=user_id,
|
||
module=MODULE,
|
||
detail={"project_id": project_id},
|
||
original_exc=exc,
|
||
)
|
||
raise
|
||
except Exception as exc: # noqa: BLE001
|
||
await safe_rollback_with_log(
|
||
db,
|
||
event_type=UploadResourceEventEnum.HOT_OPENING_DELETE_PROJECT_ROLLBACK_FAILED.value,
|
||
message="删除爆款开头复刻项目主事务回滚失败",
|
||
user_id=user_id,
|
||
module=MODULE,
|
||
detail={"project_id": project_id},
|
||
original_exc=exc,
|
||
)
|
||
_log_api_error(
|
||
event_type=HotOpeningLogEventEnum.API_REQUEST_FAILED.value,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
message=f"删除爆款开头复刻项目失败: {exc}",
|
||
detail={"project_id": project_id},
|
||
exc=exc,
|
||
)
|
||
log_upload_resource_exception(
|
||
event_type=UploadResourceEventEnum.HOT_OPENING_DELETE_PROJECT_FAILED.value,
|
||
message=f"删除爆款开头复刻项目失败: {exc}",
|
||
user_id=user_id,
|
||
module=MODULE,
|
||
detail={"project_id": project_id},
|
||
exc=exc,
|
||
)
|
||
raise HTTPException(status_code=500, detail=f"删除爆款开头复刻项目失败: {exc}")
|
||
|
||
if pending_ids:
|
||
try:
|
||
await cleanup_upload_resource_files_after_commit(db, resource_ids=pending_ids)
|
||
await db.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
await safe_rollback_with_log(
|
||
db,
|
||
event_type=UploadResourceEventEnum.HOT_OPENING_DELETE_PROJECT_ROLLBACK_FAILED.value,
|
||
message="删除爆款开头复刻项目 cleanup 事务回滚失败",
|
||
user_id=user_id,
|
||
module=MODULE,
|
||
detail={"project_id": project_id, "pending_ids": pending_ids},
|
||
original_exc=exc,
|
||
)
|
||
log_upload_resource_exception(
|
||
event_type=UploadResourceEventEnum.HOT_OPENING_DELETE_PROJECT_CLEANUP_FAILED.value,
|
||
message=f"删除爆款开头复刻项目后清理真实文件失败: {exc}",
|
||
user_id=user_id,
|
||
resource_ids=pending_ids,
|
||
module=MODULE,
|
||
detail={"project_id": project_id},
|
||
exc=exc,
|
||
)
|
||
return result
|