From 2d419b171e12f4adab4b34f753dfb10d85e5fb26 Mon Sep 17 00:00:00 2001 From: GinHa <15201596918@163.com> Date: Tue, 7 Jul 2026 19:52:12 +0800 Subject: [PATCH] =?UTF-8?q?=E7=88=86=E6=AC=BE=E5=BC=80=E5=A4=B4=E5=A4=8D?= =?UTF-8?q?=E5=88=BB/=E6=8B=86=E9=95=9C=E5=A4=8D=E5=88=BB=E8=BF=BD?= =?UTF-8?q?=E5=8A=A0=E6=97=A5=E5=BF=97|=E6=8B=86=E9=95=9C=E5=A4=8D?= =?UTF-8?q?=E5=88=BB=E8=BF=BD=E5=8A=A0=E8=A7=86=E9=A2=91AI=E5=88=86?= =?UTF-8?q?=E6=9E=90API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/api/v1/hot_opening_replicate.py | 16 +- video-gen-api/app/api/v1/shot_replicate.py | 164 +++++++++++- video-gen-api/app/config.py | 2 +- video-gen-api/app/enums/common.py | 24 ++ .../app/enums/hot_opening_replicate.py | 29 ++ video-gen-api/app/enums/shot_replicate.py | 51 ++++ video-gen-api/app/schemas/shot_replicate.py | 22 ++ .../services/hot_opening_replicate_service.py | 4 + .../hot_opening_video_prompt_service.py | 250 +++++++++++++++++- .../services/module_async_recovery_service.py | 16 +- .../services/module_generation_log_service.py | 42 ++- .../app/services/operation_log_service.py | 228 +++++++++++++--- .../services/shot_replicate_flow_service.py | 4 + .../shot_replicate_taskset_service.py | 123 +++++++++ .../services/shot_video_analysis_service.py | 248 ++++++++++++++++- .../app/tasks/shot_replicate_tasks.py | 21 +- 16 files changed, 1153 insertions(+), 91 deletions(-) diff --git a/video-gen-api/app/api/v1/hot_opening_replicate.py b/video-gen-api/app/api/v1/hot_opening_replicate.py index 0550b97b..e0258556 100644 --- a/video-gen-api/app/api/v1/hot_opening_replicate.py +++ b/video-gen-api/app/api/v1/hot_opening_replicate.py @@ -10,7 +10,7 @@ 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 -from app.enums.hot_opening_replicate import HotOpeningStepCodeEnum, ModuleCodeEnum +from app.enums.hot_opening_replicate import HotOpeningLogEventEnum, HotOpeningStepCodeEnum, ModuleCodeEnum from app.schemas.hot_opening_replicate import ( HotOpeningActionOut, HotOpeningDeleteOut, @@ -125,7 +125,7 @@ def _log_api_exception_from_locals(exc: BaseException, local_values: dict, messa 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", + 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, @@ -172,7 +172,7 @@ async def _mark_dispatch_failed_and_raise( except Exception as exc: await db.rollback() _log_api_error( - event_type="CELERY_DISPATCH_MARK_FAILED", + event_type=HotOpeningLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value, current_user=current_user, project_id=project_id, step_id=step_id, @@ -182,7 +182,7 @@ async def _mark_dispatch_failed_and_raise( ) log_module_error( module=MODULE, - event_type="CELERY_DISPATCH_FAILED", + event_type=HotOpeningLogEventEnum.CELERY_DISPATCH_FAILED.value, project_id=project_id, step_id=step_id, user_id=_safe_user_id(current_user), @@ -439,7 +439,7 @@ async def generate_image_prompt( _ = req if celery_app is None: _log_api_error( - event_type="CELERY_DISABLED", + event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value, current_user=current_user, project_id=project_id, step_id=step_id, @@ -509,7 +509,7 @@ async def generate_image( ): if celery_app is None: _log_api_error( - event_type="CELERY_DISABLED", + event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value, current_user=current_user, project_id=project_id, step_id=step_id, @@ -575,7 +575,7 @@ async def generate_video_prompt( ): if celery_app is None: _log_api_error( - event_type="CELERY_DISABLED", + event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value, current_user=current_user, project_id=project_id, step_id=step_id, @@ -646,7 +646,7 @@ async def generate_video( ): if celery_app is None: _log_api_error( - event_type="CELERY_DISABLED", + event_type=HotOpeningLogEventEnum.CELERY_DISABLED.value, current_user=current_user, project_id=project_id, step_id=step_id, diff --git a/video-gen-api/app/api/v1/shot_replicate.py b/video-gen-api/app/api/v1/shot_replicate.py index b64019b2..c809fda2 100644 --- a/video-gen-api/app/api/v1/shot_replicate.py +++ b/video-gen-api/app/api/v1/shot_replicate.py @@ -12,6 +12,7 @@ from app.models.user import User from app.enums.shot_replicate import ( ModuleCodeEnum, ShotAnalysisStatusEnum, + ShotReplicateLogEventEnum, ShotReplicateStepCodeEnum, ShotSegmentAnalysisStatusEnum, ShotSegmentReplicateStatusEnum, @@ -27,6 +28,8 @@ from app.schemas.shot_replicate import ( ShotReplicateGenerateVideoPromptRequest, ShotReplicateGenerateVideoRequest, ShotReplicateImagePromptUpdateRequest, + ShotReanalyzeOut, + ShotReanalyzeRequest, ShotReplicateMaterialUpdateRequest, ShotReplicateSpecOut, ShotReplicateTaskDetailOut, @@ -65,6 +68,8 @@ from app.services.shot_replicate_taskset_service import ( get_segment_for_user, list_segments, list_task_sets, + prepare_reanalyze_segment, + prepare_reanalyze_task_set, segment_detail, task_set_detail, ) @@ -152,7 +157,7 @@ def _log_api_exception_from_locals(exc: BaseException, local_values: dict, messa 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", + event_type=ShotReplicateLogEventEnum.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, @@ -168,7 +173,7 @@ def _ensure_celery_enabled(*, current_user: User | None = None, project_id: str return message = "Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker" _log_api_error( - event_type="CELERY_DISABLED", + event_type=ShotReplicateLogEventEnum.CELERY_DISABLED.value, current_user=current_user, project_id=project_id, step_id=step_id, @@ -209,7 +214,7 @@ async def _mark_dispatch_failed_and_raise( except Exception as exc: await db.rollback() _log_api_error( - event_type="CELERY_DISPATCH_MARK_FAILED", + event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value, current_user=current_user, project_id=project_id, step_id=step_id, @@ -219,7 +224,7 @@ async def _mark_dispatch_failed_and_raise( ) log_module_error( module=MODULE, - event_type="CELERY_DISPATCH_FAILED", + event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, project_id=project_id, step_id=step_id, user_id=_safe_user_id(current_user), @@ -275,7 +280,7 @@ async def create_shot_task_set( 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", + event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, current_user=current_user, project_id=task_set_id, message=f"拆镜分析任务投递失败: {exc}", @@ -345,6 +350,79 @@ async def get_shot_task_set( 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}/reanalyze", + response_model=ShotReanalyzeOut, + summary="重新投递原视频 AI 分析任务", + description="用于处理原视频分析失败或待处理的异常数据；重置分析状态后重新投递 analyze_original_video。", +) +async def reanalyze_task_set( + task_set_id: str = Path(..., description="拆镜总任务集ID，即 shot_replicate_task_sets.id"), + req: ShotReanalyzeRequest = Body(default_factory=ShotReanalyzeRequest, description="再次分析参数"), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + _ensure_celery_enabled(current_user=current_user, project_id=task_set_id) + try: + out = await prepare_reanalyze_task_set( + db, + current_user=current_user, + task_set_id=task_set_id, + force=req.force, + reason=req.reason, + ) + await db.commit() + except HTTPException as exc: + await db.rollback() + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value, + project_id=task_set_id, + user_id=_safe_user_id(current_user), + message="原视频再次分析请求被拒绝", + detail={"task_set_id": task_set_id, "request": req.model_dump(), "http_status": exc.status_code, "detail": exc.detail}, + error=str(exc.detail), + ) + raise + except Exception as exc: + await db.rollback() + _log_api_error( + event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_FAILED.value, + current_user=current_user, + project_id=task_set_id, + message=f"原视频再次分析状态重置失败: {exc}", + detail={"task_set_id": task_set_id, "request": req.model_dump()}, + exc=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) + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_SUBMITTED.value, + project_id=task_set_id, + user_id=_safe_user_id(current_user), + message="原视频再次分析任务已投递", + detail={"task_set_id": task_set_id, "task": "analyze_original_video", "request": req.model_dump()}, + ) + except Exception as exc: + _log_api_error( + event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, + 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}") + out.message = "原视频再次分析任务已提交" + return out + + @router.post( "/task-sets/{task_set_id}/split-by-ai", response_model=ShotSplitByAIOut, @@ -457,6 +535,82 @@ async def get_segment( return await segment_detail(db, current_user=current_user, segment_id=segment_id) +@router.post( + "/segments/{segment_id}/reanalyze", + response_model=ShotReanalyzeOut, + summary="重新投递切片视频 AI 分析任务", + description="用于处理自定义切片视频分析失败或待处理的异常数据；重置分析状态后重新投递 analyze_custom_segment_video。", +) +async def reanalyze_segment( + segment_id: str = Path(..., description="拆镜片段ID，即 shot_replicate_segments.id"), + req: ShotReanalyzeRequest = Body(default_factory=ShotReanalyzeRequest, description="再次分析参数"), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + _ensure_celery_enabled(current_user=current_user, step_id=segment_id) + try: + out = await prepare_reanalyze_segment( + db, + current_user=current_user, + segment_id=segment_id, + force=req.force, + reason=req.reason, + ) + task_set_id = out.task_set_id + await db.commit() + except HTTPException as exc: + await db.rollback() + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value, + step_id=segment_id, + user_id=_safe_user_id(current_user), + message="切片视频再次分析请求被拒绝", + detail={"segment_id": segment_id, "request": req.model_dump(), "http_status": exc.status_code, "detail": exc.detail}, + error=str(exc.detail), + ) + raise + except Exception as exc: + await db.rollback() + _log_api_error( + event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_FAILED.value, + current_user=current_user, + step_id=segment_id, + message=f"切片视频再次分析状态重置失败: {exc}", + detail={"segment_id": segment_id, "request": req.model_dump()}, + exc=exc, + ) + raise HTTPException(status_code=500, detail=f"切片视频再次分析状态重置失败: {exc}") + + try: + from app.tasks.shot_replicate_tasks import analyze_custom_segment_video + + await register_shot_segment_analysis_task(segment_id, task_set_id=task_set_id) + analyze_custom_segment_video.apply_async(args=[segment_id], queue="gen_chatapi_create", countdown=0) + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_SUBMITTED.value, + project_id=task_set_id, + step_id=segment_id, + user_id=_safe_user_id(current_user), + message="切片视频再次分析任务已投递", + detail={"segment_id": segment_id, "task_set_id": task_set_id, "task": "analyze_custom_segment_video", "request": req.model_dump()}, + ) + except Exception as exc: + _log_api_error( + event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, + current_user=current_user, + project_id=task_set_id, + step_id=segment_id, + message=f"切片视频再次分析任务投递失败: {exc}", + detail={"segment_id": segment_id, "task_set_id": task_set_id, "task": "analyze_custom_segment_video"}, + exc=exc, + ) + raise HTTPException(status_code=503, detail=f"切片视频再次分析任务投递失败: {exc}") + out.message = "切片视频再次分析任务已提交" + return out + + @router.delete( "/segments/{segment_id}", response_model=ShotSegmentDeleteOut, diff --git a/video-gen-api/app/config.py b/video-gen-api/app/config.py index cdcef903..52d9779b 100644 --- a/video-gen-api/app/config.py +++ b/video-gen-api/app/config.py @@ -228,7 +228,7 @@ class Settings(BaseSettings): # 拆镜复刻配置。 # 原始上传视频和拆镜片段都属于 uploads 素材域；只有 generate 生成结果走 token 验签。 - SHOT_ANALYSIS_TIMEOUT_SECONDS: int = 180 + SHOT_ANALYSIS_TIMEOUT_SECONDS: int = 600 SHOT_ANALYSIS_TEMPERATURE: float = 0.1 SHOT_ANALYSIS_MAX_TOKENS: int = 5000 SHOT_ANALYSIS_VIDEO_FPS: float = 1.0 diff --git a/video-gen-api/app/enums/common.py b/video-gen-api/app/enums/common.py index 251e44b5..4457a5b1 100644 --- a/video-gen-api/app/enums/common.py +++ b/video-gen-api/app/enums/common.py @@ -3,6 +3,30 @@ from __future__ import annotations from enum import StrEnum + + +class LogEventStatusEnum(StrEnum): + """通用日志事件状态。""" + + PENDING = "pending" + STARTED = "started" + SUBMITTED = "submitted" + SUCCESS = "success" + FAILED = "failed" + REJECTED = "rejected" + SKIPPED = "skipped" + WARNING = "warning" + + +class LogSourceEnum(StrEnum): + """通用日志来源。""" + + API = "api" + SERVICE = "service" + CELERY = "celery" + RECOVERY = "recovery" + REMOTE_API = "remote_api" + class ModuleProjectStatusEnum(StrEnum): """通用模块项目状态。""" diff --git a/video-gen-api/app/enums/hot_opening_replicate.py b/video-gen-api/app/enums/hot_opening_replicate.py index a43d11ba..6d167827 100644 --- a/video-gen-api/app/enums/hot_opening_replicate.py +++ b/video-gen-api/app/enums/hot_opening_replicate.py @@ -30,3 +30,32 @@ class HotOpeningStepIOSchemaVersionEnum(StrEnum): """爆款开头复刻子任务 input_json/output_json 结构版本。""" V1 = "hot_opening_step_io_v1" + + +class HotOpeningLogEventEnum(StrEnum): + """爆款开头复刻模块业务日志事件。""" + + API_REQUEST_RECEIVED = "HOT_OPENING_API_REQUEST_RECEIVED" + API_REQUEST_SUBMITTED = "HOT_OPENING_API_REQUEST_SUBMITTED" + API_REQUEST_REJECTED = "HOT_OPENING_API_REQUEST_REJECTED" + API_REQUEST_FAILED = "HOT_OPENING_API_REQUEST_FAILED" + CELERY_DISABLED = "HOT_OPENING_CELERY_DISABLED" + CELERY_DISPATCH_FAILED = "HOT_OPENING_CELERY_DISPATCH_FAILED" + CELERY_DISPATCH_MARK_FAILED = "HOT_OPENING_CELERY_DISPATCH_MARK_FAILED" + + IMAGE_PROMPT_SUBMITTED = "HOT_OPENING_IMAGE_PROMPT_SUBMITTED" + IMAGE_GENERATE_SUBMITTED = "HOT_OPENING_IMAGE_GENERATE_SUBMITTED" + VIDEO_PROMPT_SUBMITTED = "HOT_OPENING_VIDEO_PROMPT_SUBMITTED" + VIDEO_GENERATE_SUBMITTED = "HOT_OPENING_VIDEO_GENERATE_SUBMITTED" + + VIDEO_PROMPT_REMOTE_API_STARTED = "HOT_OPENING_VIDEO_PROMPT_REMOTE_API_STARTED" + VIDEO_PROMPT_REMOTE_API_SUCCESS = "HOT_OPENING_VIDEO_PROMPT_REMOTE_API_SUCCESS" + VIDEO_PROMPT_REMOTE_API_FAILED = "HOT_OPENING_VIDEO_PROMPT_REMOTE_API_FAILED" + VIDEO_PROMPT_RESPONSE_PARSE_FAILED = "HOT_OPENING_VIDEO_PROMPT_RESPONSE_PARSE_FAILED" + VIDEO_PROMPT_RESPONSE_EMPTY = "HOT_OPENING_VIDEO_PROMPT_RESPONSE_EMPTY" + + +class HotOpeningRemoteActionEnum(StrEnum): + """爆款开头复刻远程模型动作。""" + + VIDEO_PROMPT_OPTIMIZE = "video_prompt_optimize" diff --git a/video-gen-api/app/enums/shot_replicate.py b/video-gen-api/app/enums/shot_replicate.py index 2c48914c..868f25cc 100644 --- a/video-gen-api/app/enums/shot_replicate.py +++ b/video-gen-api/app/enums/shot_replicate.py @@ -90,3 +90,54 @@ class ShotSegmentReplicateStatusEnum(StrEnum): PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" + + +class ShotReplicateLogEventEnum(StrEnum): + """拆镜复刻模块业务日志事件。""" + + API_REQUEST_RECEIVED = "SHOT_API_REQUEST_RECEIVED" + API_REQUEST_SUBMITTED = "SHOT_API_REQUEST_SUBMITTED" + API_REQUEST_REJECTED = "SHOT_API_REQUEST_REJECTED" + API_REQUEST_FAILED = "SHOT_API_REQUEST_FAILED" + CELERY_DISABLED = "SHOT_CELERY_DISABLED" + CELERY_DISPATCH_FAILED = "SHOT_CELERY_DISPATCH_FAILED" + CELERY_DISPATCH_MARK_FAILED = "SHOT_CELERY_DISPATCH_MARK_FAILED" + + TASK_SET_CREATED = "SHOT_TASK_SET_CREATED" + TASK_SET_REANALYZE_RECEIVED = "SHOT_TASK_SET_REANALYZE_RECEIVED" + TASK_SET_REANALYZE_SUBMITTED = "SHOT_TASK_SET_REANALYZE_SUBMITTED" + TASK_SET_REANALYZE_REJECTED = "SHOT_TASK_SET_REANALYZE_REJECTED" + TASK_SET_REANALYZE_FAILED = "SHOT_TASK_SET_REANALYZE_FAILED" + + ANALYSIS_STARTED = "SHOT_ANALYSIS_STARTED" + ANALYSIS_SUCCESS = "SHOT_ANALYSIS_SUCCESS" + ANALYSIS_FAILED = "SHOT_ANALYSIS_FAILED" + ANALYSIS_REMOTE_API_STARTED = "SHOT_ANALYSIS_REMOTE_API_STARTED" + ANALYSIS_REMOTE_API_SUCCESS = "SHOT_ANALYSIS_REMOTE_API_SUCCESS" + ANALYSIS_REMOTE_API_FAILED = "SHOT_ANALYSIS_REMOTE_API_FAILED" + ANALYSIS_RESPONSE_PARSE_FAILED = "SHOT_ANALYSIS_RESPONSE_PARSE_FAILED" + ANALYSIS_RESPONSE_EMPTY = "SHOT_ANALYSIS_RESPONSE_EMPTY" + + SEGMENT_REANALYZE_RECEIVED = "SHOT_SEGMENT_REANALYZE_RECEIVED" + SEGMENT_REANALYZE_SUBMITTED = "SHOT_SEGMENT_REANALYZE_SUBMITTED" + SEGMENT_REANALYZE_REJECTED = "SHOT_SEGMENT_REANALYZE_REJECTED" + SEGMENT_REANALYZE_FAILED = "SHOT_SEGMENT_REANALYZE_FAILED" + SEGMENT_ANALYSIS_STARTED = "SHOT_SEGMENT_ANALYSIS_STARTED" + SEGMENT_ANALYSIS_SUCCESS = "SHOT_SEGMENT_ANALYSIS_SUCCESS" + SEGMENT_ANALYSIS_FAILED = "SHOT_SEGMENT_ANALYSIS_FAILED" + SEGMENT_ANALYSIS_REMOTE_API_STARTED = "SHOT_SEGMENT_ANALYSIS_REMOTE_API_STARTED" + SEGMENT_ANALYSIS_REMOTE_API_SUCCESS = "SHOT_SEGMENT_ANALYSIS_REMOTE_API_SUCCESS" + SEGMENT_ANALYSIS_REMOTE_API_FAILED = "SHOT_SEGMENT_ANALYSIS_REMOTE_API_FAILED" + + SPLIT_STATUS_CHANGED = "SHOT_SPLIT_STATUS_CHANGED" + SPLIT_BY_AI_SUBMITTED = "SHOT_SPLIT_BY_AI_SUBMITTED" + SPLIT_CUSTOM_SUBMITTED = "SHOT_SPLIT_CUSTOM_SUBMITTED" + SEGMENT_DELETED = "SHOT_SEGMENT_DELETED" + + +class ShotReplicateRemoteActionEnum(StrEnum): + """拆镜复刻远程模型动作。""" + + ANALYZE_ORIGINAL_VIDEO = "analyze_original_video" + ANALYZE_CUSTOM_SEGMENT_VIDEO = "analyze_custom_segment_video" + VIDEO_PROMPT_OPTIMIZE = "video_prompt_optimize" diff --git a/video-gen-api/app/schemas/shot_replicate.py b/video-gen-api/app/schemas/shot_replicate.py index 86c809f2..3ac1ab9c 100644 --- a/video-gen-api/app/schemas/shot_replicate.py +++ b/video-gen-api/app/schemas/shot_replicate.py @@ -700,6 +700,28 @@ class ShotSegmentListOut(BaseModel): items: list[ShotSegmentOut] = Field(default_factory=list, description="拆镜片段列表") + + +class ShotReanalyzeRequest(BaseModel): + force: bool = Field(False, description="是否强制重跑；默认 false。当前仅允许失败/待处理数据重试，已完成数据不建议强制覆盖") + reason: str | None = Field(None, max_length=200, description="再次分析原因，会写入模块日志") + + @field_validator("reason", mode="before") + @classmethod + def _strip_reason(cls, value: str | None) -> str | None: + if value is None: + return None + value = str(value).strip() + return value or None + + +class ShotReanalyzeOut(BaseModel): + message: str = Field(..., description="操作结果提示") + task_set_id: str | None = Field(None, description="拆镜总任务集ID") + segment_id: str | None = Field(None, description="拆镜片段ID") + analysis_status: str = Field(..., description="重置后的分析状态") + celery_task_name: str = Field(..., description="已投递或待投递的 Celery 任务名") + class ShotSplitByAIOut(BaseModel): task_set_id: str = Field(..., description="拆镜总任务集ID") status: str = Field(..., description="总任务状态：pending_analysis/analyzing/analysis_completed/analysis_failed/splitting/split_completed/partial_failed/failed/deleted") diff --git a/video-gen-api/app/services/hot_opening_replicate_service.py b/video-gen-api/app/services/hot_opening_replicate_service.py index 2a76bb99..a79e4d24 100644 --- a/video-gen-api/app/services/hot_opening_replicate_service.py +++ b/video-gen-api/app/services/hot_opening_replicate_service.py @@ -1183,6 +1183,10 @@ async def run_video_prompt_optimize(db: AsyncSession, *, project_id: str, step_i video_config=video_config, target_platform=target_platform, schema_config_snapshot=schema_config_snapshot, + module=project.module, + project_id=project.id, + step_id=step.id, + trace_id=f"hot-video-prompt:{step.id}", ) billing = await charge_module_prompt_usage( db, diff --git a/video-gen-api/app/services/hot_opening_video_prompt_service.py b/video-gen-api/app/services/hot_opening_video_prompt_service.py index 1a384a41..df1dd923 100644 --- a/video-gen-api/app/services/hot_opening_video_prompt_service.py +++ b/video-gen-api/app/services/hot_opening_video_prompt_service.py @@ -10,6 +10,10 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings +from app.enums.common import LogEventStatusEnum, LogSourceEnum +from app.enums.hot_opening_replicate import HotOpeningLogEventEnum, HotOpeningRemoteActionEnum, ModuleCodeEnum as HotModuleCodeEnum +from app.enums.shot_replicate import ModuleCodeEnum as ShotModuleCodeEnum, ShotReplicateLogEventEnum, ShotReplicateRemoteActionEnum +from app.services.operation_log_service import log_ai_model_event from app.enums.common import ( VIDEO_SCHEMA_CONFIG_DATABASE_SOURCE, VIDEO_SCHEMA_CONFIG_DEFAULT_SOURCE, @@ -1415,6 +1419,100 @@ def _mock_result(video_config: dict[str, Any], target_platform: str) -> dict[str return schema + + +def _safe_response_json(response: httpx.Response | None) -> Any: + if response is None: + return None + try: + return response.json() + except Exception: + return {"raw_text": response.text} + + +def _extract_remote_request_id(data: Any) -> str | None: + if isinstance(data, dict): + error = data.get("error") if isinstance(data.get("error"), dict) else {} + request_id = data.get("request_id") or error.get("request_id") + message = error.get("message") or data.get("message") + if not request_id and isinstance(message, str): + import re + match = re.search(r"Request id:\s*([a-zA-Z0-9_.:-]+)", message, flags=re.IGNORECASE) + if match: + request_id = match.group(1) + return str(request_id) if request_id else None + return None + + +def _video_prompt_remote_event(module: str, *, started: bool = False, success: bool = False, parse_failed: bool = False, empty: bool = False) -> tuple[str, str]: + if module == ShotModuleCodeEnum.SHOT_REPLICATE.value: + action = ShotReplicateRemoteActionEnum.VIDEO_PROMPT_OPTIMIZE.value + if started: + return ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_STARTED.value, action + if success: + return ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_SUCCESS.value, action + if parse_failed: + return ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_PARSE_FAILED.value, action + if empty: + return ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_EMPTY.value, action + return ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_FAILED.value, action + action = HotOpeningRemoteActionEnum.VIDEO_PROMPT_OPTIMIZE.value + if started: + return HotOpeningLogEventEnum.VIDEO_PROMPT_REMOTE_API_STARTED.value, action + if success: + return HotOpeningLogEventEnum.VIDEO_PROMPT_REMOTE_API_SUCCESS.value, action + if parse_failed: + return HotOpeningLogEventEnum.VIDEO_PROMPT_RESPONSE_PARSE_FAILED.value, action + if empty: + return HotOpeningLogEventEnum.VIDEO_PROMPT_RESPONSE_EMPTY.value, action + return HotOpeningLogEventEnum.VIDEO_PROMPT_REMOTE_API_FAILED.value, action + + +def _log_video_prompt_ai_event( + *, + module: str, + event_type: str, + action: str, + event_status: str, + config: ModelConfig, + trace_id: str | None, + user_id: str | None, + project_id: str | None, + step_id: str | None, + request_data: dict[str, Any] | None = None, + response_data: Any = None, + token_usage: dict[str, Any] | None = None, + http_status: int | None = None, + remote_request_id: str | None = None, + message: str | None = None, + error: str | None = None, + detail: dict[str, Any] | None = None, +) -> None: + log_ai_model_event( + event_type=event_type, + event_status=event_status, + source=LogSourceEnum.REMOTE_API.value, + module=module, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + remote_action=action, + remote_request_id=remote_request_id, + model_config_id=str(config.id), + model_config_name=config.name, + model_name=config.model_name, + provider=config.provider, + api_base=config.api_base, + http_status=http_status, + request=request_data, + response=response_data, + token_usage=token_usage, + message=message, + detail=detail, + error=error, + ) + async def _select_model_config(db: AsyncSession) -> ModelConfig | None: result = await db.execute(select(ModelConfig).where(ModelConfig.is_active == True).order_by(ModelConfig.priority.desc()).limit(1)) return result.scalar_one_or_none() @@ -1432,6 +1530,10 @@ async def optimize_hot_opening_video_prompt( video_config: dict[str, Any], target_platform: str = "抖音", schema_config_snapshot: Any | None = None, + module: str = HotModuleCodeEnum.HOT_OPENING_REPLICATE.value, + project_id: str | None = None, + step_id: str | None = None, + trace_id: str | None = None, ) -> tuple[dict[str, Any], str, dict[str, Any]]: duration = int(video_config["duration"]) references = [ @@ -1469,18 +1571,109 @@ async def optimize_hot_opening_video_prompt( "temperature": 0.15, "response_format": {"type": "json_object"}, } + api_url = f"{config.api_base.rstrip('/')}/chat/completions" + log_request_data = { + **request_data, + "messages": [{"role": "system", "content": build_system_prompt()}, log_user_message], + "model_config_id": config.id, + "model_config_name": config.name, + "provider": config.provider, + "module": module, + "project_id": project_id, + "step_id": step_id, + "material_video_url": material_video_url, + "generated_image_url": generated_image_url, + "target_platform": target_platform, + "api_url": api_url, + } + started_event, remote_action = _video_prompt_remote_event(module, started=True) + _log_video_prompt_ai_event( + module=module, + event_type=started_event, + action=remote_action, + event_status=LogEventStatusEnum.STARTED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + request_data=log_request_data, + message="视频提词模型请求开始", + ) - async with httpx.AsyncClient(timeout=int(settings.CHATAPI_REQUEST_TIMEOUT_SECONDS or 180)) as client: - response = await client.post( - f"{config.api_base.rstrip('/')}/chat/completions", - headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"}, - json=request_data, + try: + async with httpx.AsyncClient(timeout=int(settings.CHATAPI_REQUEST_TIMEOUT_SECONDS or 180)) as client: + response = await client.post( + api_url, + headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"}, + json=request_data, + ) + except Exception as exc: + failed_event, remote_action = _video_prompt_remote_event(module) + _log_video_prompt_ai_event( + module=module, + event_type=failed_event, + action=remote_action, + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + request_data=log_request_data, + message="视频提词模型请求异常", + error=str(exc), + detail={"exception_type": type(exc).__name__}, ) + raise + response_data = _safe_response_json(response) + remote_request_id = _extract_remote_request_id(response_data) if response.status_code >= 400: + failed_event, remote_action = _video_prompt_remote_event(module) + _log_video_prompt_ai_event( + module=module, + event_type=failed_event, + action=remote_action, + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + request_data=log_request_data, + response_data=response_data, + http_status=response.status_code, + remote_request_id=remote_request_id, + message="视频提词模型请求失败", + error=f"HTTP {response.status_code}: {response.text}", + ) raise RuntimeError(f"视频提词优化失败 HTTP {response.status_code}: {response.text}") - data = response.json() - content = data["choices"][0]["message"]["content"].strip() + try: + data = response.json() + content = data["choices"][0]["message"]["content"].strip() + if not content: + raise RuntimeError("视频提词模型响应 content 为空") + except Exception as exc: + parse_event, remote_action = _video_prompt_remote_event(module, empty="content 为空" in str(exc), parse_failed="content 为空" not in str(exc)) + _log_video_prompt_ai_event( + module=module, + event_type=parse_event, + action=remote_action, + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + request_data=log_request_data, + response_data=response_data, + http_status=response.status_code, + remote_request_id=remote_request_id, + message="视频提词模型响应解析失败", + error=str(exc), + ) + raise usage = data.get("usage", {}) or {} token_usage = { "input_tokens": int(usage.get("prompt_tokens") or 0), @@ -1508,8 +1701,47 @@ async def optimize_hot_opening_video_prompt( "model_name": config.model_name, }) - result = parse_model_json(content) - result = normalize_video_prompt_schema_from_ai(result, video_config, schema_config_snapshot) + try: + result = parse_model_json(content) + result = normalize_video_prompt_schema_from_ai(result, video_config, schema_config_snapshot) + except Exception as exc: + parse_event, remote_action = _video_prompt_remote_event(module, parse_failed=True) + _log_video_prompt_ai_event( + module=module, + event_type=parse_event, + action=remote_action, + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + request_data=log_request_data, + response_data=data, + http_status=response.status_code, + remote_request_id=remote_request_id, + message="视频提词业务 JSON 解析失败", + error=str(exc), + ) + raise + success_event, remote_action = _video_prompt_remote_event(module, success=True) + _log_video_prompt_ai_event( + module=module, + event_type=success_event, + action=remote_action, + event_status=LogEventStatusEnum.SUCCESS.value, + config=config, + trace_id=trace_id, + user_id=user_id, + project_id=project_id, + step_id=step_id, + request_data=log_request_data, + response_data=data, + token_usage=token_usage, + http_status=response.status_code, + remote_request_id=remote_request_id, + message="视频提词模型请求成功", + ) return result, build_final_video_prompt(result), token_usage def _build_file_url_or_data_uri(file_url: str) -> str: diff --git a/video-gen-api/app/services/module_async_recovery_service.py b/video-gen-api/app/services/module_async_recovery_service.py index cff60b3c..2eda39cb 100644 --- a/video-gen-api/app/services/module_async_recovery_service.py +++ b/video-gen-api/app/services/module_async_recovery_service.py @@ -98,6 +98,11 @@ def _lease_seconds() -> int: return max(1, int(settings.MODULE_ASYNC_LEASE_SECONDS or 600)) +def _shot_analysis_lease_seconds() -> int: + timeout = max(1, int(getattr(settings, "SHOT_ANALYSIS_TIMEOUT_SECONDS", 600) or 600)) + return max(_lease_seconds(), timeout + 120) + + def _queue_timeout_seconds() -> int: return max(1, int(settings.MODULE_ASYNC_QUEUE_TIMEOUT_SECONDS or 300)) @@ -227,7 +232,7 @@ async def register_shot_task_set_analysis_task(task_set_id: str) -> str: module=SHOT_MODULE, project_id=task_set_id, task_set_id=task_set_id, - check_after_seconds=_lease_seconds(), + check_after_seconds=_shot_analysis_lease_seconds(), ) @@ -242,7 +247,7 @@ async def register_shot_segment_analysis_task(segment_id: str, *, task_set_id: s project_id=task_set_id, task_set_id=task_set_id, segment_id=segment_id, - check_after_seconds=_lease_seconds(), + check_after_seconds=_shot_analysis_lease_seconds(), ) @@ -297,10 +302,15 @@ async def postpone_active_task( async def mark_active_started(*, object_type: str, object_id: str, reason: str = "started") -> None: + delay_seconds = _lease_seconds() + if object_type in {OBJECT_SHOT_TASK_SET_ANALYSIS, OBJECT_SHOT_SEGMENT_ANALYSIS}: + delay_seconds = _shot_analysis_lease_seconds() + elif object_type == OBJECT_SHOT_SPLIT_SEGMENT: + delay_seconds = max(_lease_seconds(), int(settings.SHOT_SPLIT_LEASE_SECONDS or 600)) await postpone_active_task( object_type=object_type, object_id=object_id, - delay_seconds=_lease_seconds(), + delay_seconds=delay_seconds, reason=reason, ) diff --git a/video-gen-api/app/services/module_generation_log_service.py b/video-gen-api/app/services/module_generation_log_service.py index 0567d1bf..badb9fb3 100644 --- a/video-gen-api/app/services/module_generation_log_service.py +++ b/video-gen-api/app/services/module_generation_log_service.py @@ -2,7 +2,11 @@ from __future__ import annotations from typing import Any -from app.services.operation_log_service import build_exception_detail, log_operation_error, log_operation_event +from app.enums.common import LogEventStatusEnum, LogSourceEnum +from app.services.operation_log_service import ( + build_exception_detail, + log_module_generation_event, +) def log_module_event_file( @@ -11,23 +15,28 @@ def log_module_event_file( event_type: str, project_id: str | None = None, step_id: str | None = None, + task_id: str | None = None, user_id: str | None = None, + trace_id: str | None = None, + source: str | None = None, + event_status: str | None = None, message: str | None = None, detail: dict[str, Any] | None = None, error: str | None = None, ) -> None: - log_operation_event( - domain="module_generation", + log_module_generation_event( module=module, event_type=event_type, project_id=project_id, step_id=step_id, + task_id=task_id, user_id=user_id, + trace_id=trace_id, message=message, detail=detail, error=error, - event_status="failed" if error else "success", - source="service", + event_status=event_status or (LogEventStatusEnum.FAILED.value if error else LogEventStatusEnum.SUCCESS.value), + source=source or LogSourceEnum.SERVICE.value, ) @@ -39,23 +48,24 @@ def log_module_prompt_event( user_id: str, module: str, prompt_type: str, + trace_id: str | None = None, request: dict[str, Any] | None = None, response: dict[str, Any] | None = None, token_usage: dict[str, Any] | None = None, error: str | None = None, ) -> None: - log_operation_event( - domain="module_generation", + log_module_generation_event( module=module, event_type=event_type, project_id=project_id, step_id=step_id, user_id=user_id, + trace_id=trace_id, message=f"模块 AI 请求：{prompt_type}", detail={"prompt_type": prompt_type, "request": request or {}, "response": response or {}, "token_usage": token_usage or {}}, error=error, - event_status="failed" if error else "success", - source="service", + event_status=LogEventStatusEnum.FAILED.value if error else LogEventStatusEnum.SUCCESS.value, + source=LogSourceEnum.SERVICE.value, ) @@ -65,22 +75,26 @@ def log_module_error( event_type: str, project_id: str | None = None, step_id: str | None = None, + task_id: str | None = None, user_id: str | None = None, + trace_id: str | None = None, + source: str | None = None, message: str | None = None, detail: dict[str, Any] | None = None, error: str | None = None, exc: BaseException | None = None, ) -> None: - log_operation_error( - domain="module_generation", + log_module_generation_event( module=module, event_type=event_type, project_id=project_id, step_id=step_id, + task_id=task_id, user_id=user_id, + trace_id=trace_id, message=message, - detail=detail, + detail=build_exception_detail(exc, detail), error=error if error is not None else (str(exc) if exc else None), - exc=exc, - source="service", + event_status=LogEventStatusEnum.FAILED.value, + source=source or LogSourceEnum.SERVICE.value, ) diff --git a/video-gen-api/app/services/operation_log_service.py b/video-gen-api/app/services/operation_log_service.py index 248c8beb..8e2c5056 100644 --- a/video-gen-api/app/services/operation_log_service.py +++ b/video-gen-api/app/services/operation_log_service.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import logging import os import re import traceback @@ -10,9 +11,14 @@ from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from app.services.log_config import LOG_DATE_FORMAT, LOG_DIR, is_enabled +logger = logging.getLogger("video_gen") + MAX_LOG_FIELD_LENGTH = 20000 MAX_TRACEBACK_LENGTH = 12000 -OPERATION_LOG_ROOT = os.path.join(os.path.dirname(LOG_DIR), "OperationLogs") +LOG_BASE_DIR = os.path.dirname(LOG_DIR) +OPERATION_LOG_ROOT = os.path.join(LOG_BASE_DIR, "OperationLogs") +MODULE_GENERATION_LOG_ROOT = os.path.join(LOG_BASE_DIR, "ModuleGeneration") +AI_MODEL_LOG_ROOT = LOG_DIR SENSITIVE_KEY_PATTERNS = ( "secret", "token", @@ -94,17 +100,69 @@ def build_exception_detail(exc: BaseException | None, extra: dict[str, Any] | No return detail -def _append_operation_log(domain: str, entry: dict[str, Any]) -> None: +def _append_json_log(root_dir: str, domain: str | None, entry: dict[str, Any]) -> None: if not is_enabled(): return try: - domain_dir = os.path.join(OPERATION_LOG_ROOT, _safe_name(domain, "default")) - os.makedirs(domain_dir, exist_ok=True) + target_dir = root_dir if domain is None else os.path.join(root_dir, _safe_name(domain, "default")) + os.makedirs(target_dir, exist_ok=True) today = datetime.now().strftime(LOG_DATE_FORMAT) - with open(os.path.join(domain_dir, f"{today}.log"), "a", encoding="utf-8") as f: + with open(os.path.join(target_dir, f"{today}.log"), "a", encoding="utf-8") as f: f.write(json.dumps(sanitize_log_value(entry), ensure_ascii=False, default=str) + "\n") - except Exception: - pass + except Exception as exc: + logger.warning("operation log write failed: root_dir=%s domain=%s error=%s", root_dir, domain, exc, exc_info=True) + + +def _append_operation_log(domain: str, entry: dict[str, Any]) -> None: + _append_json_log(OPERATION_LOG_ROOT, domain, entry) + + +def _base_entry( + *, + log_type: str, + domain: str, + event_type: str, + module: str | None = None, + event_status: str = "success", + source: str | None = None, + trace_id: str | None = None, + request_id: str | None = None, + user_id: str | None = None, + project_id: str | None = None, + session_id: str | None = None, + group_id: str | None = None, + asset_id: str | None = None, + task_id: str | None = None, + step_id: str | None = None, + remote_action: str | None = None, + remote_request_id: str | None = None, + message: str | None = None, + detail: dict[str, Any] | None = None, + error: str | None = None, +) -> dict[str, Any]: + return { + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "log_type": log_type, + "domain": domain, + "module": module or domain, + "event_type": event_type, + "event_status": event_status, + "source": source, + "trace_id": trace_id, + "request_id": request_id, + "user_id": user_id, + "project_id": project_id, + "session_id": session_id, + "group_id": group_id, + "asset_id": asset_id, + "task_id": task_id, + "step_id": step_id, + "remote_action": remote_action, + "remote_request_id": remote_request_id, + "message": message, + "detail": detail or {}, + "error": error, + } def log_operation_event( @@ -131,32 +189,142 @@ def log_operation_event( ) -> None: _append_operation_log( domain, - { - "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "log_type": "operation_event", - "domain": domain, - "module": module or domain, - "event_type": event_type, - "event_status": event_status, - "source": source, - "trace_id": trace_id, - "request_id": request_id, - "user_id": user_id, - "project_id": project_id, - "session_id": session_id, - "group_id": group_id, - "asset_id": asset_id, - "task_id": task_id, - "step_id": step_id, - "remote_action": remote_action, - "remote_request_id": remote_request_id, - "message": message, - "detail": detail or {}, - "error": error, - }, + _base_entry( + log_type="operation_event", + domain=domain, + module=module, + event_type=event_type, + event_status=event_status, + source=source, + trace_id=trace_id, + request_id=request_id, + user_id=user_id, + project_id=project_id, + session_id=session_id, + group_id=group_id, + asset_id=asset_id, + task_id=task_id, + step_id=step_id, + remote_action=remote_action, + remote_request_id=remote_request_id, + message=message, + detail=detail, + error=error, + ), ) +def log_module_generation_event( + *, + module: str, + event_type: str, + event_status: str = "success", + source: str | None = None, + trace_id: str | None = None, + request_id: str | None = None, + user_id: str | None = None, + project_id: str | None = None, + task_id: str | None = None, + step_id: str | None = None, + remote_action: str | None = None, + remote_request_id: str | None = None, + message: str | None = None, + detail: dict[str, Any] | None = None, + error: str | None = None, +) -> None: + """Write module business logs to log/ModuleGeneration/{module}/YYYY-MM-DD.log.""" + _append_json_log( + MODULE_GENERATION_LOG_ROOT, + module, + _base_entry( + log_type="module_generation_event", + domain="module_generation", + module=module, + event_type=event_type, + event_status=event_status, + source=source, + trace_id=trace_id, + request_id=request_id, + user_id=user_id, + project_id=project_id, + task_id=task_id, + step_id=step_id, + remote_action=remote_action, + remote_request_id=remote_request_id, + message=message, + detail=detail, + error=error, + ), + ) + + +def log_ai_model_event( + *, + event_type: str, + module: str | None = None, + event_status: str = "success", + source: str | None = None, + trace_id: str | None = None, + request_id: str | None = None, + user_id: str | None = None, + project_id: str | None = None, + task_id: str | None = None, + step_id: str | None = None, + remote_action: str | None = None, + remote_request_id: str | None = None, + model_config_id: str | None = None, + model_config_name: str | None = None, + model_name: str | None = None, + provider: str | None = None, + api_base: str | None = None, + http_status: int | None = None, + request: dict[str, Any] | None = None, + response: dict[str, Any] | list[Any] | str | None = None, + token_usage: dict[str, Any] | None = None, + message: str | None = None, + detail: dict[str, Any] | None = None, + error: str | None = None, +) -> None: + """Write AI model call logs to existing log/AiModel/YYYY-MM-DD.log.""" + final_detail = dict(detail or {}) + if request is not None: + final_detail["request"] = request + if response is not None: + final_detail["response"] = response + if token_usage is not None: + final_detail["token_usage"] = token_usage + entry = _base_entry( + log_type="ai_model_event", + domain="ai_model", + module=module or "ai_model", + event_type=event_type, + event_status=event_status, + source=source, + trace_id=trace_id, + request_id=request_id, + user_id=user_id, + project_id=project_id, + task_id=task_id, + step_id=step_id, + remote_action=remote_action, + remote_request_id=remote_request_id, + message=message, + detail=final_detail, + error=error, + ) + entry.update( + { + "model_name": model_config_name, + "model_id": model_name, + "model_config_id": model_config_id, + "provider": provider, + "api_base": api_base, + "http_status": http_status, + } + ) + _append_json_log(AI_MODEL_LOG_ROOT, None, entry) + + def log_operation_error(*, domain: str, event_type: str, exc: BaseException | None = None, detail: dict[str, Any] | None = None, **kwargs: Any) -> None: kwargs.setdefault("event_status", "failed") kwargs["detail"] = build_exception_detail(exc, detail) diff --git a/video-gen-api/app/services/shot_replicate_flow_service.py b/video-gen-api/app/services/shot_replicate_flow_service.py index fd5869dc..91397bd3 100644 --- a/video-gen-api/app/services/shot_replicate_flow_service.py +++ b/video-gen-api/app/services/shot_replicate_flow_service.py @@ -1145,6 +1145,10 @@ async def run_video_prompt_optimize(db: AsyncSession, *, project_id: str, step_i video_config=video_config, target_platform=target_platform, schema_config_snapshot=schema_config_snapshot, + module=project.module, + project_id=project.id, + step_id=step.id, + trace_id=f"shot-video-prompt:{step.id}", ) billing = await charge_module_prompt_usage( db, diff --git a/video-gen-api/app/services/shot_replicate_taskset_service.py b/video-gen-api/app/services/shot_replicate_taskset_service.py index e94a4a99..9026698b 100644 --- a/video-gen-api/app/services/shot_replicate_taskset_service.py +++ b/video-gen-api/app/services/shot_replicate_taskset_service.py @@ -15,6 +15,7 @@ from app.enums.shot_replicate import ( ShotAnalysisStatusEnum, ShotSegmentAnalysisStatusEnum, ShotSegmentReplicateStatusEnum, + ShotReplicateLogEventEnum, ShotSegmentSourceModeEnum, ShotSplitStatusEnum, ShotTaskSetStatusEnum, @@ -28,6 +29,7 @@ from app.schemas.shot_replicate import ( ShotSegmentDeleteOut, ShotSegmentDetailOut, ShotSegmentListOut, + ShotReanalyzeOut, ShotSegmentOut, ShotSplitByAIOut, ShotSplitByAIRequest, @@ -716,3 +718,124 @@ async def delete_segment( released_size_bytes=int(released_size_bytes or 0), ) + + +async def prepare_reanalyze_task_set( + db: AsyncSession, + *, + current_user: User, + task_set_id: str, + force: bool = False, + reason: str | None = None, +) -> ShotReanalyzeOut: + """重置原视频分析状态，供 API 重新投递 Celery。""" + task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) + if task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value: + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value, + project_id=task_set.id, + user_id=task_set.user_id, + message="原视频分析正在处理中，拒绝再次分析", + detail={"task_set_id": task_set.id, "analysis_status": task_set.analysis_status, "reason": reason}, + event_status="rejected", + ) + raise HTTPException(status_code=409, detail="原视频分析正在处理中，不能重复投递") + if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value and not force: + raise HTTPException(status_code=409, detail="原视频分析已完成，如确需重跑请传 force=true") + if force: + active_segments_result = await db.execute( + select(func.count()) + .select_from(ShotReplicateSegment) + .where( + ShotReplicateSegment.task_set_id == task_set.id, + ShotReplicateSegment.deleted_at.is_(None), + ) + ) + if int(active_segments_result.scalar() or 0) > 0: + raise HTTPException(status_code=409, detail="当前总任务集已存在拆镜片段，不能强制重跑原视频分析") + + task_set.status = ShotTaskSetStatusEnum.PENDING_ANALYSIS.value + task_set.analysis_status = ShotAnalysisStatusEnum.PENDING.value + task_set.analysis_error_message = None + task_set.original_video_content = None + task_set.original_video_category = None + task_set.original_video_audience = None + task_set.ai_suggestion_json = None + task_set.analysis_raw_json = None + task_set.analysis_result_json = None + await db.flush() + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_RECEIVED.value, + project_id=task_set.id, + user_id=task_set.user_id, + message="原视频再次分析已重置状态", + detail={"task_set_id": task_set.id, "force": force, "reason": reason, "video_url": task_set.video_url}, + ) + return ShotReanalyzeOut( + message="原视频再次分析任务已准备投递", + task_set_id=task_set.id, + segment_id=None, + analysis_status=task_set.analysis_status, + celery_task_name="shot_replicate.analyze_original_video", + ) + + +async def prepare_reanalyze_segment( + db: AsyncSession, + *, + current_user: User, + segment_id: str, + force: bool = False, + reason: str | None = None, +) -> ShotReanalyzeOut: + """重置自定义切片视频分析状态，供 API 重新投递 Celery。""" + segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) + if segment.split_status != ShotSplitStatusEnum.COMPLETED.value: + raise HTTPException(status_code=409, detail="当前片段还未切割完成，不能再次分析") + if not segment.segment_video_url: + raise HTTPException(status_code=409, detail="当前片段缺少 segment_video_url，不能再次分析") + if segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value: + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value, + project_id=segment.task_set_id, + step_id=segment.id, + user_id=segment.user_id, + message="切片视频分析正在处理中，拒绝再次分析", + detail={"segment_id": segment.id, "analysis_status": segment.analysis_status, "reason": reason}, + event_status="rejected", + ) + raise HTTPException(status_code=409, detail="切片视频分析正在处理中，不能重复投递") + if segment.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value and not force: + raise HTTPException(status_code=409, detail="切片视频分析已完成，如确需重跑请传 force=true") + if segment.source_mode != ShotSegmentSourceModeEnum.CUSTOM.value and not force: + raise HTTPException(status_code=409, detail="AI 建议片段默认无需单独分析，如确需重跑请传 force=true") + + segment.analysis_status = ShotSegmentAnalysisStatusEnum.PENDING.value + segment.analysis_error_message = None + segment.analysis_json = None + segment.original_video_content = None + segment.original_video_category = None + segment.original_video_audience = None + segment.segment_content = None + segment.segment_category = None + segment.segment_audience = None + await db.flush() + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_RECEIVED.value, + project_id=segment.task_set_id, + step_id=segment.id, + user_id=segment.user_id, + message="切片视频再次分析已重置状态", + detail={"segment_id": segment.id, "task_set_id": segment.task_set_id, "force": force, "reason": reason, "video_url": segment.segment_video_url}, + ) + return ShotReanalyzeOut( + message="切片视频再次分析任务已准备投递", + task_set_id=segment.task_set_id, + segment_id=segment.id, + analysis_status=segment.analysis_status, + celery_task_name="shot_replicate.analyze_custom_segment_video", + ) diff --git a/video-gen-api/app/services/shot_video_analysis_service.py b/video-gen-api/app/services/shot_video_analysis_service.py index 643a5c2e..10500e55 100644 --- a/video-gen-api/app/services/shot_video_analysis_service.py +++ b/video-gen-api/app/services/shot_video_analysis_service.py @@ -19,6 +19,9 @@ from app.models.token_usage import TokenUsage from app.services.upload_video_asset_service import resolve_upload_video_path from app.services.resource_signed_url_service import build_resource_signed_url from app.utils.id_gen import generate_id +from app.enums.common import LogEventStatusEnum, LogSourceEnum +from app.enums.shot_replicate import ModuleCodeEnum, ShotReplicateLogEventEnum, ShotReplicateRemoteActionEnum +from app.services.operation_log_service import log_ai_model_event AnalysisMode = Literal["full_breakdown", "summary_only"] @@ -74,7 +77,7 @@ def build_file_url_or_data_uri(file_url: str, fallback_mime: str = "video/mp4") # return f"data:{mime};base64,{b64}" -def build_user_message(user_text: str, video_url: str) -> tuple[dict[str, Any], dict[str, Any]]: +def build_user_message(user_text: str, video_url: str) -> tuple[dict[str, Any], dict[str, Any], str]: real_url = build_file_url_or_data_uri(video_url) content_parts = [ { @@ -96,7 +99,7 @@ def build_user_message(user_text: str, video_url: str) -> tuple[dict[str, Any], }, {"type": "text", "text": user_text}, ] - return {"role": "user", "content": content_parts}, {"role": "user", "content": log_content_parts} + return {"role": "user", "content": content_parts}, {"role": "user", "content": log_content_parts}, real_url def build_video_analysis_system_prompt(*, mode: AnalysisMode) -> str: @@ -420,12 +423,106 @@ def _int_usage(value: Any) -> int: except Exception: return 0 + + +def _safe_response_json(response: httpx.Response | None) -> Any: + if response is None: + return None + try: + return response.json() + except Exception: + return {"raw_text": response.text} + + +def _extract_remote_error(data: Any) -> tuple[str | None, str | None, str | None, str | None]: + """Return remote_request_id, remote_code, remote_message, remote_param.""" + if isinstance(data, dict): + err = data.get("error") if isinstance(data.get("error"), dict) else data + remote_code = err.get("code") or err.get("type") if isinstance(err, dict) else None + remote_message = err.get("message") if isinstance(err, dict) else None + remote_param = err.get("param") if isinstance(err, dict) else None + remote_request_id = err.get("request_id") or data.get("request_id") if isinstance(err, dict) else data.get("request_id") + if not remote_request_id and isinstance(remote_message, str): + match = re.search(r"Request id:\s*([a-zA-Z0-9_.:-]+)", remote_message, flags=re.IGNORECASE) + if match: + remote_request_id = match.group(1) + return ( + str(remote_request_id) if remote_request_id else None, + str(remote_code) if remote_code else None, + str(remote_message) if remote_message else None, + str(remote_param) if remote_param else None, + ) + return None, None, str(data) if data is not None else None, None + + +def _log_shot_ai_model_event( + *, + event_type: str, + event_status: str, + config: ModelConfig, + trace_id: str, + user_id: str | None, + task_set_id: str | None, + segment_id: str | None, + mode: AnalysisMode, + request_data: dict[str, Any] | None = None, + response_data: Any = None, + token_usage: dict[str, Any] | None = None, + http_status: int | None = None, + remote_request_id: str | None = None, + remote_code: str | None = None, + remote_message: str | None = None, + remote_param: str | None = None, + message: str | None = None, + error: str | None = None, + extra_detail: dict[str, Any] | None = None, +) -> None: + action = ( + ShotReplicateRemoteActionEnum.ANALYZE_ORIGINAL_VIDEO.value + if mode == "full_breakdown" + else ShotReplicateRemoteActionEnum.ANALYZE_CUSTOM_SEGMENT_VIDEO.value + ) + detail = dict(extra_detail or {}) + detail.update({ + "analysis_mode": mode, + "remote_code": remote_code, + "remote_message": remote_message, + "remote_param": remote_param, + }) + log_ai_model_event( + event_type=event_type, + event_status=event_status, + source=LogSourceEnum.REMOTE_API.value, + module=ModuleCodeEnum.SHOT_REPLICATE.value, + trace_id=trace_id, + user_id=user_id, + project_id=task_set_id, + step_id=segment_id, + remote_action=action, + remote_request_id=remote_request_id, + model_config_id=str(config.id), + model_config_name=config.name, + model_name=config.model_name, + provider=config.provider, + api_base=config.api_base, + http_status=http_status, + request=request_data, + response=response_data, + token_usage=token_usage, + message=message, + detail=detail, + error=error, + ) + async def analyze_video_for_shot_split( db: AsyncSession, video_url: str, *, user_id: str | None = None, mode: AnalysisMode = "full_breakdown", + task_set_id: str | None = None, + segment_id: str | None = None, + trace_id: str | None = None, ) -> ShotVideoAnalysisResult: """调用模型完成拆镜/片段分析。 @@ -433,6 +530,7 @@ async def analyze_video_for_shot_split( 不再读取 SHOT_ANALYSIS_API_BASE / SHOT_ANALYSIS_API_KEY / SHOT_ANALYSIS_MODEL_NAME， 也不再 fallback 到 SEEDANCE_*，避免拆镜分析走错通道。 """ + trace_id = trace_id or generate_id() config = await _select_model_config(db) if not config: raise RuntimeError("拆镜分析模型未配置：请先在 model_configs 表启用可用模型") @@ -445,7 +543,7 @@ async def analyze_video_for_shot_split( system_prompt = build_video_analysis_system_prompt(mode=mode) user_text = build_video_analysis_user_text(mode=mode) - user_message, log_user_message = build_user_message(user_text, video_url) + user_message, log_user_message, real_video_url = build_user_message(user_text, video_url) request_data: dict[str, Any] = { "model": config.model_name, @@ -467,21 +565,128 @@ async def analyze_video_for_shot_split( "model_config_name": config.name, "provider": config.provider, "analysis_mode": mode, + "video_url": video_url, + "signed_video_url": real_video_url, + "video_fps": _video_fps(), + "timeout_seconds": _timeout_seconds(), } url = f"{str(config.api_base).rstrip('/')}/chat/completions" - async with httpx.AsyncClient(timeout=_timeout_seconds()) as client: - response = await client.post( - url, - headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"}, - json=request_data, + _log_shot_ai_model_event( + event_type=( + ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_STARTED.value + if mode == "full_breakdown" + else ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_REMOTE_API_STARTED.value + ), + event_status=LogEventStatusEnum.STARTED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + task_set_id=task_set_id, + segment_id=segment_id, + mode=mode, + request_data={**log_request_data, "api_url": url}, + message="拆镜视频分析模型请求开始", + ) + + try: + async with httpx.AsyncClient(timeout=_timeout_seconds()) as client: + response = await client.post( + url, + headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"}, + json=request_data, + ) + except Exception as exc: + _log_shot_ai_model_event( + event_type=( + ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_FAILED.value + if mode == "full_breakdown" + else ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_REMOTE_API_FAILED.value + ), + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + task_set_id=task_set_id, + segment_id=segment_id, + mode=mode, + request_data={**log_request_data, "api_url": url}, + message="拆镜视频分析模型请求异常", + error=str(exc), + extra_detail={"exception_type": type(exc).__name__}, ) + raise + + response_data = _safe_response_json(response) + remote_request_id, remote_code, remote_message, remote_param = _extract_remote_error(response_data) if response.status_code >= 400: + _log_shot_ai_model_event( + event_type=( + ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_FAILED.value + if mode == "full_breakdown" + else ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_REMOTE_API_FAILED.value + ), + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + task_set_id=task_set_id, + segment_id=segment_id, + mode=mode, + request_data={**log_request_data, "api_url": url}, + response_data=response_data, + http_status=response.status_code, + remote_request_id=remote_request_id, + remote_code=remote_code, + remote_message=remote_message, + remote_param=remote_param, + message="拆镜视频分析模型请求失败", + error=f"HTTP {response.status_code}: {response.text}", + ) raise RuntimeError(f"视频拆镜分析 API 请求失败: HTTP {response.status_code}: {response.text}") - raw = response.json() - content = get_message_content_or_raise(raw) - result = parse_model_json(content) + try: + raw = response.json() + except Exception as exc: + _log_shot_ai_model_event( + event_type=ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_PARSE_FAILED.value, + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + task_set_id=task_set_id, + segment_id=segment_id, + mode=mode, + request_data={**log_request_data, "api_url": url}, + response_data={"raw_text": response.text}, + http_status=response.status_code, + message="拆镜视频分析模型响应 JSON 解析失败", + error=str(exc), + ) + raise + + try: + content = get_message_content_or_raise(raw) + result = parse_model_json(content) + except Exception as exc: + event_type = ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_EMPTY.value if "content 为空" in str(exc) else ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_PARSE_FAILED.value + _log_shot_ai_model_event( + event_type=event_type, + event_status=LogEventStatusEnum.FAILED.value, + config=config, + trace_id=trace_id, + user_id=user_id, + task_set_id=task_set_id, + segment_id=segment_id, + mode=mode, + request_data={**log_request_data, "api_url": url}, + response_data=raw, + http_status=response.status_code, + message="拆镜视频分析模型内容解析失败", + error=str(exc), + ) + raise + result = fill_none_with_wu(result) result = ensure_result_schema(result) result = filter_and_normalize_breakdown(result, mode=mode) @@ -500,6 +705,7 @@ async def analyze_video_for_shot_split( "split_min_seconds": _split_min_seconds(), "split_max_seconds": _split_max_seconds(), "analysis_mode": mode, + "trace_id": trace_id, "log_request": log_request_data, } if not token_usage["total_tokens"]: @@ -525,5 +731,25 @@ async def analyze_video_for_shot_split( "model_name": config.model_name, }) + _log_shot_ai_model_event( + event_type=( + ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_SUCCESS.value + if mode == "full_breakdown" + else ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_REMOTE_API_SUCCESS.value + ), + event_status=LogEventStatusEnum.SUCCESS.value, + config=config, + trace_id=trace_id, + user_id=user_id, + task_set_id=task_set_id, + segment_id=segment_id, + mode=mode, + request_data={**log_request_data, "api_url": url}, + response_data=raw, + token_usage=token_usage, + http_status=response.status_code, + message="拆镜视频分析模型请求成功", + ) + return ShotVideoAnalysisResult(result=result, raw_response=raw, usage=token_usage) diff --git a/video-gen-api/app/tasks/shot_replicate_tasks.py b/video-gen-api/app/tasks/shot_replicate_tasks.py index db149962..3233b36f 100644 --- a/video-gen-api/app/tasks/shot_replicate_tasks.py +++ b/video-gen-api/app/tasks/shot_replicate_tasks.py @@ -11,6 +11,7 @@ from app.enums.credit_record import CreditRecordBillingScene, CreditRecordOwnerT from app.enums.shot_replicate import ( ModuleCodeEnum, ShotAnalysisStatusEnum, + ShotReplicateLogEventEnum, ShotSegmentAnalysisStatusEnum, ShotSegmentSourceModeEnum, ShotSplitStatusEnum, @@ -113,7 +114,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None: log_module_event_file( module=MODULE, - event_type="SHOT_ANALYSIS_STARTED", + event_type=ShotReplicateLogEventEnum.ANALYSIS_STARTED.value, project_id=task_set_id, user_id=task_set_user_id, message="原视频拆镜分析开始", @@ -121,7 +122,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None: ) async with async_session() as db: - analyzed = await analyze_video_for_shot_split(db, video_url or "", user_id=task_set_user_id, mode="full_breakdown") + analyzed = await analyze_video_for_shot_split(db, video_url or "", user_id=task_set_user_id, mode="full_breakdown", task_set_id=task_set_id, trace_id=f"shot-task-set-analysis:{task_set_id}") result = await db.execute( select(ShotReplicateTaskSet) .where(ShotReplicateTaskSet.id == task_set_id, ShotReplicateTaskSet.deleted_at.is_(None)) @@ -157,7 +158,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None: await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id) log_module_prompt_event( - event_type="SHOT_ANALYSIS_SUCCESS", + event_type=ShotReplicateLogEventEnum.ANALYSIS_SUCCESS.value, project_id=task_set_id, step_id=task_set_id, user_id=task_set_user_id or "", @@ -169,7 +170,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None: ) log_module_event_file( module=MODULE, - event_type="SHOT_ANALYSIS_SUCCESS", + event_type=ShotReplicateLogEventEnum.ANALYSIS_SUCCESS.value, project_id=task_set_id, user_id=task_set_user_id, message="原视频拆镜分析成功", @@ -195,7 +196,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None: await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id) log_module_error( module=MODULE, - event_type="SHOT_ANALYSIS_FAILED", + event_type=ShotReplicateLogEventEnum.ANALYSIS_FAILED.value, project_id=task_set_id, user_id=task_set_user_id, message="原视频拆镜分析失败", @@ -241,7 +242,7 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: log_module_event_file( module=MODULE, - event_type="SHOT_SEGMENT_ANALYSIS_STARTED", + event_type=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_STARTED.value, project_id=task_set_id, step_id=segment_id, user_id=user_id, @@ -250,7 +251,7 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: ) async with async_session() as db: - analyzed = await analyze_video_for_shot_split(db, video_url or "", user_id=user_id, mode="summary_only") + analyzed = await analyze_video_for_shot_split(db, video_url or "", user_id=user_id, mode="summary_only", task_set_id=task_set_id, segment_id=segment_id, trace_id=f"shot-segment-analysis:{segment_id}") result = await db.execute( select(ShotReplicateSegment) .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None)) @@ -287,7 +288,7 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id) log_module_prompt_event( - event_type="SHOT_SEGMENT_ANALYSIS_SUCCESS", + event_type=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_SUCCESS.value, project_id=task_set_id, step_id=segment_id, user_id=user_id or "", @@ -299,7 +300,7 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: ) log_module_event_file( module=MODULE, - event_type="SHOT_SEGMENT_ANALYSIS_SUCCESS", + event_type=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_SUCCESS.value, project_id=task_set_id, step_id=segment_id, user_id=user_id, @@ -326,7 +327,7 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id) log_module_error( module=MODULE, - event_type="SHOT_SEGMENT_ANALYSIS_FAILED", + event_type=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_FAILED.value, project_id=task_set_id, step_id=segment_id, user_id=user_id,