From 3453a6445c3aa3d20d6c45ade47051d70770943e Mon Sep 17 00:00:00 2001 From: GinHa <15201596918@163.com> Date: Thu, 18 Jun 2026 10:22:42 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=8B=86=E6=9C=BA=E5=A4=8D?= =?UTF-8?q?=E5=88=BBjson=E7=B1=BB=E5=9E=8B=E8=90=BD=E5=BA=93=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5BUG=EF=BC=8C=E4=BF=AE=E8=A1=A5celery=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E9=81=97=E6=BC=8F=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/shot_replicate_flow_service.py | 135 ++++++++++++------ .../app/tasks/hot_opening_replicate_tasks.py | 24 +++- .../app/tasks/shot_replicate_flow_tasks.py | 22 ++- 3 files changed, 135 insertions(+), 46 deletions(-) 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 0d000938..7383ee9c 100644 --- a/video-gen-api/app/services/shot_replicate_flow_service.py +++ b/video-gen-api/app/services/shot_replicate_flow_service.py @@ -1,12 +1,14 @@ from __future__ import annotations import json +from copy import deepcopy from datetime import datetime, timezone from typing import Any from fastapi import HTTPException from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm.attributes import flag_modified from app.config import settings from app.enums.common import ModuleEventTypeEnum, ModuleProjectStatusEnum, ModulePromptTypeEnum, ModuleStepStatusEnum @@ -191,6 +193,19 @@ def _merge_dict(old: dict[str, Any] | None, new: dict[str, Any] | None) -> dict[ return merged +def _force_set_json(model_obj: Any, field_name: str, value: Any) -> None: + """强制持久化 JSON / JSONB 字段。 + + SQLAlchemy 对 dict/list 的嵌套原地修改不会稳定触发 dirty 判定。 + 所有编辑类接口在写入 input_json / output_json 时统一走这里： + 1. deepcopy 断开旧引用； + 2. 整体重新赋值； + 3. flag_modified 显式标记字段已变更。 + """ + setattr(model_obj, field_name, deepcopy(value)) + flag_modified(model_obj, field_name) + + async def log_module_event( db: AsyncSession, *, @@ -729,8 +744,21 @@ async def update_shot_replicate_step( if req.output_json: output_data = _merge_dict(output_data, req.output_json) - step.input_json = _step_input(step_code=step.step_code, payload=input_data, source_step_id=step.source_step_id, parent_step_id=step.parent_step_id) - step.output_json = _step_output(step_code=step.step_code, status=ModuleStepStatusEnum.COMPLETED.value, payload=output_data) + _force_set_json( + step, + "input_json", + _step_input( + step_code=step.step_code, + payload=input_data, + source_step_id=step.source_step_id, + parent_step_id=step.parent_step_id, + ), + ) + _force_set_json( + step, + "output_json", + _step_output(step_code=step.step_code, status=ModuleStepStatusEnum.COMPLETED.value, payload=output_data), + ) step.status = ModuleStepStatusEnum.COMPLETED.value step.error_message = None step.completed_at = _now() @@ -740,6 +768,7 @@ async def update_shot_replicate_step( await _soft_delete_steps_from_index(db, project=project, start_index=step.step_index + 1) await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.STEP_UPDATED.value, message="用户修改子任务内容") + await db.flush() return project, step @@ -834,11 +863,15 @@ async def update_shot_replicate_image_prompt( output_data["manual_edited"] = True output_data["manual_edited_at"] = _now().isoformat() - step.output_json = _step_output( - step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value, - status=ModuleStepStatusEnum.COMPLETED.value, - payload=output_data, - usage=usage, + _force_set_json( + step, + "output_json", + _step_output( + step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value, + status=ModuleStepStatusEnum.COMPLETED.value, + payload=output_data, + usage=usage, + ), ) step.status = ModuleStepStatusEnum.COMPLETED.value step.error_message = None @@ -862,6 +895,7 @@ async def update_shot_replicate_image_prompt( message="用户直接修改图片 AI 优化提词，已软删除后续步骤", detail={"start_deleted_step_index": STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_GENERATE.value]}, ) + await db.flush() return project, step @@ -906,11 +940,15 @@ async def update_shot_replicate_video_prompt_schema( output_data["manual_edited"] = True output_data["manual_edited_at"] = _now().isoformat() - step.output_json = _step_output( - step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value, - status=ModuleStepStatusEnum.COMPLETED.value, - payload=output_data, - usage=usage, + _force_set_json( + step, + "output_json", + _step_output( + step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value, + status=ModuleStepStatusEnum.COMPLETED.value, + payload=output_data, + usage=usage, + ), ) step.status = ModuleStepStatusEnum.COMPLETED.value step.error_message = None @@ -950,6 +988,7 @@ async def update_shot_replicate_video_prompt_schema( ], }, ) + await db.flush() return project, step @@ -1087,16 +1126,20 @@ async def run_image_prompt_optimize(db: AsyncSession, *, project_id: str, step_i }) step.status = ModuleStepStatusEnum.COMPLETED.value step.completed_at = _now() - step.output_json = _step_output( - step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value, - status=ModuleStepStatusEnum.COMPLETED.value, - payload={ - "optimized_prompt": optimized, - "prompt": optimized, - "original_prompt": prompt_text, - "references": references, - }, - usage=usage, + _force_set_json( + step, + "output_json", + _step_output( + step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value, + status=ModuleStepStatusEnum.COMPLETED.value, + payload={ + "optimized_prompt": optimized, + "prompt": optimized, + "original_prompt": prompt_text, + "references": references, + }, + usage=usage, + ), ) project.status = ModuleProjectStatusEnum.WAITING_USER.value project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value @@ -1386,16 +1429,20 @@ async def run_video_prompt_optimize(db: AsyncSession, *, project_id: str, step_i }) step.status = ModuleStepStatusEnum.COMPLETED.value step.completed_at = _now() - step.output_json = _step_output( - step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value, - status=ModuleStepStatusEnum.COMPLETED.value, - payload={ - "prompt_schema": prompt_schema, - "final_prompt": final_prompt, - "params_used_for_prompt": video_config, - "target_platform": target_platform, - }, - usage=usage, + _force_set_json( + step, + "output_json", + _step_output( + step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value, + status=ModuleStepStatusEnum.COMPLETED.value, + payload={ + "prompt_schema": prompt_schema, + "final_prompt": final_prompt, + "params_used_for_prompt": video_config, + "target_platform": target_platform, + }, + usage=usage, + ), ) project.status = ModuleProjectStatusEnum.WAITING_USER.value project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value @@ -1549,10 +1596,14 @@ async def handle_chat_generation_task_completed(db: AsyncSession, task: ChatGene if step.step_code == ShotReplicateStepCodeEnum.IMAGE_GENERATE.value: step.status = ModuleStepStatusEnum.COMPLETED.value step.completed_at = _now() - step.output_json = _step_output( - step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value, - status=ModuleStepStatusEnum.COMPLETED.value, - result={"result_image_url": task.image_url, "chat_task_id": task.id}, + _force_set_json( + step, + "output_json", + _step_output( + step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value, + status=ModuleStepStatusEnum.COMPLETED.value, + result={"result_image_url": task.image_url, "chat_task_id": task.id}, + ), ) project.final_image_url = task.image_url project.status = ModuleProjectStatusEnum.WAITING_USER.value @@ -1561,10 +1612,14 @@ async def handle_chat_generation_task_completed(db: AsyncSession, task: ChatGene elif step.step_code == ShotReplicateStepCodeEnum.VIDEO_GENERATE.value: step.status = ModuleStepStatusEnum.COMPLETED.value step.completed_at = _now() - step.output_json = _step_output( - step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value, - status=ModuleStepStatusEnum.COMPLETED.value, - result={"result_video_url": task.video_url, "result_video_cover_url": task.video_cover_url, "chat_task_id": task.id}, + _force_set_json( + step, + "output_json", + _step_output( + step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value, + status=ModuleStepStatusEnum.COMPLETED.value, + result={"result_video_url": task.video_url, "result_video_cover_url": task.video_cover_url, "chat_task_id": task.id}, + ), ) project.final_video_url = task.video_url project.final_video_cover_url = task.video_cover_url diff --git a/video-gen-api/app/tasks/hot_opening_replicate_tasks.py b/video-gen-api/app/tasks/hot_opening_replicate_tasks.py index 0dc9f5ba..9b172f97 100644 --- a/video-gen-api/app/tasks/hot_opening_replicate_tasks.py +++ b/video-gen-api/app/tasks/hot_opening_replicate_tasks.py @@ -77,13 +77,29 @@ async def _run_video_prompt(project_id: str, step_id: str | None = None): if celery_app: @celery_app.task(name="hot_opening.start_image_prompt_optimize", bind=True, max_retries=3, default_retry_delay=30) def start_image_prompt_optimize(self, project_id: str, step_id: str | None = None): - """手动触发后的图片 AI 提词任务。""" - return run_async(_run_image_prompt(project_id, step_id)) + """手动触发后的图片 AI 提词任务。 + + max_retries 只有在显式 self.retry() 时才会生效。 + 这里仅兜底 Celery 包装层 / DB commit / Redis registry 等未被 service 捕获的异常； + service 内部已经落库为业务失败的情况不会抛出异常，也不会重复 retry。 + """ + try: + return run_async(_run_image_prompt(project_id, step_id)) + except Exception as exc: + raise self.retry(exc=exc) from exc @celery_app.task(name="hot_opening.start_video_prompt_optimize", bind=True, max_retries=3, default_retry_delay=30) def start_video_prompt_optimize(self, project_id: str, step_id: str | None = None): - """手动触发后的视频 AI 提词任务。""" - return run_async(_run_video_prompt(project_id, step_id)) + """手动触发后的视频 AI 提词任务。 + + max_retries 只有在显式 self.retry() 时才会生效。 + 这里仅兜底 Celery 包装层 / DB commit / Redis registry 等未被 service 捕获的异常； + service 内部已经落库为业务失败的情况不会抛出异常，也不会重复 retry。 + """ + try: + return run_async(_run_video_prompt(project_id, step_id)) + except Exception as exc: + raise self.retry(exc=exc) from exc else: class _DisabledTask: def delay(self, *args: Any, **kwargs: Any): diff --git a/video-gen-api/app/tasks/shot_replicate_flow_tasks.py b/video-gen-api/app/tasks/shot_replicate_flow_tasks.py index ed603dcf..6c5105ef 100644 --- a/video-gen-api/app/tasks/shot_replicate_flow_tasks.py +++ b/video-gen-api/app/tasks/shot_replicate_flow_tasks.py @@ -77,12 +77,30 @@ if celery_app: @celery_app.task(name="shot_replicate.start_image_prompt_optimize", bind=True, max_retries=3, default_retry_delay=30) def start_image_prompt_optimize(self, project_id: str, step_id: str | None = None): - return run_async(_run_image_prompt(project_id, step_id)) + """手动触发后的图片 AI 提词任务。 + + max_retries 只有在显式 self.retry() 时才会生效。 + 这里仅兜底 Celery 包装层 / DB commit / Redis registry 等未被 service 捕获的异常； + service 内部已经落库为业务失败的情况不会抛出异常，也不会重复 retry。 + """ + try: + return run_async(_run_image_prompt(project_id, step_id)) + except Exception as exc: + raise self.retry(exc=exc) from exc @celery_app.task(name="shot_replicate.start_video_prompt_optimize", bind=True, max_retries=3, default_retry_delay=30) def start_video_prompt_optimize(self, project_id: str, step_id: str | None = None): - return run_async(_run_video_prompt(project_id, step_id)) + """手动触发后的视频 AI 提词任务。 + + max_retries 只有在显式 self.retry() 时才会生效。 + 这里仅兜底 Celery 包装层 / DB commit / Redis registry 等未被 service 捕获的异常； + service 内部已经落库为业务失败的情况不会抛出异常，也不会重复 retry。 + """ + try: + return run_async(_run_video_prompt(project_id, step_id)) + except Exception as exc: + raise self.retry(exc=exc) from exc else: