修复拆机复刻json类型落库失败BUG，修补celery重试遗漏机制

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)
+    _force_set_json(
-    step.output_json = _step_output(step_code=step.step_code, status=ModuleStepStatusEnum.COMPLETED.value, payload=output_data)
+        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(
+    _force_set_json(
-        step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+        step,
-        status=ModuleStepStatusEnum.COMPLETED.value,
+        "output_json",
-        payload=output_data,
+        _step_output(
-        usage=usage,
+            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(
+    _force_set_json(
-        step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+        step,
-        status=ModuleStepStatusEnum.COMPLETED.value,
+        "output_json",
-        payload=output_data,
+        _step_output(
-        usage=usage,
+            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(
+        _force_set_json(
-            step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            step,
-            status=ModuleStepStatusEnum.COMPLETED.value,
+            "output_json",
-            payload={
+            _step_output(
-                "optimized_prompt": optimized,
+                step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
-                "prompt": optimized,
+                status=ModuleStepStatusEnum.COMPLETED.value,
-                "original_prompt": prompt_text,
+                payload={
-                "references": references,
+                    "optimized_prompt": optimized,
-            },
+                    "prompt": optimized,
-            usage=usage,
+                    "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(
+        _force_set_json(
-            step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            step,
-            status=ModuleStepStatusEnum.COMPLETED.value,
+            "output_json",
-            payload={
+            _step_output(
-                "prompt_schema": prompt_schema,
+                step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
-                "final_prompt": final_prompt,
+                status=ModuleStepStatusEnum.COMPLETED.value,
-                "params_used_for_prompt": video_config,
+                payload={
-                "target_platform": target_platform,
+                    "prompt_schema": prompt_schema,
-            },
+                    "final_prompt": final_prompt,
-            usage=usage,
+                    "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(
+        _force_set_json(
-            step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
+            step,
-            status=ModuleStepStatusEnum.COMPLETED.value,
+            "output_json",
-            result={"result_image_url": task.image_url, "chat_task_id": task.id},
+            _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(
+        _force_set_json(
-            step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
+            step,
-            status=ModuleStepStatusEnum.COMPLETED.value,
+            "output_json",
-            result={"result_video_url": task.video_url, "result_video_cover_url": task.video_cover_url, "chat_task_id": task.id},
+            _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

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 提词任务。"""
+        """手动触发后的图片 AI 提词任务。
-        return run_async(_run_image_prompt(project_id, step_id))
+
+        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 提词任务。"""
+        """手动触发后的视频 AI 提词任务。
-        return run_async(_run_video_prompt(project_id, step_id))
+
+        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):

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: