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.shot_replicate import ModuleCodeEnum, ShotReplicateStepCodeEnum from app.schemas.shot_replicate import ( ShotReplicateActionOut, ShotReplicateDeleteOut, ShotReplicateGenerateImagePromptRequest, ShotReplicateGenerateImageRequest, ShotReplicateGenerateVideoPromptRequest, ShotReplicateGenerateVideoRequest, ShotReplicateImagePromptUpdateRequest, ShotReplicateMaterialUpdateRequest, ShotReplicateSpecOut, ShotReplicateTaskDetailOut, ShotReplicateVideoPromptSchemaUpdateRequest, ShotSegmentDetailOut, ShotSegmentListOut, ShotSegmentReplicationCreateRequest, ShotSplitByAIOut, ShotSplitByAIRequest, ShotSplitCustomOut, ShotSplitCustomRequest, ShotTaskSetCreate, ShotTaskSetDetailOut, ShotTaskSetListOut, ) from app.services.shot_replicate_flow_service import ( _get_project_for_user, create_shot_replicate_project_from_segment, delete_shot_replicate_project, generate_image_from_prompt, generate_video_from_prompt, mark_shot_replicate_step_dispatch_failed, project_to_detail_out, submit_image_prompt_optimize, submit_video_prompt_optimize, update_shot_replicate_image_prompt, update_shot_replicate_material_input, update_shot_replicate_video_prompt_schema, ) from app.services.shot_replicate_taskset_service import ( create_custom_segment, create_segments_by_ai, create_task_set, get_segment_for_user, list_segments, list_task_sets, segment_detail, task_set_detail, ) from app.services.module_generation_log_service import log_module_error, log_module_event_file from app.services.module_async_recovery_service import ( TASK_SHOT_IMAGE_PROMPT, TASK_SHOT_VIDEO_PROMPT, register_module_step_task, register_shot_segment_analysis_task, register_shot_split_task, register_shot_task_set_analysis_task, ) from app.tasks.celery_app import celery_app MODULE = ModuleCodeEnum.SHOT_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="/shot-replications", tags=["shot-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") or local_values.get("task_set_id") or local_values.get("task_set_id_value") step_id = local_values.get("step_id_value") or local_values.get("step_id") or local_values.get("segment_id") or local_values.get("segment_id_value") req = local_values.get("req") detail = {"api": local_values.get("__name__"), "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, ) def _ensure_celery_enabled(*, current_user: User | None = None, project_id: str | None = None, step_id: str | None = None) -> None: if celery_app is not None: return message = "Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker" _log_api_error( event_type="CELERY_DISABLED", current_user=current_user, project_id=project_id, step_id=step_id, message=message, detail={"api": "shot_replicate"}, ) raise HTTPException(status_code=503, detail=message) async def _reload_project_detail(db: AsyncSession, current_user: User, project_id: str) -> ShotReplicateTaskDetailOut: 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: if step_id: try: await mark_shot_replicate_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=ShotReplicateSpecOut, summary="查询拆镜复刻模块状态枚举和步骤 JSON 结构说明", ) async def get_spec(): return ShotReplicateSpecOut() @router.post( "/task-sets", response_model=ShotTaskSetDetailOut, summary="创建拆镜总任务集并异步分析原视频", ) async def create_shot_task_set( req: ShotTaskSetCreate = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: task_set = await create_task_set(db, current_user=current_user, req=req) task_set_id = task_set.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}") try: from app.tasks.shot_replicate_tasks import analyze_original_video await register_shot_task_set_analysis_task(task_set_id) analyze_original_video.apply_async(args=[task_set_id], queue="gen_chatapi_create", countdown=0) except Exception as exc: _log_api_error( event_type="CELERY_DISPATCH_FAILED", current_user=current_user, project_id=task_set_id, message=f"拆镜分析任务投递失败: {exc}", detail={"task_set_id": task_set_id, "task": "analyze_original_video"}, exc=exc, ) raise HTTPException(status_code=503, detail=f"拆镜分析任务投递失败: {exc}") return await task_set_detail(db, current_user=_user_context(current_user), task_set_id=task_set_id) @router.get( "/task-sets", response_model=ShotTaskSetListOut, summary="查询拆镜总任务集列表", ) async def list_shot_task_sets( status: str | None = Query(None, description="总任务状态，见 ShotTaskSetStatusEnum"), analysis_status: str | None = Query(None, description="分析状态，见 ShotAnalysisStatusEnum"), split_status: str | None = Query(None, description="拆镜状态，见 ShotSplitStatusEnum"), keyword: str | None = Query(None, description="标题/内容关键词"), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): return await list_task_sets( db, current_user=current_user, status=status, analysis_status=analysis_status, split_status=split_status, keyword=keyword, page=page, page_size=page_size, ) @router.get( "/task-sets/{task_set_id}", response_model=ShotTaskSetDetailOut, summary="获取拆镜总任务集详情", ) async def get_shot_task_set( task_set_id: str = Path(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): return await task_set_detail(db, current_user=_user_context(current_user), task_set_id=task_set_id) @router.post( "/task-sets/{task_set_id}/split-by-ai", response_model=ShotSplitByAIOut, summary="按 AI 建议方案异步拆镜", ) async def split_by_ai( task_set_id: str = Path(...), req: ShotSplitByAIRequest = Body(default_factory=ShotSplitByAIRequest), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: out = await create_segments_by_ai(db, current_user=current_user, task_set_id=task_set_id, req=req) segment_ids = [item.id for item in out.segments] 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}") from app.tasks.shot_replicate_tasks import split_one_segment for segment_id in segment_ids: await register_shot_split_task(segment_id, task_set_id=task_set_id) split_one_segment.apply_async(args=[segment_id], queue="gen_result_download", countdown=0) return out @router.post( "/task-sets/{task_set_id}/split-custom", response_model=ShotSplitCustomOut, summary="按用户自定义开始/结束秒异步拆单条片段", ) async def split_custom( task_set_id: str = Path(...), req: ShotSplitCustomRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: out = await create_custom_segment(db, current_user=current_user, task_set_id=task_set_id, req=req) segment_id = out.segment.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.shot_replicate_tasks import split_one_segment await register_shot_split_task(segment_id, task_set_id=task_set_id) split_one_segment.apply_async(args=[segment_id], queue="gen_result_download", countdown=0) return out @router.get( "/task-sets/{task_set_id}/segments", response_model=ShotSegmentListOut, summary="查询拆镜片段列表", ) async def list_task_set_segments( task_set_id: str = Path(...), source_mode: str | None = Query(None, description="ai_suggestion/custom"), split_status: str | None = Query(None, description="拆镜状态"), analysis_status: str | None = Query(None, description="片段分析状态"), replicate_status: str | None = Query(None, description="复刻状态"), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): return await list_segments( db, current_user=current_user, task_set_id=task_set_id, source_mode=source_mode, split_status=split_status, analysis_status=analysis_status, replicate_status=replicate_status, page=page, page_size=page_size, ) @router.get( "/segments/{segment_id}", response_model=ShotSegmentDetailOut, summary="获取拆镜片段详情", ) async def get_segment( segment_id: str = Path(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): return await segment_detail(db, current_user=current_user, segment_id=segment_id) @router.post( "/segments/{segment_id}/replication-projects", response_model=ShotReplicateActionOut, summary="将拆镜片段创建为拆镜复刻项目", ) async def create_replication_project_from_segment( segment_id: str = Path(...), req: ShotSegmentReplicationCreateRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): try: segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) project = await create_shot_replicate_project_from_segment(db, current_user=current_user, segment=segment, req=req) project_id = 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 ShotReplicateActionOut( message="已从拆镜片段创建复刻项目，素材视频已锁定", project_id=project_id, step_id=None, detail=await _reload_project_detail(db, current_user, project_id), ) @router.get( "/projects/{project_id}", response_model=ShotReplicateTaskDetailOut, summary="获取拆镜复刻项目详情", ) async def get_project( project_id: str = Path(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): return await _reload_project_detail(db, current_user, project_id) @router.put( "/projects/{project_id}/material", response_model=ShotReplicateActionOut, summary="修改拆镜复刻素材信息，素材视频不允许修改", ) async def update_material( project_id: str = Path(...), req: ShotReplicateMaterialUpdateRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): try: project_id_value, step_id_value = await update_shot_replicate_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 ShotReplicateActionOut(message="素材输入已修改，素材视频保持锁定", project_id=project_id_value, step_id=step_id_value, detail=await _reload_project_detail(db, current_user, project_id_value)) @router.put( "/projects/{project_id}/steps/{step_id}/image-prompt", response_model=ShotReplicateActionOut, summary="直接修改图片 AI 优化提词", ) async def update_image_prompt( project_id: str = Path(...), step_id: str = Path(...), req: ShotReplicateImagePromptUpdateRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): try: project, step = await update_shot_replicate_image_prompt(db, current_user=current_user, project_id=project_id, step_id=step_id, req=req) project_id_value, step_id_value = project.id, 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 ShotReplicateActionOut(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( "/projects/{project_id}/steps/{step_id}/video-prompt-schema", response_model=ShotReplicateActionOut, summary="修改视频 AI 提词 JSON schema", ) async def update_video_prompt_schema( project_id: str = Path(...), step_id: str = Path(...), req: ShotReplicateVideoPromptSchemaUpdateRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): try: project, step = await update_shot_replicate_video_prompt_schema(db, current_user=current_user, project_id=project_id, step_id=step_id, req=req) project_id_value, step_id_value = project.id, 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 ShotReplicateActionOut(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( "/projects/{project_id}/generate-image-prompt", response_model=ShotReplicateActionOut, summary="生成图片 AI 提词", ) async def generate_image_prompt( project_id: str = Path(...), req: ShotReplicateGenerateImagePromptRequest = Body(default_factory=ShotReplicateGenerateImagePromptRequest), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: project, step = await submit_image_prompt_optimize(db, current_user=current_user, project_id=project_id, req=req) project_id_value, step_id_value = project.id, 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}") try: from app.tasks.shot_replicate_flow_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=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value, task_name=TASK_SHOT_IMAGE_PROMPT, ) 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"图片 AI 提词任务投递失败: {exc}") return ShotReplicateActionOut(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( "/projects/{project_id}/generate-image", response_model=ShotReplicateActionOut, summary="根据图片 AI 提词生成图片", ) async def generate_image( project_id: str = Path(...), req: ShotReplicateGenerateImageRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: project, step, chat_task = await generate_image_from_prompt(db, current_user=current_user, project_id=project_id, req=req) project_id_value, step_id_value, chat_task_id_value = project.id, step.id, 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}") try: from app.tasks.generation_create_tasks import chatapi_create_generation_task chatapi_create_generation_task.apply_async(args=[chat_task_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 ShotReplicateActionOut(message="图片生成任务已提交", project_id=project_id_value, step_id=step_id_value, detail=await _reload_project_detail(db, current_user, project_id_value)) @router.post( "/projects/{project_id}/generate-video-prompt", response_model=ShotReplicateActionOut, summary="生成视频 AI 提词 JSON schema", ) async def generate_video_prompt( project_id: str = Path(...), req: ShotReplicateGenerateVideoPromptRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: project, step = await submit_video_prompt_optimize(db, current_user=current_user, project_id=project_id, req=req) project_id_value, step_id_value = project.id, 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}") try: from app.tasks.shot_replicate_flow_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=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value, task_name=TASK_SHOT_VIDEO_PROMPT, ) 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"视频 AI 提词任务投递失败: {exc}") return ShotReplicateActionOut(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( "/projects/{project_id}/generate-video", response_model=ShotReplicateActionOut, summary="根据视频 AI 提词生成视频", ) async def generate_video( project_id: str = Path(...), req: ShotReplicateGenerateVideoRequest = Body(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id")) try: project, step, chat_task = await generate_video_from_prompt(db, current_user=current_user, project_id=project_id, req=req) project_id_value, step_id_value, chat_task_id_value = project.id, step.id, 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}") try: from app.tasks.generation_create_tasks import chatapi_create_generation_task chatapi_create_generation_task.apply_async(args=[chat_task_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 ShotReplicateActionOut(message="视频生成任务已提交", project_id=project_id_value, step_id=step_id_value, detail=await _reload_project_detail(db, current_user, project_id_value)) @router.delete( "/projects/{project_id}", response_model=ShotReplicateDeleteOut, summary="软删除拆镜复刻项目", ) async def delete_project( project_id: str = Path(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): try: out = await delete_shot_replicate_project(db, current_user=current_user, project_id=project_id) await db.commit() return out 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}")