json类型数据强制落库操作

video-gen-api/app/services/hot_opening_replicate_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 String, cast, func, or_, 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
@@ -184,6 +186,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,
     *,
@@ -789,8 +804,21 @@ async def update_hot_opening_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()
@@ -800,6 +828,7 @@ async def update_hot_opening_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
 
 
@@ -892,11 +921,15 @@ async def update_hot_opening_image_prompt(
     output_data["manual_edited"] = True
     output_data["manual_edited_at"] = _now().isoformat()
 
-    step.output_json = _step_output(
+    _force_set_json(
-        step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+        step,
-        status=ModuleStepStatusEnum.COMPLETED.value,
+        "output_json",
-        payload=output_data,
+        _step_output(
-        usage=usage,
+            step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            status=ModuleStepStatusEnum.COMPLETED.value,
+            payload=output_data,
+            usage=usage,
+        ),
     )
     step.status = ModuleStepStatusEnum.COMPLETED.value
     step.error_message = None
@@ -920,6 +953,7 @@ async def update_hot_opening_image_prompt(
         message="用户直接修改图片 AI 优化提词，已软删除后续步骤",
         detail={"start_deleted_step_index": STEP_INDEX_MAP[HotOpeningStepCodeEnum.IMAGE_GENERATE.value]},
     )
+    await db.flush()
     return project, step
 
 
@@ -964,11 +998,15 @@ async def update_hot_opening_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=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+        step,
-        status=ModuleStepStatusEnum.COMPLETED.value,
+        "output_json",
-        payload=output_data,
+        _step_output(
-        usage=usage,
+            step_code=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            status=ModuleStepStatusEnum.COMPLETED.value,
+            payload=output_data,
+            usage=usage,
+        ),
     )
     step.status = ModuleStepStatusEnum.COMPLETED.value
     step.error_message = None
@@ -1008,6 +1046,7 @@ async def update_hot_opening_video_prompt_schema(
             ],
         },
     )
+    await db.flush()
     return project, step
 
 
@@ -1138,16 +1177,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=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            step,
-            status=ModuleStepStatusEnum.COMPLETED.value,
+            "output_json",
-            payload={
+            _step_output(
-                "optimized_prompt": optimized,
+                step_code=HotOpeningStepCodeEnum.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 = HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
@@ -1427,16 +1470,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=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            step,
-            status=ModuleStepStatusEnum.COMPLETED.value,
+            "output_json",
-            payload={
+            _step_output(
-                "prompt_schema": prompt_schema,
+                step_code=HotOpeningStepCodeEnum.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 = HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value
@@ -1585,10 +1632,14 @@ async def handle_chat_generation_task_completed(db: AsyncSession, task: ChatGene
     if step.step_code == HotOpeningStepCodeEnum.IMAGE_GENERATE.value:
         step.status = ModuleStepStatusEnum.COMPLETED.value
         step.completed_at = _now()
-        step.output_json = _step_output(
+        _force_set_json(
-            step_code=HotOpeningStepCodeEnum.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=HotOpeningStepCodeEnum.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
@@ -1597,10 +1648,14 @@ async def handle_chat_generation_task_completed(db: AsyncSession, task: ChatGene
     elif step.step_code == HotOpeningStepCodeEnum.VIDEO_GENERATE.value:
         step.status = ModuleStepStatusEnum.COMPLETED.value
         step.completed_at = _now()
-        step.output_json = _step_output(
+        _force_set_json(
-            step_code=HotOpeningStepCodeEnum.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=HotOpeningStepCodeEnum.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