1847 lines
80 KiB
Python
1847 lines
80 KiB
Python
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
|
||
from app.enums.shot_replicate import ShotReplicateGenerationModeEnum, ShotReplicateStepCodeEnum, ModuleCodeEnum
|
||
from app.models.chat_generation_task import ChatGenerationTask
|
||
from app.models.module_generation_project import ModuleGenerationProject
|
||
from app.models.module_generation_step import ModuleGenerationStep
|
||
from app.models.user import User
|
||
from app.schemas.shot_replicate import (
|
||
ShotReplicateDeleteOut,
|
||
ShotReplicateGenerateImageRequest,
|
||
ShotReplicateGenerateVideoPromptRequest,
|
||
ShotReplicateGenerateVideoRequest,
|
||
ShotReplicateImageGenerationOut,
|
||
ShotReplicateImagePromptUpdateRequest,
|
||
ShotReplicateMaterialOut,
|
||
ShotReplicateMaterialUpdateRequest,
|
||
ShotReplicateStepOut,
|
||
ShotReplicateStepUpdate,
|
||
ShotReplicateTaskCreate,
|
||
ShotReplicateTaskDetailOut,
|
||
ShotReplicateTaskListItemOut,
|
||
ShotReplicateTaskListOut,
|
||
ShotReplicateVideoGenerationOut,
|
||
ShotReplicateVideoPromptSchemaUpdateRequest,
|
||
)
|
||
from app.services.generation_ai_service import (
|
||
VIDEO_DEFAULT_DURATION,
|
||
VIDEO_DEFAULT_RATIO,
|
||
VIDEO_DEFAULT_RESOLUTION,
|
||
_get_video_engine,
|
||
_parse_list,
|
||
)
|
||
from app.services.generation_billing_service import charge_module_prompt_usage
|
||
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
|
||
from app.services.generation_task_factory_service import create_chat_generation_task_for_module
|
||
from app.services.hot_opening_video_prompt_service import (
|
||
build_final_video_prompt,
|
||
optimize_hot_opening_video_prompt as optimize_shot_replicate_video_prompt,
|
||
patch_video_prompt_schema_from_client,
|
||
)
|
||
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.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
|
||
from app.enums.shot_replicate import ShotSegmentReplicateStatusEnum, ShotSplitStatusEnum
|
||
from app.schemas.shot_replicate import ShotSegmentReplicationCreateRequest
|
||
|
||
MODULE = ModuleCodeEnum.SHOT_REPLICATE.value
|
||
GENERATION_MODE = ShotReplicateGenerationModeEnum.SHOT_REPLICATE.value
|
||
|
||
STEP_INDEX_MAP = {
|
||
ShotReplicateStepCodeEnum.MATERIAL_INPUT.value: 1,
|
||
ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value: 2,
|
||
ShotReplicateStepCodeEnum.IMAGE_GENERATE.value: 3,
|
||
ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value: 4,
|
||
ShotReplicateStepCodeEnum.VIDEO_GENERATE.value: 5,
|
||
}
|
||
|
||
STEP_IO_SCHEMA_VERSION = "shot_replicate_step_io_v1"
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
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
|
||
|
||
|
||
def _step_input(
|
||
*,
|
||
step_code: str,
|
||
payload: dict[str, Any] | None = None,
|
||
source_step_id: str | None = None,
|
||
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 {},
|
||
}
|
||
|
||
|
||
def _step_output(
|
||
*,
|
||
step_code: str,
|
||
status: str,
|
||
payload: dict[str, Any] | None = None,
|
||
result: dict[str, Any] | None = None,
|
||
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 {},
|
||
}
|
||
|
||
|
||
def _is_wrapped_step_io(value: Any) -> bool:
|
||
return isinstance(value, dict) and value.get("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 {}
|
||
|
||
|
||
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 {}
|
||
|
||
|
||
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 {}
|
||
|
||
|
||
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)
|
||
|
||
|
||
async def log_module_event(
|
||
db: AsyncSession,
|
||
*,
|
||
project: ModuleGenerationProject,
|
||
event_type: str,
|
||
step: ModuleGenerationStep | None = None,
|
||
message: str | None = None,
|
||
detail: dict[str, Any] | None = None,
|
||
) -> None:
|
||
"""模块事件日志只落盘，不再写 module_generation_events 表。"""
|
||
_ = db
|
||
log_module_event_file(
|
||
module=project.module,
|
||
event_type=event_type,
|
||
project_id=project.id,
|
||
step_id=step.id if step else None,
|
||
user_id=project.user_id,
|
||
message=message,
|
||
detail=detail,
|
||
)
|
||
|
||
|
||
|
||
def _log_project_error(
|
||
*,
|
||
project: ModuleGenerationProject | None,
|
||
event_type: str,
|
||
message: str,
|
||
exc: BaseException | None = None,
|
||
step: ModuleGenerationStep | None = None,
|
||
detail: dict[str, Any] | None = None,
|
||
) -> None:
|
||
log_module_error(
|
||
module=(project.module if project else MODULE),
|
||
event_type=event_type,
|
||
project_id=(project.id if project else None),
|
||
step_id=(step.id if step else None),
|
||
user_id=(project.user_id if project else None),
|
||
message=message,
|
||
detail=detail,
|
||
exc=exc,
|
||
)
|
||
|
||
|
||
async def _get_project_for_user(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
user: 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),
|
||
)
|
||
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(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
step_id: str,
|
||
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,
|
||
)
|
||
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())
|
||
|
||
|
||
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()
|
||
|
||
|
||
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
|
||
|
||
|
||
async def _create_step(
|
||
db: AsyncSession,
|
||
*,
|
||
project: ModuleGenerationProject,
|
||
step_code: str,
|
||
status: str = ModuleStepStatusEnum.PENDING.value,
|
||
parent_step_id: str | None = None,
|
||
source_step_id: str | None = None,
|
||
chat_task_id: str | None = None,
|
||
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],
|
||
step_code=step_code,
|
||
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,
|
||
)
|
||
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(
|
||
db: AsyncSession,
|
||
*,
|
||
project: ModuleGenerationProject,
|
||
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()
|
||
)
|
||
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:
|
||
return ShotReplicateStepOut(
|
||
id=step.id,
|
||
project_id=step.project_id,
|
||
module=step.module,
|
||
step_index=step.step_index,
|
||
step_code=step.step_code,
|
||
status=step.status,
|
||
version=step.version,
|
||
is_current=step.is_current,
|
||
parent_step_id=step.parent_step_id,
|
||
source_step_id=step.source_step_id,
|
||
chat_task_id=step.chat_task_id,
|
||
input=_parse_json(step.input_json, {}),
|
||
output=_parse_json(step.output_json, {}),
|
||
error_message=step.error_message,
|
||
created_at=step.created_at,
|
||
updated_at=step.updated_at,
|
||
completed_at=step.completed_at,
|
||
)
|
||
|
||
|
||
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 {}
|
||
|
||
|
||
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()}
|
||
|
||
|
||
async def project_to_detail_out(db: AsyncSession, project: ModuleGenerationProject) -> ShotReplicateTaskDetailOut:
|
||
steps = await _get_current_steps(db, project.id)
|
||
by_code = {step.step_code: step for step in steps}
|
||
chats = await _chat_tasks_by_id(db, steps)
|
||
|
||
material_step = by_code.get(ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
|
||
image_prompt_step = by_code.get(ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
|
||
image_generate_step = by_code.get(ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
|
||
video_prompt_step = by_code.get(ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
|
||
video_generate_step = by_code.get(ShotReplicateStepCodeEnum.VIDEO_GENERATE.value)
|
||
|
||
material_input = _step_payload(material_step.input_json if material_step else None)
|
||
image_prompt_output = _unwrap_step_output(image_prompt_step.output_json if image_prompt_step else None)
|
||
image_generate_input = _step_payload(image_generate_step.input_json if image_generate_step else None)
|
||
image_generate_output = _unwrap_step_output(image_generate_step.output_json if image_generate_step else None)
|
||
video_prompt_input = _step_payload(video_prompt_step.input_json if video_prompt_step else None)
|
||
video_prompt_output = _unwrap_step_output(video_prompt_step.output_json if video_prompt_step else None)
|
||
video_generate_input = _step_payload(video_generate_step.input_json if video_generate_step else None)
|
||
video_generate_output = _unwrap_step_output(video_generate_step.output_json if video_generate_step else None)
|
||
|
||
image_chat = chats.get(image_generate_step.chat_task_id) if image_generate_step and image_generate_step.chat_task_id else None
|
||
video_chat = chats.get(video_generate_step.chat_task_id) if video_generate_step and video_generate_step.chat_task_id else None
|
||
image_snapshot = _snapshot_from_chat(image_chat)
|
||
video_snapshot = _snapshot_from_chat(video_chat)
|
||
|
||
image_url = image_generate_output.get("result_image_url") or (image_chat.image_url if image_chat else None) or project.final_image_url
|
||
video_url = video_generate_output.get("result_video_url") or (video_chat.video_url if video_chat else None) or project.final_video_url
|
||
cover_url = video_generate_output.get("result_video_cover_url") or (video_chat.video_cover_url if video_chat else None) or project.final_video_cover_url
|
||
|
||
user_name: str | None = None
|
||
if project.user_id:
|
||
user_result = await db.execute(select(User.username).where(User.id == project.user_id).limit(1))
|
||
user_name = user_result.scalar_one_or_none()
|
||
|
||
return ShotReplicateTaskDetailOut(
|
||
id=project.id,
|
||
project_id=project.id,
|
||
user_id=project.user_id,
|
||
user_name=user_name,
|
||
module=project.module,
|
||
title=project.title,
|
||
status=project.status,
|
||
current_step_code=project.current_step_code,
|
||
final_image_url=build_resource_signed_url(project.final_image_url) if project.final_image_url else None,
|
||
final_video_url=build_resource_signed_url(project.final_video_url) if project.final_video_url else None,
|
||
final_video_cover_url=build_resource_signed_url(project.final_video_cover_url) if project.final_video_cover_url else None,
|
||
error_message=project.error_message,
|
||
material=ShotReplicateMaterialOut(
|
||
material_step_id=material_step.id if material_step else None,
|
||
material_video_url=material_input.get("material_video_url"),
|
||
material_image_url=material_input.get("material_image_url"),
|
||
source_project_name=material_input.get("source_project_name"),
|
||
target_project_name=material_input.get("target_project_name"),
|
||
core_content_point=material_input.get("core_content_point"),
|
||
),
|
||
image_generation=ShotReplicateImageGenerationOut(
|
||
prompt_step_id=image_prompt_step.id if image_prompt_step else None,
|
||
generate_step_id=image_generate_step.id if image_generate_step else None,
|
||
prompt=image_prompt_output.get("optimized_prompt") or image_prompt_output.get("prompt"),
|
||
engine_id=image_snapshot.get("id") or image_generate_input.get("engine_id"),
|
||
engine_name=image_snapshot.get("name") or image_generate_input.get("engine_name"),
|
||
params=image_generate_input.get("params") or image_generate_input,
|
||
chat_task_id=image_generate_step.chat_task_id if image_generate_step else None,
|
||
status=image_chat.status if image_chat else (image_generate_step.status if image_generate_step else None),
|
||
result_image_url=build_resource_signed_url(image_url) if image_url else None,
|
||
error_message=image_chat.error_message if image_chat else (image_generate_step.error_message if image_generate_step else None),
|
||
),
|
||
video_generation=ShotReplicateVideoGenerationOut(
|
||
prompt_step_id=video_prompt_step.id if video_prompt_step else None,
|
||
generate_step_id=video_generate_step.id if video_generate_step else None,
|
||
prompt_schema=video_prompt_output.get("prompt_schema"),
|
||
final_prompt=video_prompt_output.get("final_prompt"),
|
||
prompt_params=video_prompt_output.get("params_used_for_prompt") or video_prompt_input.get("video_config"),
|
||
engine_id=video_snapshot.get("id") or video_generate_input.get("engine_id"),
|
||
engine_name=video_snapshot.get("name") or video_generate_input.get("engine_name"),
|
||
params=video_generate_input.get("params") or video_generate_input,
|
||
chat_task_id=video_generate_step.chat_task_id if video_generate_step else None,
|
||
status=video_chat.status if video_chat else (video_generate_step.status if video_generate_step else None),
|
||
result_video_url=build_resource_signed_url(video_url) if video_url else None,
|
||
result_video_cover_url=build_resource_signed_url(cover_url) if cover_url else None,
|
||
error_message=video_chat.error_message if video_chat else (video_generate_step.error_message if video_generate_step else None),
|
||
),
|
||
steps=[_step_to_out(step) for step in steps],
|
||
created_at=project.created_at,
|
||
updated_at=project.updated_at,
|
||
completed_at=project.completed_at,
|
||
)
|
||
|
||
|
||
async def create_shot_replicate_project(db: AsyncSession, current_user: User, req: ShotReplicateTaskCreate) -> ModuleGenerationProject:
|
||
if req.idempotency_key:
|
||
result = await db.execute(
|
||
select(ModuleGenerationProject)
|
||
.where(
|
||
ModuleGenerationProject.user_id == current_user.id,
|
||
ModuleGenerationProject.module == MODULE,
|
||
ModuleGenerationProject.idempotency_key == req.idempotency_key,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
)
|
||
.order_by(ModuleGenerationProject.created_at.desc())
|
||
.limit(1)
|
||
)
|
||
existing = result.scalar_one_or_none()
|
||
if existing:
|
||
return existing
|
||
|
||
project = ModuleGenerationProject(
|
||
id=generate_id(),
|
||
user_id=current_user.id,
|
||
module=MODULE,
|
||
title=req.target_project_name,
|
||
status=ModuleProjectStatusEnum.WAITING_USER.value,
|
||
current_step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
db.add(project)
|
||
await db.flush()
|
||
|
||
await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
|
||
status=ModuleStepStatusEnum.COMPLETED.value,
|
||
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,
|
||
},
|
||
output_data={"message": "素材输入已提交，后端不做素材文件校验。下一步请手动生成图片AI提词。"},
|
||
)
|
||
await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_CREATED.value, message="创建拆镜复刻项目")
|
||
return project
|
||
|
||
|
||
async def _current_step_map_by_project_ids(
|
||
db: AsyncSession,
|
||
*,
|
||
project_ids: set[str],
|
||
step_code: str,
|
||
) -> dict[str, ModuleGenerationStep]:
|
||
"""一次性查询当前页项目的指定步骤，避免列表逐条查 material_input。"""
|
||
if not project_ids:
|
||
return {}
|
||
result = await db.execute(
|
||
select(ModuleGenerationStep).where(
|
||
ModuleGenerationStep.project_id.in_(list(project_ids)),
|
||
ModuleGenerationStep.step_code == step_code,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
)
|
||
return {step.project_id: step for step in result.scalars().all()}
|
||
|
||
|
||
async def list_shot_replicate_projects(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
status: str | None,
|
||
page: int,
|
||
page_size: int,
|
||
) -> ShotReplicateTaskListOut:
|
||
"""
|
||
拆镜复刻项目列表查询。
|
||
|
||
这个列表目前不是后台主入口，但仍避免逐条查 material_input，保持和爆款开头列表一致的批量查询策略。
|
||
"""
|
||
query = select(ModuleGenerationProject).where(
|
||
ModuleGenerationProject.module == MODULE,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
)
|
||
if not current_user.is_admin:
|
||
query = query.where(ModuleGenerationProject.user_id == current_user.id)
|
||
if status:
|
||
query = query.where(ModuleGenerationProject.status == status)
|
||
|
||
total = int((await db.execute(select(func.count()).select_from(query.subquery()))).scalar() or 0)
|
||
result = await db.execute(
|
||
query.order_by(ModuleGenerationProject.created_at.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
)
|
||
projects = list(result.scalars().unique().all())
|
||
project_ids = {project.id for project in projects if project.id}
|
||
material_step_map = await _current_step_map_by_project_ids(
|
||
db,
|
||
project_ids=project_ids,
|
||
step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
|
||
)
|
||
|
||
items: list[ShotReplicateTaskListItemOut] = []
|
||
for project in projects:
|
||
material_step = material_step_map.get(project.id)
|
||
material = _step_payload(material_step.input_json if material_step else None)
|
||
items.append(
|
||
ShotReplicateTaskListItemOut(
|
||
id=project.id,
|
||
project_id=project.id,
|
||
module=project.module,
|
||
title=project.title,
|
||
status=project.status,
|
||
current_step_code=project.current_step_code,
|
||
target_project_name=material.get("target_project_name"),
|
||
final_image_url=build_resource_signed_url(project.final_image_url) if project.final_image_url else None,
|
||
final_video_url=build_resource_signed_url(project.final_video_url) if project.final_video_url else None,
|
||
error_message=project.error_message,
|
||
created_at=project.created_at,
|
||
updated_at=project.updated_at,
|
||
completed_at=project.completed_at,
|
||
)
|
||
)
|
||
return ShotReplicateTaskListOut(total=total, items=items)
|
||
|
||
|
||
async def update_shot_replicate_step(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
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,
|
||
),
|
||
)
|
||
_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(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
req: ShotReplicateMaterialUpdateRequest,
|
||
) -> tuple[str, str]:
|
||
"""修改第1步素材输入。
|
||
|
||
采用方案 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(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
|
||
status=ModuleStepStatusEnum.COMPLETED.value,
|
||
input_data=material,
|
||
output_data={"message": "素材输入已修改，旧步骤已软删除。下一步请重新生成图片AI提词。"},
|
||
)
|
||
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(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
step_id: str,
|
||
req: ShotReplicateImagePromptUpdateRequest,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
|
||
"""直接修改第2步图片 AI 优化提词，不调用 AI、不扣积分。
|
||
|
||
修改后软删除第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(
|
||
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]},
|
||
)
|
||
await db.flush()
|
||
return project, step
|
||
|
||
|
||
async def update_shot_replicate_video_prompt_schema(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
step_id: str,
|
||
req: ShotReplicateVideoPromptSchemaUpdateRequest,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
|
||
"""以前端 schema 为 patch 修改第4步视频 AI 提词，不调用 AI、不扣积分。
|
||
|
||
服务端已有 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(
|
||
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",
|
||
"画面属性.视频时长",
|
||
"画面属性.视频比例",
|
||
"画面属性.清晰度",
|
||
"画面属性.帧率",
|
||
"画面属性.推荐分辨率",
|
||
"动作流程[*].时间段",
|
||
"镜头流程[*].时间段",
|
||
"动态时间规划",
|
||
"输出规格限制",
|
||
"质量控制",
|
||
"合规控制",
|
||
],
|
||
},
|
||
)
|
||
await db.flush()
|
||
return project, step
|
||
|
||
|
||
async def submit_image_prompt_optimize(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
material_step_id: str | None = None,
|
||
req: Any | None = None,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
|
||
_ = req
|
||
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
||
if material_step_id:
|
||
material_step = await _get_step_for_user(db, project_id=project_id, step_id=material_step_id, user=current_user, for_update=True)
|
||
else:
|
||
material_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
|
||
if not material_step:
|
||
raise HTTPException(status_code=400, detail="缺少第1步素材输入子任务")
|
||
if material_step.step_code != ShotReplicateStepCodeEnum.MATERIAL_INPUT.value:
|
||
raise HTTPException(status_code=400, detail="请基于第1步素材输入子任务生成图片 AI 提词")
|
||
if material_step.status != ModuleStepStatusEnum.COMPLETED.value:
|
||
raise HTTPException(status_code=400, detail="素材输入子任务未完成，不能生成图片 AI 提词")
|
||
|
||
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value])
|
||
step = await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=material_step.id,
|
||
source_step_id=material_step.id,
|
||
input_data={"source_step_id": material_step.id},
|
||
)
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
|
||
project.error_message = None
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_PROMPT_SUBMITTED.value, message="图片 AI 提词任务已提交")
|
||
return project, step
|
||
|
||
|
||
async def run_image_prompt_optimize(db: AsyncSession, *, project_id: str, step_id: str | None = None) -> ModuleGenerationStep | None:
|
||
project_result = await db.execute(
|
||
select(ModuleGenerationProject)
|
||
.where(ModuleGenerationProject.id == project_id, ModuleGenerationProject.module == MODULE, ModuleGenerationProject.deleted_at.is_(None))
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
project = project_result.scalar_one_or_none()
|
||
if not project:
|
||
return None
|
||
|
||
material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
|
||
if not material_step:
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = "缺少素材输入子任务"
|
||
return None
|
||
|
||
if step_id:
|
||
result = await db.execute(
|
||
select(ModuleGenerationStep)
|
||
.where(
|
||
ModuleGenerationStep.id == step_id,
|
||
ModuleGenerationStep.project_id == project.id,
|
||
ModuleGenerationStep.step_code == ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
ModuleGenerationStep.is_current == True,
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
step = result.scalar_one_or_none()
|
||
else:
|
||
step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
|
||
if not step:
|
||
if step_id:
|
||
# 用户重复提交后，旧 Celery 消息对应的 step 可能已被软删。
|
||
# 指定 step_id 查不到时必须静默忽略，不能重新创建步骤导致旧任务复活。
|
||
return None
|
||
step = await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=material_step.id,
|
||
source_step_id=material_step.id,
|
||
input_data={"source_step_id": material_step.id},
|
||
)
|
||
else:
|
||
step.status = ModuleStepStatusEnum.PROCESSING.value
|
||
step.started_at = _now()
|
||
step.error_message = None
|
||
|
||
material = _step_payload(material_step.input_json)
|
||
prompt_text = (
|
||
"请基于参考素材复刻拆镜视觉风格，用于生成新项目图片。\n"
|
||
f"视频素材内容项目名称：{material.get('source_project_name')}\n"
|
||
f"生成项目名称：{material.get('target_project_name')}\n"
|
||
f"生成项目核心内容点：{material.get('core_content_point')}\n"
|
||
"要求：参考素材视频的开头构图、主体位置、节奏和风格；结合新产品图片生成新项目推广图片；不要照抄原素材品牌、文字、水印；适合作为后续图生视频首帧。"
|
||
)
|
||
references = [
|
||
{"type": "video", "url": material.get("material_video_url"), "name": "参考素材视频"},
|
||
{"type": "image", "url": material.get("material_image_url"), "name": "新产品图片"},
|
||
]
|
||
try:
|
||
request_log = {"original_prompt": prompt_text, "references": references, "gen_type": "image"}
|
||
log_module_prompt_event(
|
||
event_type="module_prompt_request",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
module=project.module,
|
||
prompt_type=ModulePromptTypeEnum.IMAGE_PROMPT.value,
|
||
request=request_log,
|
||
)
|
||
optimized, token_usage = await optimize_prompt(
|
||
db,
|
||
original_prompt=prompt_text,
|
||
user_id=project.user_id,
|
||
references=references,
|
||
gen_type="image",
|
||
)
|
||
billing = await charge_module_prompt_usage(
|
||
db,
|
||
user_id=project.user_id,
|
||
step_id=step.id,
|
||
usage=token_usage,
|
||
description="拆镜复刻-图片AI提词优化",
|
||
)
|
||
usage = dict(token_usage or {})
|
||
usage.update({
|
||
"text_credits_cost": (billing.items[0].amount if billing.items else billing.total_charged),
|
||
"credit_biz_key": billing.items[0].biz_key if billing.items else None,
|
||
})
|
||
step.status = ModuleStepStatusEnum.COMPLETED.value
|
||
step.completed_at = _now()
|
||
_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
|
||
project.error_message = None
|
||
log_module_prompt_event(
|
||
event_type="module_prompt_response",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
module=project.module,
|
||
prompt_type=ModulePromptTypeEnum.IMAGE_PROMPT.value,
|
||
request=request_log,
|
||
response={"optimized_prompt": optimized},
|
||
token_usage=usage,
|
||
)
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_PROMPT_SUCCESS.value, message="图片 AI 提词生成成功")
|
||
except Exception as exc:
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = str(exc)
|
||
step.completed_at = _now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = f"图片 AI 提词生成失败: {exc}"
|
||
log_module_prompt_event(
|
||
event_type="module_prompt_error",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
module=project.module,
|
||
prompt_type=ModulePromptTypeEnum.IMAGE_PROMPT.value,
|
||
request=locals().get("request_log", {}),
|
||
error=str(exc),
|
||
)
|
||
_log_project_error(project=project, step=step, event_type="IMAGE_PROMPT_FAILED", message=project.error_message, exc=exc)
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_PROMPT_FAILED.value, message=project.error_message)
|
||
return step
|
||
|
||
|
||
async def generate_image_from_prompt(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
req: ShotReplicateGenerateImageRequest,
|
||
prompt_step_id: str | None = None,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep, ChatGenerationTask]:
|
||
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
||
if prompt_step_id:
|
||
prompt_step = await _get_step_for_user(db, project_id=project_id, step_id=prompt_step_id, user=current_user, for_update=True)
|
||
else:
|
||
prompt_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
|
||
if not prompt_step:
|
||
raise HTTPException(status_code=400, detail="缺少第2步图片 AI 提词子任务")
|
||
if prompt_step.step_code != ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
|
||
raise HTTPException(status_code=400, detail="请基于第2步图片 AI 提词子任务生成图片")
|
||
if prompt_step.status != ModuleStepStatusEnum.COMPLETED.value:
|
||
raise HTTPException(status_code=400, detail="图片 AI 提词未完成，不能生成图片")
|
||
|
||
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_GENERATE.value])
|
||
|
||
material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
|
||
material = _step_payload(material_step.input_json if material_step else None)
|
||
prompt_output = _unwrap_step_output(prompt_step.output_json)
|
||
optimized_prompt = prompt_output.get("optimized_prompt") or prompt_output.get("prompt") or ""
|
||
refs = [
|
||
{"type": "image", "url": material.get("material_image_url"), "name": "新产品图片"},
|
||
]
|
||
|
||
chat_task = await create_chat_generation_task_for_module(
|
||
db,
|
||
current_user=current_user,
|
||
generation_mode=GENERATION_MODE,
|
||
gen_type="image",
|
||
original_prompt=prompt_output.get("original_prompt") or optimized_prompt,
|
||
optimized_prompt=optimized_prompt,
|
||
engine_id=req.engine_id,
|
||
media_references=refs,
|
||
image_size=req.image_size,
|
||
image_proportion=req.image_proportion,
|
||
image_px=req.image_px,
|
||
billing_project_name=project.title or "拆镜复刻",
|
||
billing_description_prefix="拆镜复刻图片生成",
|
||
)
|
||
step = await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=prompt_step.id,
|
||
source_step_id=prompt_step.id,
|
||
chat_task_id=chat_task.id,
|
||
input_data={
|
||
"engine_id": chat_task.engine_id,
|
||
"params": {"image_size": chat_task.image_size, "image_proportion": chat_task.image_proportion, "image_px": chat_task.image_px},
|
||
"prompt": optimized_prompt,
|
||
"media_references": refs,
|
||
},
|
||
)
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_GENERATE.value
|
||
project.error_message = None
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_GENERATE_SUBMITTED.value, message="图片生成任务已提交", detail={"chat_task_id": chat_task.id})
|
||
return project, step, chat_task
|
||
|
||
|
||
async def _resolve_video_prompt_config(db: AsyncSession, req: ShotReplicateGenerateVideoPromptRequest) -> dict[str, Any]:
|
||
engine = await _get_video_engine(db, req.engine_id)
|
||
supported_ratios = _parse_list(engine.supported_ratios, [])
|
||
supported_resolutions = _parse_list(engine.supported_resolutions, [])
|
||
supported_durations = _parse_list(engine.supported_durations, [])
|
||
|
||
default_ratio = getattr(settings, "SHOT_REPLICATE_DEFAULT_VIDEO_RATIO", None) or VIDEO_DEFAULT_RATIO
|
||
default_resolution = getattr(settings, "SHOT_REPLICATE_DEFAULT_VIDEO_RESOLUTION", None) or VIDEO_DEFAULT_RESOLUTION
|
||
default_duration = int(getattr(settings, "SHOT_REPLICATE_DEFAULT_VIDEO_DURATION", None) or VIDEO_DEFAULT_DURATION)
|
||
|
||
selected_ratio = req.aspect_ratio or (default_ratio if not supported_ratios or default_ratio in supported_ratios else supported_ratios[0])
|
||
selected_resolution = req.resolution or (default_resolution if not supported_resolutions or default_resolution in supported_resolutions else supported_resolutions[0])
|
||
selected_duration = req.duration or (default_duration if not supported_durations or default_duration in supported_durations else supported_durations[0])
|
||
|
||
if supported_ratios and selected_ratio not in supported_ratios:
|
||
raise HTTPException(status_code=400, detail=f"视频比例不支持: {selected_ratio}")
|
||
if supported_resolutions and selected_resolution not in supported_resolutions:
|
||
raise HTTPException(status_code=400, detail=f"视频分辨率不支持: {selected_resolution}")
|
||
if supported_durations and selected_duration not in supported_durations:
|
||
raise HTTPException(status_code=400, detail=f"视频时长不支持: {selected_duration}")
|
||
if engine.max_duration and int(selected_duration) > int(engine.max_duration):
|
||
raise HTTPException(status_code=400, detail=f"视频时长不能超过 {engine.max_duration} 秒")
|
||
|
||
return {
|
||
"engine_id": engine.id,
|
||
"engine_name": engine.name,
|
||
"duration": int(selected_duration),
|
||
"aspect_ratio": selected_ratio,
|
||
"resolution": selected_resolution,
|
||
"supported_ratios": supported_ratios,
|
||
"supported_resolutions": supported_resolutions,
|
||
"supported_durations": supported_durations,
|
||
"max_duration": engine.max_duration,
|
||
"frame_rate": "30fps",
|
||
"reference_video_fps": max(1, int(settings.CHATAPI_VIDEO_FPS or 1)),
|
||
}
|
||
|
||
|
||
async def submit_video_prompt_optimize(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
req: ShotReplicateGenerateVideoPromptRequest,
|
||
image_step_id: str | None = None,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
|
||
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
||
if image_step_id:
|
||
image_step = await _get_step_for_user(db, project_id=project_id, step_id=image_step_id, user=current_user, for_update=True)
|
||
else:
|
||
image_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
|
||
if not image_step:
|
||
raise HTTPException(status_code=400, detail="缺少第3步图片生成子任务")
|
||
if image_step.step_code != ShotReplicateStepCodeEnum.IMAGE_GENERATE.value:
|
||
raise HTTPException(status_code=400, detail="请基于第3步图片生成子任务生成视频 AI 提词")
|
||
if image_step.status != ModuleStepStatusEnum.COMPLETED.value:
|
||
raise HTTPException(status_code=400, detail="图片生成子任务未完成，不能生成视频 AI 提词")
|
||
|
||
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value])
|
||
video_config = await _resolve_video_prompt_config(db, req)
|
||
step = await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=image_step.id,
|
||
source_step_id=image_step.id,
|
||
input_data={
|
||
"source_step_id": image_step.id,
|
||
"video_config": video_config,
|
||
"target_platform": req.target_platform or getattr(settings, "SHOT_REPLICATE_DEFAULT_TARGET_PLATFORM", "抖音") or "抖音",
|
||
},
|
||
)
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value
|
||
project.error_message = None
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_SUBMITTED.value, message="视频 AI 提词任务已提交")
|
||
return project, step
|
||
|
||
|
||
async def run_video_prompt_optimize(db: AsyncSession, *, project_id: str, step_id: str | None = None) -> ModuleGenerationStep | None:
|
||
project_result = await db.execute(
|
||
select(ModuleGenerationProject)
|
||
.where(ModuleGenerationProject.id == project_id, ModuleGenerationProject.module == MODULE, ModuleGenerationProject.deleted_at.is_(None))
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
project = project_result.scalar_one_or_none()
|
||
if not project:
|
||
return None
|
||
|
||
material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
|
||
image_prompt_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
|
||
image_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
|
||
if not material_step or not image_step:
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = "生成视频提词失败：缺少素材输入或图片生成结果"
|
||
return None
|
||
|
||
if step_id:
|
||
result = await db.execute(
|
||
select(ModuleGenerationStep)
|
||
.where(
|
||
ModuleGenerationStep.id == step_id,
|
||
ModuleGenerationStep.project_id == project.id,
|
||
ModuleGenerationStep.step_code == ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
ModuleGenerationStep.is_current == True,
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
step = result.scalar_one_or_none()
|
||
else:
|
||
step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
|
||
if not step:
|
||
if step_id:
|
||
# 用户重复提交后，旧 Celery 消息对应的 step 可能已被软删。
|
||
# 指定 step_id 查不到时必须静默忽略，不能把当前项目标记失败。
|
||
return None
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = "缺少视频 AI 提词子任务，请先手动提交视频提词生成"
|
||
return None
|
||
|
||
step.status = ModuleStepStatusEnum.PROCESSING.value
|
||
step.started_at = _now()
|
||
step.error_message = None
|
||
|
||
material = _step_payload(material_step.input_json)
|
||
image_output = _unwrap_step_output(image_step.output_json)
|
||
step_input = _step_payload(step.input_json)
|
||
video_config = step_input.get("video_config") or {}
|
||
target_platform = step_input.get("target_platform") or getattr(settings, "SHOT_REPLICATE_DEFAULT_TARGET_PLATFORM", "抖音") or "抖音"
|
||
generated_image_url = image_output.get("result_image_url") or project.final_image_url
|
||
if not generated_image_url:
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = "缺少新项目图片结果，不能生成视频提词"
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = step.error_message
|
||
return step
|
||
|
||
try:
|
||
request_log = {
|
||
"source_project_name": material.get("source_project_name") or "无",
|
||
"target_project_name": material.get("target_project_name") or "无",
|
||
"core_content_point": material.get("core_content_point") or "无",
|
||
"material_video_url": material.get("material_video_url") or "",
|
||
"generated_image_url": generated_image_url,
|
||
"video_config": video_config,
|
||
"target_platform": target_platform,
|
||
}
|
||
log_module_prompt_event(
|
||
event_type="module_prompt_request",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
module=project.module,
|
||
prompt_type=ModulePromptTypeEnum.VIDEO_PROMPT.value,
|
||
request=request_log,
|
||
)
|
||
prompt_schema, final_prompt, token_usage = await optimize_shot_replicate_video_prompt(
|
||
db,
|
||
user_id=project.user_id,
|
||
source_project_name=request_log["source_project_name"],
|
||
target_project_name=request_log["target_project_name"],
|
||
core_content_point=request_log["core_content_point"],
|
||
material_video_url=request_log["material_video_url"],
|
||
generated_image_url=generated_image_url,
|
||
video_config=video_config,
|
||
target_platform=target_platform,
|
||
)
|
||
billing = await charge_module_prompt_usage(
|
||
db,
|
||
user_id=project.user_id,
|
||
step_id=step.id,
|
||
usage=token_usage,
|
||
description="拆镜复刻-视频AI提词优化",
|
||
)
|
||
usage = dict(token_usage or {})
|
||
usage.update({
|
||
"text_credits_cost": (billing.items[0].amount if billing.items else billing.total_charged),
|
||
"credit_biz_key": billing.items[0].biz_key if billing.items else None,
|
||
})
|
||
step.status = ModuleStepStatusEnum.COMPLETED.value
|
||
step.completed_at = _now()
|
||
_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
|
||
project.error_message = None
|
||
log_module_prompt_event(
|
||
event_type="module_prompt_response",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
module=project.module,
|
||
prompt_type=ModulePromptTypeEnum.VIDEO_PROMPT.value,
|
||
request=request_log,
|
||
response={"prompt_schema": prompt_schema, "final_prompt": final_prompt},
|
||
token_usage=usage,
|
||
)
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_SUCCESS.value, message="视频 AI 提词生成成功")
|
||
except Exception as exc:
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = str(exc)
|
||
step.completed_at = _now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = f"视频 AI 提词生成失败: {exc}"
|
||
log_module_prompt_event(
|
||
event_type="module_prompt_error",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
module=project.module,
|
||
prompt_type=ModulePromptTypeEnum.VIDEO_PROMPT.value,
|
||
request=locals().get("request_log", {}),
|
||
error=str(exc),
|
||
)
|
||
_log_project_error(project=project, step=step, event_type="VIDEO_PROMPT_FAILED", message=project.error_message, exc=exc)
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_FAILED.value, message=project.error_message)
|
||
return step
|
||
|
||
|
||
async def generate_video_from_prompt(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
req: ShotReplicateGenerateVideoRequest,
|
||
prompt_step_id: str | None = None,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep, ChatGenerationTask]:
|
||
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
||
if prompt_step_id:
|
||
prompt_step = await _get_step_for_user(db, project_id=project_id, step_id=prompt_step_id, user=current_user, for_update=True)
|
||
else:
|
||
prompt_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
|
||
if not prompt_step:
|
||
raise HTTPException(status_code=400, detail="缺少第4步视频 AI 提词子任务")
|
||
if prompt_step.step_code != ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
|
||
raise HTTPException(status_code=400, detail="请基于第4步视频 AI 提词子任务生成视频")
|
||
if prompt_step.status != ModuleStepStatusEnum.COMPLETED.value:
|
||
raise HTTPException(status_code=400, detail="视频 AI 提词未完成，不能生成视频")
|
||
|
||
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.VIDEO_GENERATE.value])
|
||
|
||
image_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
|
||
image_output = _unwrap_step_output(image_step.output_json if image_step else None)
|
||
prompt_output = _unwrap_step_output(prompt_step.output_json)
|
||
final_prompt = prompt_output.get("final_prompt") or ""
|
||
prompt_schema = prompt_output.get("prompt_schema") or {}
|
||
prompt_schema_str = json.dumps(prompt_schema, ensure_ascii=False, default=str) if prompt_schema else ""
|
||
prompt_input = _step_payload(prompt_step.input_json)
|
||
prompt_params = prompt_output.get("params_used_for_prompt") or prompt_input.get("video_config") or {}
|
||
duration = int(prompt_params.get("duration") or settings.SHOT_REPLICATE_DEFAULT_VIDEO_DURATION or 4)
|
||
aspect_ratio = prompt_params.get("aspect_ratio") or settings.SHOT_REPLICATE_DEFAULT_VIDEO_RATIO or "9:16"
|
||
resolution = prompt_params.get("resolution") or settings.SHOT_REPLICATE_DEFAULT_VIDEO_RESOLUTION or "480p"
|
||
generated_image_url = image_output.get("result_image_url") or project.final_image_url
|
||
if not generated_image_url:
|
||
raise HTTPException(status_code=400, detail="缺少新项目图片结果，不能生成视频")
|
||
|
||
refs = [
|
||
{"type": "image", "url": _build_file_url_or_data_uri(generated_image_url), "name": "新项目图片"},
|
||
]
|
||
|
||
chat_task = await create_chat_generation_task_for_module(
|
||
db,
|
||
current_user=current_user,
|
||
generation_mode=GENERATION_MODE,
|
||
gen_type="video",
|
||
original_prompt=prompt_schema_str or final_prompt,
|
||
optimized_prompt=prompt_schema_str or final_prompt,
|
||
engine_id=req.engine_id or prompt_params.get("engine_id"),
|
||
media_references=refs,
|
||
duration=duration,
|
||
aspect_ratio=aspect_ratio,
|
||
resolution=resolution,
|
||
billing_project_name=project.title or "拆镜复刻",
|
||
billing_description_prefix="拆镜复刻视频生成",
|
||
)
|
||
step = await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=prompt_step.id,
|
||
source_step_id=prompt_step.id,
|
||
chat_task_id=chat_task.id,
|
||
input_data={
|
||
"engine_id": chat_task.engine_id,
|
||
"params": {
|
||
"duration": chat_task.duration,
|
||
"aspect_ratio": chat_task.aspect_ratio,
|
||
"resolution": chat_task.resolution,
|
||
"image_size": chat_task.image_size,
|
||
"image_proportion": chat_task.image_proportion,
|
||
"image_px": chat_task.image_px,
|
||
},
|
||
"prompt_schema": prompt_schema,
|
||
"final_prompt": final_prompt,
|
||
"media_references": refs,
|
||
},
|
||
)
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_GENERATE.value
|
||
project.error_message = None
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_GENERATE_SUBMITTED.value, message="视频生成任务已提交", detail={"chat_task_id": chat_task.id})
|
||
return project, step, chat_task
|
||
|
||
|
||
async def handle_chat_generation_task_completed(db: AsyncSession, task: ChatGenerationTask) -> None:
|
||
if not task or task.generation_mode != GENERATION_MODE:
|
||
return
|
||
result = await db.execute(
|
||
select(ModuleGenerationStep)
|
||
.where(
|
||
ModuleGenerationStep.chat_task_id == task.id,
|
||
ModuleGenerationStep.module == MODULE,
|
||
ModuleGenerationStep.is_current == True,
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
step = result.scalar_one_or_none()
|
||
if not step:
|
||
return
|
||
project_result = await db.execute(
|
||
select(ModuleGenerationProject)
|
||
.where(ModuleGenerationProject.id == step.project_id, ModuleGenerationProject.deleted_at.is_(None))
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
project = project_result.scalar_one_or_none()
|
||
if not project:
|
||
return
|
||
|
||
if step.step_code == ShotReplicateStepCodeEnum.IMAGE_GENERATE.value:
|
||
step.status = ModuleStepStatusEnum.COMPLETED.value
|
||
step.completed_at = _now()
|
||
_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
|
||
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_GENERATE.value
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_GENERATE_SUCCESS.value, message="图片生成完成，等待用户手动生成视频 AI 提词")
|
||
elif step.step_code == ShotReplicateStepCodeEnum.VIDEO_GENERATE.value:
|
||
step.status = ModuleStepStatusEnum.COMPLETED.value
|
||
step.completed_at = _now()
|
||
_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
|
||
project.status = ModuleProjectStatusEnum.COMPLETED.value
|
||
project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_GENERATE.value
|
||
project.completed_at = _now()
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_GENERATE_SUCCESS.value, message="视频生成完成，总任务完成")
|
||
|
||
|
||
async def handle_chat_generation_task_failed(db: AsyncSession, task: ChatGenerationTask) -> None:
|
||
if not task or task.generation_mode != GENERATION_MODE:
|
||
return
|
||
result = await db.execute(
|
||
select(ModuleGenerationStep)
|
||
.where(ModuleGenerationStep.chat_task_id == task.id, ModuleGenerationStep.module == MODULE, ModuleGenerationStep.is_current == True, ModuleGenerationStep.deleted_at.is_(None))
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
step = result.scalar_one_or_none()
|
||
if not step:
|
||
return
|
||
project_result = await db.execute(select(ModuleGenerationProject).where(ModuleGenerationProject.id == step.project_id).with_for_update().limit(1))
|
||
project = project_result.scalar_one_or_none()
|
||
if not project:
|
||
return
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = task.error_message
|
||
step.completed_at = _now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = task.error_message or "生成失败"
|
||
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.CHAT_TASK_FAILED.value, message=project.error_message, detail={"chat_task_id": task.id})
|
||
|
||
|
||
async def mark_shot_replicate_step_dispatch_failed(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
project_id: str,
|
||
step_id: str,
|
||
error_message: str,
|
||
) -> None:
|
||
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
||
result = await db.execute(
|
||
select(ModuleGenerationStep)
|
||
.where(
|
||
ModuleGenerationStep.id == step_id,
|
||
ModuleGenerationStep.project_id == project.id,
|
||
ModuleGenerationStep.module == MODULE,
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
ModuleGenerationStep.is_current == True,
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
step = result.scalar_one_or_none()
|
||
if not step:
|
||
return
|
||
if step.chat_task_id:
|
||
await mark_chat_generation_task_failed_and_refund_once(
|
||
db,
|
||
task_id=step.chat_task_id,
|
||
error_message=error_message,
|
||
pipeline_stage="failed",
|
||
)
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = error_message
|
||
step.completed_at = _now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = error_message
|
||
log_module_error(
|
||
module=project.module,
|
||
event_type="CELERY_DISPATCH_FAILED",
|
||
project_id=project.id,
|
||
step_id=step.id,
|
||
user_id=project.user_id,
|
||
message=error_message,
|
||
detail={"reason": "celery_dispatch_failed", "chat_task_id": step.chat_task_id},
|
||
error=error_message,
|
||
)
|
||
await log_module_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.CHAT_TASK_FAILED.value,
|
||
message=error_message,
|
||
detail={"reason": "celery_dispatch_failed"},
|
||
)
|
||
|
||
|
||
async def delete_shot_replicate_project(db: AsyncSession, *, current_user: User, project_id: str) -> ShotReplicateDeleteOut:
|
||
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
||
deleted_at = _now()
|
||
project.deleted_at = deleted_at
|
||
await _soft_delete_steps_from_index(db, project=project, start_index=1, deleted_at=deleted_at)
|
||
await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_DELETED.value, message="软删除拆镜复刻项目")
|
||
return ShotReplicateDeleteOut(message="项目已删除", project_id=project.id, deleted=True)
|
||
|
||
|
||
async def create_shot_replicate_project_from_segment(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
segment: ShotReplicateSegment,
|
||
req: ShotSegmentReplicationCreateRequest,
|
||
) -> ModuleGenerationProject:
|
||
"""从拆镜片段创建拆镜复刻项目。
|
||
|
||
素材视频固定取 segment.segment_video_url，不允许前端传入或后续修改。
|
||
"""
|
||
if segment.user_id != current_user.id and not current_user.is_admin:
|
||
raise HTTPException(status_code=404, detail="拆镜片段不存在")
|
||
if segment.split_status != ShotSplitStatusEnum.COMPLETED.value or not segment.segment_video_url:
|
||
raise HTTPException(status_code=400, detail="拆镜片段视频未完成，不能进入复刻流程")
|
||
if segment.module_project_id:
|
||
existing = await _get_project_for_user(
|
||
db,
|
||
project_id=segment.module_project_id,
|
||
user=current_user,
|
||
for_update=False,
|
||
)
|
||
return existing
|
||
|
||
if req.idempotency_key:
|
||
existing_result = await db.execute(
|
||
select(ModuleGenerationProject).where(
|
||
ModuleGenerationProject.user_id == current_user.id,
|
||
ModuleGenerationProject.module == MODULE,
|
||
ModuleGenerationProject.idempotency_key == req.idempotency_key,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
).limit(1)
|
||
)
|
||
existing = existing_result.scalar_one_or_none()
|
||
if existing:
|
||
segment.module_project_id = existing.id
|
||
segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
|
||
await db.flush()
|
||
return existing
|
||
|
||
project = ModuleGenerationProject(
|
||
id=generate_id(),
|
||
user_id=current_user.id,
|
||
module=MODULE,
|
||
title=req.target_project_name,
|
||
status=ModuleProjectStatusEnum.WAITING_USER.value,
|
||
current_step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
db.add(project)
|
||
await db.flush()
|
||
|
||
payload = {
|
||
"material_video_url": segment.segment_video_url,
|
||
"material_video_locked": True,
|
||
"material_image_url": req.material_image_url,
|
||
"source_project_name": segment.segment_category or segment.original_video_category or "拆镜片段",
|
||
"target_project_name": req.target_project_name,
|
||
"core_content_point": req.core_content_point,
|
||
"source_shot_task_set_id": segment.task_set_id,
|
||
"source_segment_id": segment.id,
|
||
"source_segment_index": segment.segment_index,
|
||
"segment_time_node": segment.time_node,
|
||
}
|
||
context = {
|
||
"shot_segment": {
|
||
"id": segment.id,
|
||
"task_set_id": segment.task_set_id,
|
||
"segment_index": segment.segment_index,
|
||
"time_node": segment.time_node,
|
||
"start_second": segment.start_second,
|
||
"end_second": segment.end_second,
|
||
"source_mode": segment.source_mode,
|
||
},
|
||
"analysis": {
|
||
"original_video_content": segment.original_video_content,
|
||
"original_video_category": segment.original_video_category,
|
||
"original_video_audience": segment.original_video_audience,
|
||
"segment_content": segment.segment_content,
|
||
"segment_category": segment.segment_category,
|
||
"segment_audience": segment.segment_audience,
|
||
},
|
||
}
|
||
|
||
step = await _create_step(
|
||
db,
|
||
project=project,
|
||
step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
|
||
status=ModuleStepStatusEnum.COMPLETED.value,
|
||
input_data={**payload, "source_context": context},
|
||
output_data={
|
||
"accepted": True,
|
||
"message": "拆镜片段素材输入已提交，素材视频已锁定",
|
||
"next_step_code": ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
||
**context["analysis"],
|
||
},
|
||
)
|
||
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
|
||
project.status = ModuleProjectStatusEnum.WAITING_USER.value
|
||
segment.module_project_id = project.id
|
||
segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
|
||
await log_module_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.PROJECT_CREATED.value,
|
||
message="从拆镜片段创建拆镜复刻项目",
|
||
detail={"segment_id": segment.id, "task_set_id": segment.task_set_id},
|
||
)
|
||
await db.flush()
|
||
return project
|
||
|
||
|
||
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
|