from __future__ import annotations from types import SimpleNamespace from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query 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.hot_opening_replicate import 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 from app.tasks.celery_app import celery_app 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="API_REQUEST_FAILED", 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="CELERY_DISPATCH_MARK_FAILED", 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="CELERY_DISPATCH_FAILED", 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="返回总任务状态、子任务状态、5个固定步骤编码以及每个步骤 input_json/output_json 的统一结构示例，方便前端和排查人员对照。", ) async def get_spec(): return HotOpeningSpecOut() @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), ): try: project = await create_hot_opening_project(db, current_user, req) project_id_value = str(project.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}") return await _reload_project_detail(db, current_user, project_id_value) @router.get( "/tasks", response_model=HotOpeningTaskListOut, summary="查询爆款开头复刻总任务项目列表", description="分页查询爆款开头复刻总任务项目列表。普通用户只能查看自己的项目，管理员可查看全部。", ) async def list_tasks( status: str | None = Query(None, description="总任务状态筛选，例如 waiting_user、processing、completed、failed；为空不过滤"), 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), ): return await list_hot_opening_projects(db, current_user=current_user, status=status, 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="CELERY_DISABLED", 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 try: start_image_prompt_optimize.delay(project_id_value, step_id_value) 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="CELERY_DISABLED", 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.delay(chat_task_id_value) 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="图片生成任务已提交", 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="CELERY_DISABLED", 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 try: start_video_prompt_optimize.delay(project_id_value, step_id_value) 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="CELERY_DISABLED", 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.delay(chat_task_id_value) 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="视频生成任务已提交", 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), ): try: result = await delete_hot_opening_project(db, current_user=current_user, project_id=project_id) await db.commit() return result except HTTPException: await db.rollback() raise except Exception as exc: await db.rollback() raise HTTPException(status_code=500, detail=f"删除爆款开头复刻项目失败: {exc}")