mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

拆镜复刻、爆款开头复刻抽象公共逻辑,拆镜复刻关联项目步骤状态BUG修复

Browse Source
This commit is contained in:
GinHa
2026-06-18 13:03:11 +08:00
parent 158c680a18
commit 0be8091181
6 changed files with 1093 additions and 962 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/shot_replicate_flow_service.py
+140 -485
View File
@@ -53,6 +53,39 @@ from app.services.hot_opening_video_prompt_service import (
from app.services.module_generation_log_service import log_module_error, log_module_event_file, log_module_prompt_event
from app.services.llm import optimize_prompt
from app.services.resource_accounting_service import soft_delete_chat_task_resources
from app.services.module_generation_flow_base_service import (
chat_tasks_by_id as _base_chat_tasks_by_id,
create_module_step as _base_create_step,
get_current_step_by_code as _base_get_current_step_by_code,
get_current_steps as _base_get_current_steps,
get_project_for_user as _base_get_project_for_user,
get_step_for_user as _base_get_step_for_user,
next_version as _base_next_version,
soft_delete_steps_from_index as _base_soft_delete_steps_from_index,
)
from app.enums.module_generation_flow import ModuleGenerationFlowConfig
from app.services.module_generation_step_common_service import (
build_file_url_or_data_uri as _common_build_file_url_or_data_uri,
build_step_input as _common_build_step_input,
build_step_output as _common_build_step_output,
force_set_json as _common_force_set_json,
is_wrapped_step_io as _common_is_wrapped_step_io,
json_dumps as _common_json,
merge_dict as _common_merge_dict,
parse_json as _common_parse_json,
snapshot_from_chat as _common_snapshot_from_chat,
step_payload as _common_step_payload,
step_result as _common_step_result,
step_usage as _common_step_usage,
unwrap_step_output as _common_unwrap_step_output,
utc_now as _common_now,
)
from app.services.module_generation_step_update_service import (
update_module_image_prompt,
update_module_material_input,
update_module_step,
update_module_video_prompt_schema,
)
from app.services.resource_signed_url_service import build_resource_signed_url
from app.utils.id_gen import generate_id
from app.models.shot_replicate_segment import ShotReplicateSegment
@@ -72,28 +105,27 @@ STEP_INDEX_MAP = {
STEP_IO_SCHEMA_VERSION = "shot_replicate_step_io_v1"
def _now() -> datetime:
return datetime.now(timezone.utc)
FLOW_CONFIG = ModuleGenerationFlowConfig(
module=MODULE,
step_index_map=STEP_INDEX_MAP,
material_step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
image_prompt_step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
image_generate_step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
video_prompt_step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
video_generate_step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
project_not_found_message="拆镜复刻项目不存在",
step_not_found_message="子任务不存在",
cancel_chat_task_error_message="拆镜复刻步骤被重新生成或删除，旧生成任务已取消",
material_video_url_editable=False,
step_io_schema_version=STEP_IO_SCHEMA_VERSION,
)
def _json(data: Any) -> str | None:
if data is None:
return None
return json.dumps(data, ensure_ascii=False, default=str)
def _parse_json(value: Any, fallback: Any = None) -> Any:
if value is None or value == "":
return fallback
if isinstance(value, (dict, list)):
return value
if isinstance(value, str):
try:
return json.loads(value)
except Exception:
return fallback
return fallback
_now = _common_now
_json = _common_json
_parse_json = _common_parse_json
_merge_dict = _common_merge_dict
_force_set_json = _common_force_set_json
def _step_input(
@@ -104,16 +136,14 @@ def _step_input(
parent_step_id: str | None = None,
context: dict[str, Any] | None = None,
) -> dict[str, Any]:
return {
"schema_version": STEP_IO_SCHEMA_VERSION,
"step_code": step_code,
"source": {
"source_step_id": source_step_id,
"parent_step_id": parent_step_id,
},
"payload": payload or {},
"context": context or {},
}
return _common_build_step_input(
step_code=step_code,
payload=payload,
source_step_id=source_step_id,
parent_step_id=parent_step_id,
context=context,
schema_version=STEP_IO_SCHEMA_VERSION,
)
def _step_output(
@@ -125,85 +155,35 @@ def _step_output(
usage: dict[str, Any] | None = None,
error: dict[str, Any] | None = None,
) -> dict[str, Any]:
return {
"schema_version": STEP_IO_SCHEMA_VERSION,
"step_code": step_code,
"status": status,
"payload": payload or {},
"result": result or {},
"usage": usage or {},
"error": error or {},
}
return _common_build_step_output(
step_code=step_code,
status=status,
payload=payload,
result=result,
usage=usage,
error=error,
schema_version=STEP_IO_SCHEMA_VERSION,
)
def _is_wrapped_step_io(value: Any) -> bool:
return isinstance(value, dict) and value.get("schema_version") == STEP_IO_SCHEMA_VERSION
return _common_is_wrapped_step_io(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _step_payload(value: Any) -> dict[str, Any]:
data = _parse_json(value, {}) or {}
if _is_wrapped_step_io(data):
payload = data.get("payload")
return payload if isinstance(payload, dict) else {}
return data if isinstance(data, dict) else {}
return _common_step_payload(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _step_result(value: Any) -> dict[str, Any]:
data = _parse_json(value, {}) or {}
if _is_wrapped_step_io(data):
result = data.get("result")
if isinstance(result, dict) and result:
return result
payload = data.get("payload")
return payload if isinstance(payload, dict) else {}
return data if isinstance(data, dict) else {}
return _common_step_result(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _step_usage(value: Any) -> dict[str, Any]:
data = _parse_json(value, {}) or {}
if _is_wrapped_step_io(data):
usage = data.get("usage")
return usage if isinstance(usage, dict) else {}
usage = data.get("token_usage") if isinstance(data, dict) else {}
return usage if isinstance(usage, dict) else {}
return _common_step_usage(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _unwrap_step_output(value: Any) -> dict[str, Any]:
data = _parse_json(value, {}) or {}
if not _is_wrapped_step_io(data):
return data if isinstance(data, dict) else {}
merged: dict[str, Any] = {}
payload = data.get("payload")
result = data.get("result")
usage = data.get("usage")
if isinstance(payload, dict):
merged.update(payload)
if isinstance(result, dict):
merged.update(result)
if isinstance(usage, dict) and usage:
merged["token_usage"] = usage
return merged
def _merge_dict(old: dict[str, Any] | None, new: dict[str, Any] | None) -> dict[str, Any]:
merged = dict(old or {})
for key, value in (new or {}).items():
if value is not None:
merged[key] = value
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)
return _common_unwrap_step_output(value, schema_version=STEP_IO_SCHEMA_VERSION)
async def log_module_event(
@@ -258,22 +238,14 @@ async def _get_project_for_user(
for_update: bool = False,
populate_existing: bool = False,
) -> ModuleGenerationProject:
query = select(ModuleGenerationProject).where(
ModuleGenerationProject.id == project_id,
ModuleGenerationProject.module == MODULE,
ModuleGenerationProject.deleted_at.is_(None),
return await _base_get_project_for_user(
db,
project_id=project_id,
user=user,
config=FLOW_CONFIG,
for_update=for_update,
populate_existing=populate_existing,
)
if not user.is_admin:
query = query.where(ModuleGenerationProject.user_id == user.id)
if populate_existing:
query = query.execution_options(populate_existing=True)
if for_update:
query = query.with_for_update()
result = await db.execute(query.limit(1))
project = result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=404, detail="拆镜复刻项目不存在")
return project
async def _get_step_for_user(
@@ -284,64 +256,26 @@ async def _get_step_for_user(
user: User,
for_update: bool = False,
) -> ModuleGenerationStep:
await _get_project_for_user(db, project_id=project_id, user=user, for_update=for_update)
query = select(ModuleGenerationStep).where(
ModuleGenerationStep.id == step_id,
ModuleGenerationStep.project_id == project_id,
ModuleGenerationStep.module == MODULE,
ModuleGenerationStep.deleted_at.is_(None),
ModuleGenerationStep.is_current == True,
return await _base_get_step_for_user(
db,
project_id=project_id,
step_id=step_id,
user=user,
config=FLOW_CONFIG,
for_update=for_update,
)
if not user.is_admin:
query = query.where(ModuleGenerationStep.user_id == user.id)
if for_update:
query = query.with_for_update()
result = await db.execute(query.limit(1))
step = result.scalar_one_or_none()
if not step:
raise HTTPException(status_code=404, detail="子任务不存在")
return step
async def _get_current_steps(db: AsyncSession, project_id: str) -> list[ModuleGenerationStep]:
result = await db.execute(
select(ModuleGenerationStep)
.where(
ModuleGenerationStep.project_id == project_id,
ModuleGenerationStep.module == MODULE,
ModuleGenerationStep.deleted_at.is_(None),
ModuleGenerationStep.is_current == True,
)
.order_by(ModuleGenerationStep.step_index.asc(), ModuleGenerationStep.created_at.asc())
)
return list(result.scalars().all())
return await _base_get_current_steps(db, project_id=project_id, config=FLOW_CONFIG)
async def _get_current_step_by_code(db: AsyncSession, project_id: str, step_code: str) -> ModuleGenerationStep | None:
result = await db.execute(
select(ModuleGenerationStep)
.where(
ModuleGenerationStep.project_id == project_id,
ModuleGenerationStep.module == MODULE,
ModuleGenerationStep.step_code == step_code,
ModuleGenerationStep.is_current == True,
ModuleGenerationStep.deleted_at.is_(None),
)
.order_by(ModuleGenerationStep.version.desc(), ModuleGenerationStep.created_at.desc())
.limit(1)
)
return result.scalar_one_or_none()
return await _base_get_current_step_by_code(db, project_id=project_id, step_code=step_code, config=FLOW_CONFIG)
async def _next_version(db: AsyncSession, project_id: str, step_code: str) -> int:
result = await db.execute(
select(func.max(ModuleGenerationStep.version)).where(
ModuleGenerationStep.project_id == project_id,
ModuleGenerationStep.module == MODULE,
ModuleGenerationStep.step_code == step_code,
)
)
return int(result.scalar_one_or_none() or 0) + 1
return await _base_next_version(db, project_id=project_id, step_code=step_code, config=FLOW_CONFIG)
async def _create_step(
@@ -356,39 +290,19 @@ async def _create_step(
input_data: dict[str, Any] | None = None,
output_data: dict[str, Any] | None = None,
) -> ModuleGenerationStep:
version = await _next_version(db, project.id, step_code)
step = ModuleGenerationStep(
id=generate_id(),
project_id=project.id,
user_id=project.user_id,
module=project.module,
step_index=STEP_INDEX_MAP[step_code],
return await _base_create_step(
db,
project=project,
step_code=step_code,
config=FLOW_CONFIG,
log_module_event=log_module_event,
status=status,
version=version,
is_current=True,
parent_step_id=parent_step_id,
source_step_id=source_step_id,
chat_task_id=chat_task_id,
input_json=_step_input(
step_code=step_code,
payload=input_data,
source_step_id=source_step_id,
parent_step_id=parent_step_id,
) if input_data is not None else None,
output_json=_step_output(
step_code=step_code,
status=status,
result=output_data,
) if output_data is not None else None,
started_at=_now() if status == ModuleStepStatusEnum.PROCESSING.value else None,
completed_at=_now() if status == ModuleStepStatusEnum.COMPLETED.value else None,
input_data=input_data,
output_data=output_data,
)
db.add(step)
project.current_step_code = step_code
await db.flush()
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.STEP_CREATED.value, detail={"step_code": step_code, "version": version})
return step
async def _soft_delete_steps_from_index(
@@ -398,49 +312,14 @@ async def _soft_delete_steps_from_index(
start_index: int,
deleted_at: datetime | None = None,
) -> None:
deleted_at = deleted_at or _now()
result = await db.execute(
select(ModuleGenerationStep)
.where(
ModuleGenerationStep.project_id == project.id,
ModuleGenerationStep.module == MODULE,
ModuleGenerationStep.is_current == True,
ModuleGenerationStep.deleted_at.is_(None),
ModuleGenerationStep.step_index >= start_index,
)
.with_for_update()
await _base_soft_delete_steps_from_index(
db,
project=project,
start_index=start_index,
config=FLOW_CONFIG,
log_module_event=log_module_event,
deleted_at=deleted_at,
)
steps = list(result.scalars().all())
for step in steps:
step.is_current = False
step.deleted_at = deleted_at
if step.chat_task_id:
chat_result = await db.execute(
select(ChatGenerationTask)
.where(ChatGenerationTask.id == step.chat_task_id, ChatGenerationTask.deleted_at.is_(None))
.with_for_update()
.limit(1)
)
chat_task = chat_result.scalar_one_or_none()
if chat_task:
if chat_task.status == "completed":
await soft_delete_chat_task_resources(db, chat_task.id, deleted_at=deleted_at)
elif chat_task.status != "failed":
await mark_chat_generation_task_failed_and_refund_once(
db,
task=chat_task,
error_message="拆镜复刻步骤被重新生成或删除，旧生成任务已取消",
pipeline_stage="failed",
)
chat_task.deleted_at = deleted_at
if steps:
await log_module_event(
db,
project=project,
event_type=ModuleEventTypeEnum.SOFT_DELETE_STEPS.value,
message=f"软删除第 {start_index} 步及之后的旧子任务",
detail={"step_ids": [step.id for step in steps]},
)
def _step_to_out(step: ModuleGenerationStep) -> ShotReplicateStepOut:
@@ -466,17 +345,11 @@ def _step_to_out(step: ModuleGenerationStep) -> ShotReplicateStepOut:
def _snapshot_from_chat(chat_task: ChatGenerationTask | None) -> dict[str, Any]:
if not chat_task:
return {}
return _parse_json(chat_task.engine_snapshot_json, {}) or {}
return _common_snapshot_from_chat(chat_task)
async def _chat_tasks_by_id(db: AsyncSession, steps: list[ModuleGenerationStep]) -> dict[str, ChatGenerationTask]:
ids = [step.chat_task_id for step in steps if step.chat_task_id]
if not ids:
return {}
result = await db.execute(select(ChatGenerationTask).where(ChatGenerationTask.id.in_(ids)))
return {task.id: task for task in result.scalars().all()}
return await _base_chat_tasks_by_id(db, steps)
async def project_to_detail_out(db: AsyncSession, project: ModuleGenerationProject) -> ShotReplicateTaskDetailOut:
@@ -703,73 +576,15 @@ async def update_shot_replicate_step(
step_id: str,
req: ShotReplicateStepUpdate,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
step = await _get_step_for_user(db, project_id=project_id, step_id=step_id, user=current_user, for_update=True)
if step.status == ModuleStepStatusEnum.PROCESSING.value:
raise HTTPException(status_code=400, detail="当前子任务正在处理中，暂不能修改")
input_data = _step_payload(step.input_json)
output_data = _unwrap_step_output(step.output_json)
if step.step_code == ShotReplicateStepCodeEnum.MATERIAL_INPUT.value:
input_data = _merge_dict(
input_data,
{
"material_video_url": req.material_video_url,
"material_image_url": req.material_image_url,
"source_project_name": req.source_project_name,
"target_project_name": req.target_project_name,
"core_content_point": req.core_content_point,
},
)
if req.target_project_name:
project.title = req.target_project_name
elif step.step_code == ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
if req.prompt is not None:
output_data["optimized_prompt"] = req.prompt
output_data["prompt"] = req.prompt
elif step.step_code == ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
if req.prompt_schema is not None:
output_data["prompt_schema"] = req.prompt_schema
if req.prompt is not None:
output_data["final_prompt"] = req.prompt
else:
if req.input_json:
input_data = _merge_dict(input_data, req.input_json)
if req.output_json:
output_data = _merge_dict(output_data, req.output_json)
if req.input_json:
input_data = _merge_dict(input_data, req.input_json)
if req.output_json:
output_data = _merge_dict(output_data, req.output_json)
_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,
),
return await update_module_step(
db,
current_user=current_user,
project_id=project_id,
step_id=step_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
)
_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()
project.status = ModuleProjectStatusEnum.WAITING_USER.value
project.current_step_code = step.step_code
project.error_message = None
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
async def update_shot_replicate_material_input(
@@ -784,56 +599,14 @@ async def update_shot_replicate_material_input(
采用方案 B：软删除旧第1步及之后的当前有效步骤，然后新建第1步 version+1。
未传字段沿用旧第1步素材输入，避免前端只改一个字段时丢失其它素材信息。
"""
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
old_material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
old_material = _step_payload(old_material_step.input_json if old_material_step else None)
# 拆镜复刻的素材视频来自 shot_replicate_segments.segment_video_url，
# 与片段强绑定，不允许前端在项目素材修改接口中覆盖。
material = {
"material_video_url": old_material.get("material_video_url"),
"material_image_url": req.material_image_url if req.material_image_url is not None else old_material.get("material_image_url"),
"source_project_name": req.source_project_name if req.source_project_name is not None else old_material.get("source_project_name"),
"target_project_name": req.target_project_name if req.target_project_name is not None else old_material.get("target_project_name"),
"core_content_point": req.core_content_point if req.core_content_point is not None else old_material.get("core_content_point"),
}
missing_fields = [key for key, value in material.items() if value is None or str(value).strip() == ""]
if missing_fields:
raise HTTPException(status_code=400, detail=f"素材输入缺少必要字段: {', '.join(missing_fields)}")
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.MATERIAL_INPUT.value])
project.title = str(material["target_project_name"])
project.status = ModuleProjectStatusEnum.WAITING_USER.value
project.current_step_code = ShotReplicateStepCodeEnum.MATERIAL_INPUT.value
project.final_image_url = None
project.final_video_url = None
project.final_video_cover_url = None
project.error_message = None
project.completed_at = None
new_step = await _create_step(
return await update_module_material_input(
db,
project=project,
step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
status=ModuleStepStatusEnum.COMPLETED.value,
input_data=material,
output_data={"message": "素材输入已修改，旧步骤已软删除。下一步请重新生成图片AI提词。"},
current_user=current_user,
project_id=project_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
)
await log_module_event(
db,
project=project,
step=new_step,
event_type=ModuleEventTypeEnum.STEP_UPDATED.value,
message="用户修改素材输入并重建第1步新版本",
detail={
"old_material_step_id": old_material_step.id if old_material_step else None,
"new_material_step_id": new_step.id,
"version": new_step.version,
},
)
return project.id, new_step.id
async def update_shot_replicate_image_prompt(
@@ -848,55 +621,15 @@ async def update_shot_replicate_image_prompt(
修改后软删除第3、4、5步当前有效任务，让用户从图片生成开始重新执行。
"""
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
step = await _get_step_for_user(db, project_id=project_id, step_id=step_id, user=current_user, for_update=True)
if step.step_code != ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
raise HTTPException(status_code=400, detail="只能修改第2步图片 AI 提词子任务")
if step.status != ModuleStepStatusEnum.COMPLETED.value:
raise HTTPException(status_code=400, detail="图片 AI 提词未完成，不能直接修改")
output_data = _step_payload(step.output_json)
usage = _step_usage(step.output_json)
new_prompt = req.prompt.strip()
output_data["optimized_prompt"] = new_prompt
output_data["prompt"] = new_prompt
output_data["manual_edited"] = True
output_data["manual_edited_at"] = _now().isoformat()
_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
step.completed_at = _now()
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_GENERATE.value])
project.status = ModuleProjectStatusEnum.WAITING_USER.value
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
project.final_image_url = None
project.final_video_url = None
project.final_video_cover_url = None
project.completed_at = None
project.error_message = None
await log_module_event(
return await update_module_image_prompt(
db,
project=project,
step=step,
event_type=ModuleEventTypeEnum.STEP_UPDATED.value,
message="用户直接修改图片 AI 优化提词，已软删除后续步骤",
detail={"start_deleted_step_index": STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_GENERATE.value]},
current_user=current_user,
project_id=project_id,
step_id=step_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
)
await db.flush()
return project, step
async def update_shot_replicate_video_prompt_schema(
@@ -912,84 +645,17 @@ async def update_shot_replicate_video_prompt_schema(
服务端已有 schema 为基准：视频规格、数组长度、时间段、合规控制、质量控制、协议字段均锁定。
最终提示词允许修改，但保存前会清洗视频时长、比例、分辨率、帧率等参数。
"""
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
step = await _get_step_for_user(db, project_id=project_id, step_id=step_id, user=current_user, for_update=True)
if step.step_code != ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
raise HTTPException(status_code=400, detail="只能修改第4步视频 AI 提词 JSON schema 子任务")
if step.status != ModuleStepStatusEnum.COMPLETED.value:
raise HTTPException(status_code=400, detail="视频 AI 提词未完成，不能直接修改")
output_data = _step_payload(step.output_json)
usage = _step_usage(step.output_json)
input_data = _step_payload(step.input_json)
server_schema = output_data.get("prompt_schema") if isinstance(output_data.get("prompt_schema"), dict) else {}
video_config = output_data.get("params_used_for_prompt") or input_data.get("video_config") or {}
if not isinstance(video_config, dict) or not video_config.get("duration") or not video_config.get("aspect_ratio") or not video_config.get("resolution"):
raise HTTPException(status_code=400, detail="缺少第4步视频参数快照，不能安全修改视频 schema")
patched_schema = patch_video_prompt_schema_from_client(
server_schema=server_schema,
client_schema=req.prompt_schema,
video_config=video_config,
)
final_prompt = build_final_video_prompt(patched_schema)
output_data["prompt_schema"] = patched_schema
output_data["final_prompt"] = final_prompt
output_data["params_used_for_prompt"] = video_config
output_data["manual_edited"] = True
output_data["manual_edited_at"] = _now().isoformat()
_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
step.completed_at = _now()
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.VIDEO_GENERATE.value])
project.status = ModuleProjectStatusEnum.WAITING_USER.value
project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value
project.final_video_url = None
project.final_video_cover_url = None
project.completed_at = None
project.error_message = None
await log_module_event(
return await update_module_video_prompt_schema(
db,
project=project,
step=step,
event_type=ModuleEventTypeEnum.STEP_UPDATED.value,
message="用户修改视频 AI 提词 schema，已软删除视频生成步骤",
detail={
"start_deleted_step_index": STEP_INDEX_MAP[ShotReplicateStepCodeEnum.VIDEO_GENERATE.value],
"locked_fields": [
"schema_version",
"schema_usage",
"画面属性.视频时长",
"画面属性.视频比例",
"画面属性.清晰度",
"画面属性.帧率",
"画面属性.推荐分辨率",
"动作流程[*].时间段",
"镜头流程[*].时间段",
"动态时间规划",
"输出规格限制",
"质量控制",
"合规控制",
],
},
current_user=current_user,
project_id=project_id,
step_id=step_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
patch_video_prompt_schema_from_client=patch_video_prompt_schema_from_client,
build_final_video_prompt=build_final_video_prompt,
)
await db.flush()
return project, step
async def submit_image_prompt_optimize(
@@ -1764,7 +1430,7 @@ async def create_shot_replicate_project_from_segment(
module=MODULE,
title=req.target_project_name,
status=ModuleProjectStatusEnum.WAITING_USER.value,
current_step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
current_step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
idempotency_key=req.idempotency_key,
)
db.add(project)
@@ -1815,7 +1481,7 @@ async def create_shot_replicate_project_from_segment(
**context["analysis"],
},
)
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
project.current_step_code = ShotReplicateStepCodeEnum.MATERIAL_INPUT.value
project.status = ModuleProjectStatusEnum.WAITING_USER.value
segment.module_project_id = project.id
segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
@@ -1832,15 +1498,4 @@ async def create_shot_replicate_project_from_segment(
def _build_file_url_or_data_uri(file_url: str) -> str:
"""构建模型可访问的素材 URL。
- /uploads 与 /uploads/shot_segments 属于上传/素材域，不走 generate token 签名。
- /generate 下的 AI 生成资源继续走 token 签名。
- 远程 URL / data URI 原样返回。
"""
if file_url.startswith(("http://", "https://", "data:")):
return file_url
if file_url.startswith("/generate/"):
file_url = build_resource_signed_url(resource_url=file_url, expire_seconds=86400)
base_url = str(settings.BASE_URL or "").rstrip("/")
return f"{base_url}{file_url}" if base_url else file_url
return _common_build_file_url_or_data_uri(file_url)