1424 lines
56 KiB
Python
1424 lines
56 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from types import SimpleNamespace
|
||
from typing import Any, Awaitable, Callable
|
||
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.enums.common import (
|
||
LogEventStatusEnum,
|
||
LogSourceEnum,
|
||
ModuleEventTypeEnum,
|
||
ModuleGenerationFlowVersionEnum,
|
||
ModuleProjectStatusEnum,
|
||
ModuleStepStatusEnum,
|
||
)
|
||
from app.enums.credit_record import CreditRecordBillingScene, CreditRecordChargeKind, CreditRecordOwnerType
|
||
from app.enums.generation_task import ChatGenerationTaskStatus
|
||
from app.enums.llm_billing import LlmBillingConfigKey
|
||
from app.enums.shot_replicate import (
|
||
ShotSegmentReplicateStatusEnum,
|
||
ShotSplitStatusEnum,
|
||
)
|
||
from app.enums.upload_resource import (
|
||
UploadResourceModuleEnum,
|
||
UploadResourceSourceModelEnum,
|
||
)
|
||
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.shot_replicate_segment import ShotReplicateSegment
|
||
from app.models.user import User
|
||
from app.schemas.module_generation_v2 import (
|
||
HotOpeningTaskCreateV2,
|
||
ModuleGenerationVideoConfigV2,
|
||
ModuleVideoPromptSchemaUpdateV2,
|
||
ShotReplicateProjectCreateV2,
|
||
)
|
||
from app.services.generation.ai.engine_service import (
|
||
build_video_snapshot,
|
||
get_video_engine,
|
||
parse_json_list,
|
||
)
|
||
from app.services.generation.pipeline.db_lock_service import (
|
||
DatabaseRowLockBusy,
|
||
execute_with_lock_timeout,
|
||
)
|
||
from app.services.generation.task_factory_service import create_chat_generation_task_for_module
|
||
from app.services.llm_billing import (
|
||
LlmBillingContext,
|
||
ensure_hold_exists,
|
||
log_provider_failure,
|
||
log_provider_start,
|
||
log_provider_success,
|
||
release_on_failure,
|
||
settle_success,
|
||
start_hold,
|
||
)
|
||
from app.services.hot_opening_video_prompt_service import (
|
||
build_final_video_prompt,
|
||
optimize_hot_opening_video_prompt,
|
||
patch_video_prompt_schema_from_client,
|
||
)
|
||
from app.services.module_generation_flow_base_service import (
|
||
assert_project_has_no_active_chat_tasks,
|
||
create_module_step,
|
||
get_current_step_by_code,
|
||
soft_delete_steps_from_index,
|
||
)
|
||
from app.services.module_generation_log_service import log_module_error, log_module_event_file
|
||
from app.services.operation_log_service import log_operation_event
|
||
from app.services.module_generation_step_common_service import (
|
||
build_step_output,
|
||
force_set_json,
|
||
step_payload,
|
||
utc_now,
|
||
)
|
||
from app.services.module_generation_step_update_service import update_module_video_prompt_schema
|
||
from app.services.module_generation_v2.config import (
|
||
CONFIG_BY_MODULE,
|
||
HOT_OPENING_V2,
|
||
MATERIAL_INPUT,
|
||
SHOT_REPLICATE_V2,
|
||
VIDEO_GENERATE,
|
||
VIDEO_PROMPT_OPTIMIZE,
|
||
ModuleGenerationV2Config,
|
||
)
|
||
from app.services.module_generation_v2.media_service import (
|
||
assert_v2_video_generation_references,
|
||
build_v2_video_generation_references,
|
||
)
|
||
from app.services.upload_resource import bind_upload_resources, release_upload_resources_by_source
|
||
from app.services.video_prompt_schema_config_service import (
|
||
get_runtime_schema_snapshot,
|
||
)
|
||
from app.utils.id_gen import generate_id
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ModuleProjectCreateV2Result:
|
||
project: ModuleGenerationProject
|
||
prompt_step: ModuleGenerationStep
|
||
created_new: bool
|
||
|
||
|
||
def is_project_idempotency_conflict(exc: Exception) -> bool:
|
||
"""只把模块项目幂等唯一索引冲突视为可回查的并发重放。"""
|
||
original = getattr(exc, "orig", None)
|
||
constraint_name = getattr(getattr(original, "diag", None), "constraint_name", None)
|
||
if constraint_name == "uq_module_generation_projects_user_module_idempotency":
|
||
return True
|
||
return "uq_module_generation_projects_user_module_idempotency" in str(original or exc)
|
||
|
||
|
||
def flow_version_of(project: ModuleGenerationProject) -> str:
|
||
return str(getattr(project, "flow_version", None) or ModuleGenerationFlowVersionEnum.V1.value)
|
||
|
||
|
||
def assert_v2_project(project: ModuleGenerationProject) -> None:
|
||
if flow_version_of(project) != ModuleGenerationFlowVersionEnum.V2.value:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"message": "项目流程版本与 V2 接口不匹配",
|
||
"project_id": project.id,
|
||
"flow_version": flow_version_of(project),
|
||
},
|
||
)
|
||
|
||
|
||
def assert_v1_project(project: ModuleGenerationProject) -> None:
|
||
if flow_version_of(project) != ModuleGenerationFlowVersionEnum.V1.value:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"message": "项目流程版本与 V1 接口不匹配",
|
||
"project_id": project.id,
|
||
"flow_version": flow_version_of(project),
|
||
},
|
||
)
|
||
|
||
|
||
async def log_v2_event(
|
||
db: AsyncSession,
|
||
*,
|
||
project: ModuleGenerationProject,
|
||
event_type: str,
|
||
step: ModuleGenerationStep | None = None,
|
||
message: str | None = None,
|
||
detail: dict[str, Any] | None = None,
|
||
) -> None:
|
||
_ = db
|
||
event_detail = {
|
||
"flow_version": ModuleGenerationFlowVersionEnum.V2.value,
|
||
"step_index": step.step_index if step else None,
|
||
"step_code": step.step_code if step else None,
|
||
"step_version": step.version if step else None,
|
||
**(detail or {}),
|
||
}
|
||
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,
|
||
event_status=LogEventStatusEnum.SUCCESS.value,
|
||
source=LogSourceEnum.SERVICE.value,
|
||
message=message,
|
||
detail=event_detail,
|
||
)
|
||
log_operation_event(
|
||
domain="module_generation",
|
||
module=project.module,
|
||
event_type=event_type,
|
||
event_status=LogEventStatusEnum.SUCCESS.value,
|
||
source=LogSourceEnum.SERVICE.value,
|
||
user_id=project.user_id,
|
||
project_id=project.id,
|
||
step_id=step.id if step else None,
|
||
message=message,
|
||
detail=event_detail,
|
||
)
|
||
|
||
|
||
async def get_v2_project_for_user(
|
||
db: AsyncSession,
|
||
*,
|
||
config: ModuleGenerationV2Config,
|
||
project_id: str,
|
||
current_user: User | SimpleNamespace,
|
||
for_update: bool = False,
|
||
) -> ModuleGenerationProject:
|
||
stmt = select(ModuleGenerationProject).where(
|
||
ModuleGenerationProject.id == project_id,
|
||
ModuleGenerationProject.module == config.module,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
)
|
||
if not bool(getattr(current_user, "is_admin", False)):
|
||
stmt = stmt.where(ModuleGenerationProject.user_id == str(current_user.id))
|
||
if for_update:
|
||
stmt = stmt.with_for_update()
|
||
result = await execute_with_lock_timeout(db, stmt.limit(1))
|
||
else:
|
||
result = await db.execute(stmt.limit(1))
|
||
project = result.scalar_one_or_none()
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail=config.project_not_found_message)
|
||
assert_v2_project(project)
|
||
return project
|
||
|
||
|
||
def _validate_video_config(
|
||
engine: Any,
|
||
video_config: dict[str, Any],
|
||
*,
|
||
has_reference_image: bool,
|
||
) -> dict[str, Any]:
|
||
duration = int(video_config.get("duration") or 0)
|
||
aspect_ratio = str(video_config.get("aspect_ratio") or "").strip()
|
||
resolution = str(video_config.get("resolution") or "").strip()
|
||
ratios = [str(item) for item in parse_json_list(engine.supported_ratios, [])]
|
||
resolutions = [str(item) for item in parse_json_list(engine.supported_resolutions, [])]
|
||
durations: list[int] = []
|
||
for item in parse_json_list(engine.supported_durations, []):
|
||
try:
|
||
durations.append(int(item))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if ratios and aspect_ratio not in ratios:
|
||
raise HTTPException(status_code=400, detail="当前视频引擎不支持所选比例")
|
||
if resolutions and resolution not in resolutions:
|
||
raise HTTPException(status_code=400, detail="当前视频引擎不支持所选分辨率")
|
||
if durations and duration not in durations:
|
||
raise HTTPException(status_code=400, detail="当前视频引擎不支持所选时长")
|
||
if duration <= 0 or (int(engine.max_duration or 0) > 0 and duration > int(engine.max_duration)):
|
||
raise HTTPException(status_code=400, detail="视频时长不合法或超过当前引擎上限")
|
||
if has_reference_image:
|
||
if not bool(getattr(engine, "supports_universal_reference", False)):
|
||
raise HTTPException(status_code=400, detail="当前视频引擎不支持参考图片生成")
|
||
if int(getattr(engine, "max_image_count", 0) or 0) < 1:
|
||
raise HTTPException(status_code=400, detail="当前视频引擎允许的参考图片数量为 0")
|
||
return {
|
||
"engine_id": engine.id,
|
||
"duration": duration,
|
||
"aspect_ratio": aspect_ratio,
|
||
"resolution": resolution,
|
||
"engine_snapshot": build_video_snapshot(engine, aspect_ratio, resolution, duration),
|
||
}
|
||
|
||
|
||
async def _load_video_config(
|
||
db: AsyncSession,
|
||
video_config: Any,
|
||
*,
|
||
has_reference_image: bool,
|
||
) -> dict[str, Any]:
|
||
raw = video_config.model_dump() if hasattr(video_config, "model_dump") else dict(video_config or {})
|
||
engine = await get_video_engine(db, str(raw.get("engine_id") or "") or None)
|
||
return _validate_video_config(engine, raw, has_reference_image=has_reference_image)
|
||
|
||
|
||
async def _find_idempotent_project(
|
||
db: AsyncSession,
|
||
*,
|
||
user_id: str,
|
||
config: ModuleGenerationV2Config,
|
||
idempotency_key: str | None,
|
||
) -> ModuleGenerationProject | None:
|
||
if not idempotency_key:
|
||
return None
|
||
result = await db.execute(
|
||
select(ModuleGenerationProject)
|
||
.where(
|
||
ModuleGenerationProject.user_id == user_id,
|
||
ModuleGenerationProject.module == config.module,
|
||
ModuleGenerationProject.idempotency_key == idempotency_key,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
)
|
||
.order_by(ModuleGenerationProject.created_at.desc())
|
||
.limit(1)
|
||
)
|
||
project = result.scalar_one_or_none()
|
||
if project:
|
||
assert_v2_project(project)
|
||
return project
|
||
|
||
|
||
async def _create_material_and_prompt_steps(
|
||
db: AsyncSession,
|
||
*,
|
||
project: ModuleGenerationProject,
|
||
config: ModuleGenerationV2Config,
|
||
material_payload: dict[str, Any],
|
||
video_config: dict[str, Any],
|
||
target_platform: str,
|
||
context: dict[str, Any] | None = None,
|
||
) -> tuple[ModuleGenerationStep, ModuleGenerationStep]:
|
||
material_step = await create_module_step(
|
||
db,
|
||
project=project,
|
||
step_code=MATERIAL_INPUT,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
status=ModuleStepStatusEnum.COMPLETED.value,
|
||
input_data={**material_payload, "source_context": context or {}},
|
||
output_data={
|
||
"accepted": True,
|
||
"message": "V2 素材和项目信息已保存",
|
||
"next_step_code": VIDEO_PROMPT_OPTIMIZE,
|
||
},
|
||
)
|
||
prompt_step = await create_module_step(
|
||
db,
|
||
project=project,
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=material_step.id,
|
||
source_step_id=material_step.id,
|
||
input_data={
|
||
"source_step_id": material_step.id,
|
||
"video_config": video_config,
|
||
"target_platform": target_platform,
|
||
"trigger": "project_create",
|
||
},
|
||
)
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = VIDEO_PROMPT_OPTIMIZE
|
||
project.error_message = None
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=prompt_step,
|
||
event_type=ModuleEventTypeEnum.V2_VIDEO_PROMPT_AUTO_CREATED.value,
|
||
message="创建 V2 项目后自动创建视频提词步骤",
|
||
detail={"material_step_id": material_step.id},
|
||
)
|
||
return material_step, prompt_step
|
||
|
||
|
||
|
||
|
||
def build_v2_video_prompt_billing_context(
|
||
*,
|
||
user_id: str,
|
||
project_id: str,
|
||
step_id: str,
|
||
step_version: int,
|
||
module: str,
|
||
display_name: str,
|
||
) -> LlmBillingContext:
|
||
return LlmBillingContext(
|
||
user_id=str(user_id),
|
||
owner_type=CreditRecordOwnerType.MODULE_GENERATION_STEP.value,
|
||
owner_id=str(step_id),
|
||
attempt_no=int(step_version or 1),
|
||
charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value,
|
||
billing_scene=(
|
||
CreditRecordBillingScene.HOT_OPENING_VIDEO_PROMPT_OPTIMIZE.value
|
||
if module == "hot_opening_replicate"
|
||
else CreditRecordBillingScene.SHOT_VIDEO_PROMPT_OPTIMIZE.value
|
||
),
|
||
source_module=str(module),
|
||
source_project_id=str(project_id),
|
||
source_step_id=str(step_id),
|
||
source_step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
related_id=str(step_id),
|
||
hold_config_key=LlmBillingConfigKey.HOLD_MODULE_VIDEO_PROMPT.value,
|
||
description_prefix=f"{display_name}视频提词优化",
|
||
trace_id=f"module-v2-video-prompt:{step_id}",
|
||
)
|
||
|
||
|
||
async def create_hot_opening_project_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
req: HotOpeningTaskCreateV2,
|
||
) -> ModuleProjectCreateV2Result:
|
||
existing = await _find_idempotent_project(
|
||
db,
|
||
user_id=current_user.id,
|
||
config=HOT_OPENING_V2,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
if existing:
|
||
step = await get_current_step_by_code(
|
||
db,
|
||
project_id=existing.id,
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
config=HOT_OPENING_V2.flow_config,
|
||
)
|
||
if not step:
|
||
raise HTTPException(status_code=409, detail="幂等项目缺少当前视频提词步骤")
|
||
return ModuleProjectCreateV2Result(existing, step, False)
|
||
|
||
video_config = await _load_video_config(
|
||
db, req.video_config, has_reference_image=bool(req.material_image_url)
|
||
)
|
||
project = ModuleGenerationProject(
|
||
id=generate_id(),
|
||
user_id=current_user.id,
|
||
module=HOT_OPENING_V2.module,
|
||
flow_version=ModuleGenerationFlowVersionEnum.V2.value,
|
||
title=req.target_project_name or req.source_project_name or "爆款开头复刻",
|
||
status=ModuleProjectStatusEnum.PROCESSING.value,
|
||
current_step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
db.add(project)
|
||
await db.flush()
|
||
material_payload = {
|
||
"material_video_url": req.material_video_url,
|
||
"material_video_resource_id": req.material_video_resource_id,
|
||
"material_video_duration_seconds": req.material_video_duration_seconds,
|
||
"material_image_url": req.material_image_url,
|
||
"material_image_resource_id": req.material_image_resource_id,
|
||
"source_project_name": req.source_project_name or "参考素材",
|
||
"target_project_name": req.target_project_name or project.title,
|
||
"core_content_point": req.core_content_point or req.project_description or "按参考素材生成视频",
|
||
"project_description": req.project_description,
|
||
"material_video_locked": True,
|
||
}
|
||
_, prompt_step = await _create_material_and_prompt_steps(
|
||
db,
|
||
project=project,
|
||
config=HOT_OPENING_V2,
|
||
material_payload=material_payload,
|
||
video_config=video_config,
|
||
target_platform=req.target_platform or "抖音",
|
||
)
|
||
await start_hold(
|
||
db,
|
||
build_v2_video_prompt_billing_context(
|
||
user_id=str(current_user.id),
|
||
project_id=str(project.id),
|
||
step_id=str(prompt_step.id),
|
||
step_version=int(prompt_step.version or 1),
|
||
module=HOT_OPENING_V2.module,
|
||
display_name=HOT_OPENING_V2.display_name,
|
||
),
|
||
)
|
||
await bind_upload_resources(
|
||
db,
|
||
user_id=current_user.id,
|
||
module=UploadResourceModuleEnum.HOT_OPENING_REPLICATE.value,
|
||
source_model=UploadResourceSourceModelEnum.MODULE_GENERATION_PROJECT.value,
|
||
source_id=project.id,
|
||
resource_ids=[req.material_video_resource_id, req.material_image_resource_id],
|
||
urls=[req.material_video_url, req.material_image_url],
|
||
allow_common_migrate=True,
|
||
)
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
event_type=ModuleEventTypeEnum.V2_PROJECT_CREATED.value,
|
||
message="创建爆款开头复刻 V2 三步骤项目",
|
||
detail={"engine_id": video_config["engine_id"]},
|
||
)
|
||
return ModuleProjectCreateV2Result(project, prompt_step, True)
|
||
|
||
|
||
async def create_shot_replicate_project_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
current_user: User,
|
||
segment: ShotReplicateSegment,
|
||
req: ShotReplicateProjectCreateV2,
|
||
) -> ModuleProjectCreateV2Result:
|
||
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:
|
||
result = await db.execute(
|
||
select(ModuleGenerationProject)
|
||
.where(
|
||
ModuleGenerationProject.id == segment.module_project_id,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
)
|
||
.limit(1)
|
||
)
|
||
existing = result.scalar_one_or_none()
|
||
if existing:
|
||
if flow_version_of(existing) != ModuleGenerationFlowVersionEnum.V2.value:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"message": "该片段已关联历史 V1 复刻项目",
|
||
"existing_project_id": existing.id,
|
||
"existing_flow_version": flow_version_of(existing),
|
||
},
|
||
)
|
||
step = await get_current_step_by_code(
|
||
db,
|
||
project_id=existing.id,
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
config=SHOT_REPLICATE_V2.flow_config,
|
||
)
|
||
if not step:
|
||
raise HTTPException(status_code=409, detail="关联 V2 项目缺少当前视频提词步骤")
|
||
return ModuleProjectCreateV2Result(existing, step, False)
|
||
|
||
existing = await _find_idempotent_project(
|
||
db,
|
||
user_id=current_user.id,
|
||
config=SHOT_REPLICATE_V2,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
if existing:
|
||
assert_v2_project(existing)
|
||
material = await _current_material_step(
|
||
db, project=existing, config=SHOT_REPLICATE_V2, for_update=False
|
||
)
|
||
existing_payload = step_payload(
|
||
material.input_json, schema_version=SHOT_REPLICATE_V2.io_schema_version
|
||
)
|
||
if str(existing_payload.get("source_segment_id") or "") != str(segment.id):
|
||
raise HTTPException(status_code=409, detail="该幂等键已用于其他拆镜片段")
|
||
segment.module_project_id = existing.id
|
||
segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
|
||
step = await get_current_step_by_code(
|
||
db,
|
||
project_id=existing.id,
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
config=SHOT_REPLICATE_V2.flow_config,
|
||
)
|
||
if not step:
|
||
raise HTTPException(status_code=409, detail="幂等项目缺少当前视频提词步骤")
|
||
return ModuleProjectCreateV2Result(existing, step, False)
|
||
|
||
video_config = await _load_video_config(
|
||
db, req.video_config, has_reference_image=bool(req.material_image_url)
|
||
)
|
||
project = ModuleGenerationProject(
|
||
id=generate_id(),
|
||
user_id=current_user.id,
|
||
module=SHOT_REPLICATE_V2.module,
|
||
flow_version=ModuleGenerationFlowVersionEnum.V2.value,
|
||
title=req.target_project_name or "拆镜复刻",
|
||
status=ModuleProjectStatusEnum.PROCESSING.value,
|
||
current_step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
db.add(project)
|
||
await db.flush()
|
||
material_payload = {
|
||
"material_video_url": segment.segment_video_url,
|
||
"material_video_locked": True,
|
||
"material_image_url": req.material_image_url,
|
||
"material_image_resource_id": req.material_image_resource_id,
|
||
"source_project_name": segment.segment_category or segment.original_video_category or "拆镜片段",
|
||
"target_project_name": req.target_project_name or project.title,
|
||
"core_content_point": req.core_content_point or req.project_description or segment.segment_content or "按拆镜片段生成视频",
|
||
"project_description": req.project_description,
|
||
"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,
|
||
},
|
||
}
|
||
_, prompt_step = await _create_material_and_prompt_steps(
|
||
db,
|
||
project=project,
|
||
config=SHOT_REPLICATE_V2,
|
||
material_payload=material_payload,
|
||
video_config=video_config,
|
||
target_platform=req.target_platform or "抖音",
|
||
context=context,
|
||
)
|
||
await bind_upload_resources(
|
||
db,
|
||
user_id=current_user.id,
|
||
module=UploadResourceModuleEnum.SHOT_REPLICATE.value,
|
||
source_model=UploadResourceSourceModelEnum.MODULE_GENERATION_PROJECT.value,
|
||
source_id=project.id,
|
||
resource_ids=[req.material_image_resource_id],
|
||
urls=[req.material_image_url],
|
||
allow_common_migrate=True,
|
||
)
|
||
await start_hold(
|
||
db,
|
||
build_v2_video_prompt_billing_context(
|
||
user_id=str(current_user.id),
|
||
project_id=str(project.id),
|
||
step_id=str(prompt_step.id),
|
||
step_version=int(prompt_step.version or 1),
|
||
module=SHOT_REPLICATE_V2.module,
|
||
display_name=SHOT_REPLICATE_V2.display_name,
|
||
),
|
||
)
|
||
segment.module_project_id = project.id
|
||
segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
event_type=ModuleEventTypeEnum.V2_PROJECT_CREATED.value,
|
||
message="从拆镜片段创建 V2 三步骤复刻项目",
|
||
detail={"segment_id": segment.id, "engine_id": video_config["engine_id"]},
|
||
)
|
||
return ModuleProjectCreateV2Result(project, prompt_step, True)
|
||
|
||
|
||
async def _current_material_step(
|
||
db: AsyncSession,
|
||
*,
|
||
project: ModuleGenerationProject,
|
||
config: ModuleGenerationV2Config,
|
||
for_update: bool = False,
|
||
) -> ModuleGenerationStep:
|
||
stmt = select(ModuleGenerationStep).where(
|
||
ModuleGenerationStep.project_id == project.id,
|
||
ModuleGenerationStep.module == config.module,
|
||
ModuleGenerationStep.step_code == MATERIAL_INPUT,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
if for_update:
|
||
stmt = stmt.with_for_update()
|
||
result = await execute_with_lock_timeout(db, stmt.limit(1))
|
||
else:
|
||
result = await db.execute(stmt.limit(1))
|
||
step = result.scalar_one_or_none()
|
||
if not step:
|
||
raise HTTPException(status_code=409, detail="V2 项目缺少素材步骤")
|
||
return step
|
||
|
||
|
||
async def rebuild_video_prompt_step_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
config: ModuleGenerationV2Config,
|
||
current_user: User,
|
||
project_id: str,
|
||
source_prompt_step_id: str,
|
||
video_config: ModuleGenerationVideoConfigV2,
|
||
trigger: str = "manual_retry",
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
|
||
project = await get_v2_project_for_user(
|
||
db,
|
||
config=config,
|
||
project_id=project_id,
|
||
current_user=current_user,
|
||
for_update=True,
|
||
)
|
||
await assert_project_has_no_active_chat_tasks(
|
||
db,
|
||
project=project,
|
||
config=config.flow_config,
|
||
detail_message=f"当前{config.display_name}项目仍有视频生成任务处理中，暂不能重新生成视频提词",
|
||
)
|
||
material_step = await _current_material_step(db, project=project, config=config, for_update=True)
|
||
current_prompt = await get_current_step_by_code(
|
||
db,
|
||
project_id=project.id,
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
config=config.flow_config,
|
||
)
|
||
if not current_prompt:
|
||
raise HTTPException(status_code=404, detail="当前视频提词步骤不存在")
|
||
if str(current_prompt.id) != str(source_prompt_step_id):
|
||
raise HTTPException(status_code=409, detail="路径中的步骤不是当前有效视频提词步骤")
|
||
if current_prompt.status == ModuleStepStatusEnum.PROCESSING.value:
|
||
raise HTTPException(status_code=409, detail="视频提词正在生成，请等待完成或失败后再重试")
|
||
|
||
prompt_input = step_payload(current_prompt.input_json, schema_version=config.io_schema_version)
|
||
old_video_config = dict(prompt_input.get("video_config") or {})
|
||
material_input = step_payload(material_step.input_json, schema_version=config.io_schema_version)
|
||
validated_video_config = await _load_video_config(
|
||
db,
|
||
video_config,
|
||
has_reference_image=bool(material_input.get("material_image_url")),
|
||
)
|
||
target_platform = str(prompt_input.get("target_platform") or "抖音")
|
||
|
||
await soft_delete_steps_from_index(
|
||
db,
|
||
project=project,
|
||
start_index=2,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
)
|
||
step = await create_module_step(
|
||
db,
|
||
project=project,
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=material_step.id,
|
||
source_step_id=material_step.id,
|
||
input_data={
|
||
"source_step_id": material_step.id,
|
||
"video_config": validated_video_config,
|
||
"target_platform": target_platform,
|
||
"trigger": trigger,
|
||
},
|
||
)
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = VIDEO_PROMPT_OPTIMIZE
|
||
project.final_video_url = None
|
||
project.final_video_cover_url = None
|
||
project.completed_at = None
|
||
project.error_message = None
|
||
await start_hold(
|
||
db,
|
||
build_v2_video_prompt_billing_context(
|
||
user_id=str(project.user_id),
|
||
project_id=str(project.id),
|
||
step_id=str(step.id),
|
||
step_version=int(step.version or 1),
|
||
module=config.module,
|
||
display_name=config.display_name,
|
||
),
|
||
)
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.V2_VIDEO_PROMPT_REGENERATED.value,
|
||
message="重复生成 V2 视频提词",
|
||
detail={
|
||
"trigger": trigger,
|
||
"material_step_id": material_step.id,
|
||
"source_prompt_step_id": current_prompt.id,
|
||
"previous_video_config": old_video_config,
|
||
"video_config": validated_video_config,
|
||
},
|
||
)
|
||
return project, step
|
||
|
||
|
||
async def mark_video_prompt_dispatch_failed_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
config: ModuleGenerationV2Config,
|
||
project_id: str,
|
||
step_id: str,
|
||
error_message: str,
|
||
) -> None:
|
||
result = await execute_with_lock_timeout(
|
||
db,
|
||
select(ModuleGenerationProject, ModuleGenerationStep)
|
||
.join(ModuleGenerationStep, ModuleGenerationStep.project_id == ModuleGenerationProject.id)
|
||
.where(
|
||
ModuleGenerationProject.id == project_id,
|
||
ModuleGenerationProject.module == config.module,
|
||
ModuleGenerationProject.flow_version == ModuleGenerationFlowVersionEnum.V2.value,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
ModuleGenerationStep.id == step_id,
|
||
ModuleGenerationStep.step_code == VIDEO_PROMPT_OPTIMIZE,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1),
|
||
)
|
||
row = result.first()
|
||
if not row:
|
||
await db.rollback()
|
||
return
|
||
project, step = row
|
||
if step.status != ModuleStepStatusEnum.PROCESSING.value:
|
||
await db.rollback()
|
||
return
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = error_message
|
||
step.completed_at = utc_now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.current_step_code = VIDEO_PROMPT_OPTIMIZE
|
||
project.error_message = error_message
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.V2_VIDEO_PROMPT_DISPATCH_FAILED.value,
|
||
message=error_message,
|
||
detail={"dispatch_compensated": True},
|
||
)
|
||
await release_on_failure(
|
||
db,
|
||
build_v2_video_prompt_billing_context(
|
||
user_id=str(project.user_id),
|
||
project_id=str(project.id),
|
||
step_id=str(step.id),
|
||
step_version=int(step.version or 1),
|
||
module=config.module,
|
||
display_name=config.display_name,
|
||
),
|
||
error=error_message,
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def run_video_prompt_optimize_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
step_id: str,
|
||
execution_guard: Callable[[], Awaitable[None]] | None = None,
|
||
) -> ModuleGenerationStep | None:
|
||
try:
|
||
meta_result = await execute_with_lock_timeout(
|
||
db,
|
||
select(ModuleGenerationProject, ModuleGenerationStep)
|
||
.join(ModuleGenerationStep, ModuleGenerationStep.project_id == ModuleGenerationProject.id)
|
||
.where(
|
||
ModuleGenerationProject.id == project_id,
|
||
ModuleGenerationProject.flow_version == ModuleGenerationFlowVersionEnum.V2.value,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
ModuleGenerationStep.id == step_id,
|
||
ModuleGenerationStep.project_id == project_id,
|
||
ModuleGenerationStep.step_code == VIDEO_PROMPT_OPTIMIZE,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1),
|
||
)
|
||
row = meta_result.first()
|
||
if not row:
|
||
await db.rollback()
|
||
return None
|
||
project, step = row
|
||
config = CONFIG_BY_MODULE.get(project.module)
|
||
if not config or step.status != ModuleStepStatusEnum.PROCESSING.value:
|
||
await db.rollback()
|
||
return step
|
||
material_step = await _current_material_step(db, project=project, config=config, for_update=False)
|
||
material_payload = step_payload(material_step.input_json, schema_version=config.io_schema_version)
|
||
prompt_input = step_payload(step.input_json, schema_version=config.io_schema_version)
|
||
video_config = dict(prompt_input.get("video_config") or {})
|
||
if not all(video_config.get(key) for key in ("engine_id", "duration", "aspect_ratio", "resolution")):
|
||
raise RuntimeError("V2 视频提词步骤缺少完整视频参数")
|
||
expected_version = int(step.version)
|
||
expected_input = json.loads(json.dumps(step.input_json, ensure_ascii=False, default=str))
|
||
project_snapshot = {
|
||
"module": project.module,
|
||
"project_id": project.id,
|
||
"step_id": step.id,
|
||
"user_id": project.user_id,
|
||
"source_project_name": material_payload.get("source_project_name") or "参考素材",
|
||
"target_project_name": material_payload.get("target_project_name") or project.title or config.display_name,
|
||
"core_content_point": material_payload.get("core_content_point") or material_payload.get("project_description") or "按参考素材生成视频",
|
||
"material_video_url": material_payload.get("material_video_url"),
|
||
"material_image_url": material_payload.get("material_image_url"),
|
||
"target_platform": prompt_input.get("target_platform") or "抖音",
|
||
"video_config": video_config,
|
||
}
|
||
if not project_snapshot["material_video_url"]:
|
||
raise RuntimeError("V2 素材步骤缺少参考视频")
|
||
schema_config_snapshot = await get_runtime_schema_snapshot(db)
|
||
llm_billing_context = LlmBillingContext(
|
||
user_id=project_snapshot["user_id"],
|
||
owner_type=CreditRecordOwnerType.MODULE_GENERATION_STEP.value,
|
||
owner_id=project_snapshot["step_id"],
|
||
attempt_no=expected_version,
|
||
charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value,
|
||
billing_scene=(
|
||
CreditRecordBillingScene.HOT_OPENING_VIDEO_PROMPT_OPTIMIZE.value
|
||
if project_snapshot["module"] == "hot_opening_replicate"
|
||
else CreditRecordBillingScene.SHOT_VIDEO_PROMPT_OPTIMIZE.value
|
||
),
|
||
source_module=project_snapshot["module"],
|
||
source_project_id=project_snapshot["project_id"],
|
||
source_step_id=project_snapshot["step_id"],
|
||
source_step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
related_id=project_snapshot["step_id"],
|
||
hold_config_key=LlmBillingConfigKey.HOLD_MODULE_VIDEO_PROMPT.value,
|
||
description_prefix=f"{config.display_name}视频提词优化",
|
||
trace_id=f"module-v2-video-prompt:{project_snapshot['step_id']}",
|
||
)
|
||
hold_validation = await ensure_hold_exists(db, llm_billing_context)
|
||
if not hold_validation.can_execute:
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = f"LLM账务状态异常（{hold_validation.state.value}），已终止任务"
|
||
step.completed_at = utc_now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = step.error_message
|
||
await log_v2_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_FAILED.value, message=step.error_message)
|
||
await db.commit()
|
||
return step
|
||
await db.commit()
|
||
|
||
provider_succeeded = False
|
||
usage: dict[str, Any] = {}
|
||
log_provider_start(llm_billing_context, detail={"prompt_type": "video", "flow_version": "v2"})
|
||
prompt_schema, final_prompt, usage = await optimize_hot_opening_video_prompt(
|
||
db,
|
||
user_id=project_snapshot["user_id"],
|
||
source_project_name=project_snapshot["source_project_name"],
|
||
target_project_name=project_snapshot["target_project_name"],
|
||
core_content_point=project_snapshot["core_content_point"],
|
||
material_video_url=project_snapshot["material_video_url"],
|
||
generated_image_url=project_snapshot["material_image_url"],
|
||
video_config=project_snapshot["video_config"],
|
||
target_platform=project_snapshot["target_platform"],
|
||
schema_config_snapshot=schema_config_snapshot,
|
||
module=project_snapshot["module"],
|
||
project_id=project_snapshot["project_id"],
|
||
step_id=project_snapshot["step_id"],
|
||
)
|
||
provider_succeeded = True
|
||
log_provider_success(llm_billing_context, usage=usage)
|
||
|
||
if execution_guard is not None:
|
||
await execution_guard()
|
||
locked = await execute_with_lock_timeout(
|
||
db,
|
||
select(ModuleGenerationProject, ModuleGenerationStep)
|
||
.join(ModuleGenerationStep, ModuleGenerationStep.project_id == ModuleGenerationProject.id)
|
||
.where(
|
||
ModuleGenerationProject.id == project_snapshot["project_id"],
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
ModuleGenerationProject.flow_version == ModuleGenerationFlowVersionEnum.V2.value,
|
||
ModuleGenerationStep.id == project_snapshot["step_id"],
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1),
|
||
)
|
||
row = locked.first()
|
||
if not row:
|
||
await db.rollback()
|
||
# Provider 已成功，即使业务对象被异常移除，也必须按真实 usage 完成幂等结算。
|
||
await settle_success(
|
||
db,
|
||
llm_billing_context,
|
||
usage=usage,
|
||
description=f"{config.display_name}-视频提词优化（业务对象失效结算）",
|
||
)
|
||
await db.commit()
|
||
return None
|
||
project, step = row
|
||
if int(step.version) != expected_version or step.input_json != expected_input or step.status != ModuleStepStatusEnum.PROCESSING.value:
|
||
await settle_success(
|
||
db,
|
||
llm_billing_context,
|
||
usage=usage,
|
||
description=f"{config.display_name}-视频提词优化（失效结果结算）",
|
||
)
|
||
await db.commit()
|
||
log_module_event_file(
|
||
module=project_snapshot["module"],
|
||
event_type=ModuleEventTypeEnum.STALE_STEP_RESULT_DISCARDED.value,
|
||
project_id=project_snapshot["project_id"],
|
||
step_id=project_snapshot["step_id"],
|
||
user_id=project_snapshot["user_id"],
|
||
message="V2 旧视频提词结果已丢弃",
|
||
detail={"expected_version": expected_version},
|
||
)
|
||
return None
|
||
billing = await settle_success(
|
||
db,
|
||
llm_billing_context,
|
||
usage=usage,
|
||
description=f"{config.display_name}-视频提词优化",
|
||
)
|
||
output_payload = {
|
||
"prompt_schema": prompt_schema,
|
||
"final_prompt": final_prompt,
|
||
"params_used_for_prompt": project_snapshot["video_config"],
|
||
"target_platform": project_snapshot["target_platform"],
|
||
"schema_config_snapshot": schema_config_snapshot,
|
||
"schema_config_source": schema_config_snapshot.get("source"),
|
||
"schema_config_version": schema_config_snapshot.get("version"),
|
||
}
|
||
force_set_json(
|
||
step,
|
||
"output_json",
|
||
build_step_output(
|
||
step_code=VIDEO_PROMPT_OPTIMIZE,
|
||
status=ModuleStepStatusEnum.COMPLETED.value,
|
||
payload=output_payload,
|
||
usage={
|
||
**dict(usage or {}),
|
||
"text_credits_cost": billing.get_amount(CreditRecordChargeKind.TEXT_PROMPT.value),
|
||
},
|
||
schema_version=config.io_schema_version,
|
||
),
|
||
)
|
||
step.status = ModuleStepStatusEnum.COMPLETED.value
|
||
step.error_message = None
|
||
step.completed_at = utc_now()
|
||
project.status = ModuleProjectStatusEnum.WAITING_USER.value
|
||
project.current_step_code = VIDEO_PROMPT_OPTIMIZE
|
||
project.error_message = None
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.VIDEO_PROMPT_SUCCESS.value,
|
||
message="V2 视频提词生成成功",
|
||
detail={"engine_id": project_snapshot["video_config"].get("engine_id")},
|
||
)
|
||
await db.commit()
|
||
return step
|
||
except DatabaseRowLockBusy:
|
||
await db.rollback()
|
||
if locals().get("provider_succeeded", False):
|
||
await settle_success(
|
||
db,
|
||
llm_billing_context,
|
||
usage=locals().get("usage") or {},
|
||
description=f"{config.display_name}-视频提词优化（行锁失败结算）",
|
||
)
|
||
await db.commit()
|
||
return None
|
||
raise
|
||
except Exception as exc:
|
||
await db.rollback()
|
||
if "llm_billing_context" in locals() and not locals().get("provider_succeeded", False):
|
||
log_provider_failure(llm_billing_context, error=str(exc))
|
||
try:
|
||
if "llm_billing_context" in locals():
|
||
if locals().get("provider_succeeded", False):
|
||
await settle_success(
|
||
db,
|
||
llm_billing_context,
|
||
usage=locals().get("usage") or {},
|
||
description=f"{config.display_name}-视频提词优化（本地失败结算）",
|
||
)
|
||
else:
|
||
await release_on_failure(db, llm_billing_context, error=str(exc))
|
||
if execution_guard is not None:
|
||
await execution_guard()
|
||
result = await execute_with_lock_timeout(
|
||
db,
|
||
select(ModuleGenerationProject, ModuleGenerationStep)
|
||
.join(ModuleGenerationStep, ModuleGenerationStep.project_id == ModuleGenerationProject.id)
|
||
.where(
|
||
ModuleGenerationProject.id == project_id,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
ModuleGenerationStep.id == step_id,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1),
|
||
)
|
||
row = result.first()
|
||
if row:
|
||
project, step = row
|
||
if step.status == ModuleStepStatusEnum.PROCESSING.value:
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = str(exc) if str(exc) else type(exc).__name__
|
||
step.completed_at = utc_now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = str(exc) if str(exc) else type(exc).__name__
|
||
# provider 已成功时 settle_success 已在当前事务写入 RELEASE/CHARGE；
|
||
# 即使业务步骤已不存在或已不是 processing，也必须提交账务结算。
|
||
await db.commit()
|
||
log_module_error(
|
||
module=row[0].module if row else "module_generation_v2",
|
||
event_type=ModuleEventTypeEnum.VIDEO_PROMPT_FAILED.value,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="V2 视频提词生成失败",
|
||
exc=exc,
|
||
)
|
||
except Exception as mark_exc:
|
||
await db.rollback()
|
||
log_module_error(
|
||
module="module_generation_v2",
|
||
event_type=ModuleEventTypeEnum.VIDEO_PROMPT_FAILED.value,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
message="V2 视频提词失败状态落库失败",
|
||
detail={"origin_error": str(exc), "provider_succeeded": locals().get("provider_succeeded", False)},
|
||
exc=mark_exc,
|
||
)
|
||
if locals().get("provider_succeeded", False):
|
||
raise
|
||
return None
|
||
|
||
|
||
async def update_video_prompt_schema_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
config: ModuleGenerationV2Config,
|
||
current_user: User,
|
||
project_id: str,
|
||
step_id: str,
|
||
req: ModuleVideoPromptSchemaUpdateV2,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
|
||
# 先做版本校验；公共更新服务负责短事务锁、白名单 patch 和下游软删。
|
||
await get_v2_project_for_user(
|
||
db,
|
||
config=config,
|
||
project_id=project_id,
|
||
current_user=current_user,
|
||
for_update=False,
|
||
)
|
||
return await update_module_video_prompt_schema(
|
||
db,
|
||
current_user=current_user,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
req=req,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
patch_video_prompt_schema_from_client=patch_video_prompt_schema_from_client,
|
||
build_final_video_prompt=build_final_video_prompt,
|
||
)
|
||
|
||
|
||
async def generate_video_from_prompt_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
config: ModuleGenerationV2Config,
|
||
current_user: User,
|
||
project_id: str,
|
||
prompt_step_id: str,
|
||
) -> tuple[ModuleGenerationProject, ModuleGenerationStep, ChatGenerationTask]:
|
||
project = await get_v2_project_for_user(
|
||
db,
|
||
config=config,
|
||
project_id=project_id,
|
||
current_user=current_user,
|
||
for_update=True,
|
||
)
|
||
await assert_project_has_no_active_chat_tasks(
|
||
db,
|
||
project=project,
|
||
config=config.flow_config,
|
||
detail_message=f"当前{config.display_name}项目已有视频生成任务处理中，请等待完成后再操作",
|
||
)
|
||
prompt_result = await execute_with_lock_timeout(
|
||
db,
|
||
select(ModuleGenerationStep)
|
||
.where(
|
||
ModuleGenerationStep.id == prompt_step_id,
|
||
ModuleGenerationStep.project_id == project.id,
|
||
ModuleGenerationStep.module == config.module,
|
||
ModuleGenerationStep.step_code == VIDEO_PROMPT_OPTIMIZE,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1),
|
||
)
|
||
prompt_step = prompt_result.scalar_one_or_none()
|
||
if not prompt_step:
|
||
raise HTTPException(status_code=404, detail="当前视频提词步骤不存在")
|
||
if prompt_step.status != ModuleStepStatusEnum.COMPLETED.value:
|
||
raise HTTPException(status_code=400, detail="视频提词未完成，不能生成视频")
|
||
material_step = await _current_material_step(db, project=project, config=config, for_update=False)
|
||
material_payload = step_payload(material_step.input_json, schema_version=config.io_schema_version)
|
||
prompt_input = step_payload(prompt_step.input_json, schema_version=config.io_schema_version)
|
||
prompt_output = step_payload(prompt_step.output_json, schema_version=config.io_schema_version)
|
||
prompt_schema = prompt_output.get("prompt_schema")
|
||
final_prompt = str(prompt_output.get("final_prompt") or "").strip()
|
||
video_config = dict(prompt_input.get("video_config") or {})
|
||
if not isinstance(prompt_schema, dict) or not prompt_schema:
|
||
raise HTTPException(status_code=400, detail="视频提词缺少 prompt_schema")
|
||
if not final_prompt:
|
||
final_prompt = build_final_video_prompt(prompt_schema)
|
||
if not all(video_config.get(key) for key in ("engine_id", "duration", "aspect_ratio", "resolution")):
|
||
raise HTTPException(status_code=400, detail="视频提词缺少完整视频参数快照")
|
||
|
||
await soft_delete_steps_from_index(
|
||
db,
|
||
project=project,
|
||
start_index=3,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
)
|
||
references = build_v2_video_generation_references(material_payload)
|
||
assert_v2_video_generation_references(references)
|
||
engine = await get_video_engine(db, str(video_config["engine_id"]))
|
||
video_config = _validate_video_config(
|
||
engine,
|
||
video_config,
|
||
has_reference_image=bool(references),
|
||
)
|
||
generate_step = await create_module_step(
|
||
db,
|
||
project=project,
|
||
step_code=VIDEO_GENERATE,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
status=ModuleStepStatusEnum.PROCESSING.value,
|
||
parent_step_id=prompt_step.id,
|
||
source_step_id=prompt_step.id,
|
||
input_data={
|
||
"engine_id": video_config["engine_id"],
|
||
"params": {
|
||
"duration": int(video_config["duration"]),
|
||
"aspect_ratio": str(video_config["aspect_ratio"]),
|
||
"resolution": str(video_config["resolution"]),
|
||
},
|
||
"prompt_schema": prompt_schema,
|
||
"final_prompt": final_prompt,
|
||
"media_references": references,
|
||
"attachment_policy": "optional_image_only_no_video",
|
||
},
|
||
)
|
||
prompt_schema_text = json.dumps(prompt_schema, ensure_ascii=False)
|
||
task = await create_chat_generation_task_for_module(
|
||
db,
|
||
current_user=current_user,
|
||
generation_mode=config.generation_mode,
|
||
gen_type="video",
|
||
original_prompt=prompt_schema_text,
|
||
optimized_prompt=prompt_schema_text,
|
||
engine_id=str(video_config["engine_id"]),
|
||
media_references=references,
|
||
duration=int(video_config["duration"]),
|
||
aspect_ratio=str(video_config["aspect_ratio"]),
|
||
resolution=str(video_config["resolution"]),
|
||
billing_project_name=project.title or config.display_name,
|
||
billing_description_prefix=f"{config.display_name}视频生成-",
|
||
billing_source_module=project.module,
|
||
billing_source_project_id=project.id,
|
||
billing_source_step_id=generate_step.id,
|
||
billing_source_step_code=VIDEO_GENERATE,
|
||
)
|
||
generate_step.chat_task_id = task.id
|
||
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
||
project.current_step_code = VIDEO_GENERATE
|
||
project.error_message = None
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=generate_step,
|
||
event_type=ModuleEventTypeEnum.VIDEO_GENERATE_SUBMITTED.value,
|
||
message="V2 视频生成任务已创建",
|
||
detail={
|
||
"chat_task_id": task.id,
|
||
"engine_id": task.engine_id,
|
||
"material_image_count": len(references),
|
||
"material_video_count": 0,
|
||
},
|
||
)
|
||
return project, generate_step, task
|
||
|
||
|
||
async def handle_chat_generation_task_finished_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
task: ChatGenerationTask,
|
||
) -> bool:
|
||
meta = await db.execute(
|
||
select(ModuleGenerationStep.id, ModuleGenerationStep.project_id)
|
||
.where(
|
||
ModuleGenerationStep.chat_task_id == task.id,
|
||
ModuleGenerationStep.step_code == VIDEO_GENERATE,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.limit(1)
|
||
)
|
||
row = meta.first()
|
||
if not row:
|
||
return False
|
||
result = await execute_with_lock_timeout(
|
||
db,
|
||
select(ModuleGenerationProject, ModuleGenerationStep)
|
||
.join(ModuleGenerationStep, ModuleGenerationStep.project_id == ModuleGenerationProject.id)
|
||
.where(
|
||
ModuleGenerationProject.id == str(row.project_id),
|
||
ModuleGenerationProject.flow_version == ModuleGenerationFlowVersionEnum.V2.value,
|
||
ModuleGenerationProject.deleted_at.is_(None),
|
||
ModuleGenerationStep.id == str(row.id),
|
||
ModuleGenerationStep.chat_task_id == task.id,
|
||
ModuleGenerationStep.is_current.is_(True),
|
||
ModuleGenerationStep.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1),
|
||
)
|
||
locked = result.first()
|
||
if not locked:
|
||
return False
|
||
project, step = locked
|
||
config = CONFIG_BY_MODULE.get(project.module)
|
||
if not config:
|
||
return False
|
||
if task.status == ChatGenerationTaskStatus.COMPLETED.value:
|
||
if step.status == ModuleStepStatusEnum.COMPLETED.value and project.status == ModuleProjectStatusEnum.COMPLETED.value:
|
||
return True
|
||
force_set_json(
|
||
step,
|
||
"output_json",
|
||
build_step_output(
|
||
step_code=VIDEO_GENERATE,
|
||
status=ModuleStepStatusEnum.COMPLETED.value,
|
||
result={
|
||
"result_video_url": task.video_url,
|
||
"result_video_cover_url": task.video_cover_url,
|
||
"chat_task_id": task.id,
|
||
},
|
||
schema_version=config.io_schema_version,
|
||
),
|
||
)
|
||
step.status = ModuleStepStatusEnum.COMPLETED.value
|
||
step.error_message = None
|
||
step.completed_at = utc_now()
|
||
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 = VIDEO_GENERATE
|
||
project.completed_at = utc_now()
|
||
project.error_message = None
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.VIDEO_GENERATE_SUCCESS.value,
|
||
message="V2 视频生成完成",
|
||
detail={"chat_task_id": task.id},
|
||
)
|
||
elif task.status == ChatGenerationTaskStatus.FAILED.value:
|
||
step.status = ModuleStepStatusEnum.FAILED.value
|
||
step.error_message = task.error_message
|
||
step.completed_at = utc_now()
|
||
project.status = ModuleProjectStatusEnum.FAILED.value
|
||
project.error_message = task.error_message or "视频生成失败"
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
step=step,
|
||
event_type=ModuleEventTypeEnum.CHAT_TASK_FAILED.value,
|
||
message=project.error_message,
|
||
detail={"chat_task_id": task.id},
|
||
)
|
||
return True
|
||
|
||
|
||
async def delete_project_v2(
|
||
db: AsyncSession,
|
||
*,
|
||
config: ModuleGenerationV2Config,
|
||
current_user: User,
|
||
project_id: str,
|
||
) -> dict[str, Any]:
|
||
"""软删除 V2 项目；不沿步骤父子链查询，直接按项目批量处理。"""
|
||
project = await get_v2_project_for_user(
|
||
db,
|
||
config=config,
|
||
project_id=project_id,
|
||
current_user=current_user,
|
||
for_update=True,
|
||
)
|
||
await assert_project_has_no_active_chat_tasks(
|
||
db,
|
||
project=project,
|
||
config=config.flow_config,
|
||
detail_message=f"当前{config.display_name}项目仍有生成中任务，暂不能删除",
|
||
)
|
||
deleted_at = utc_now()
|
||
project.deleted_at = deleted_at
|
||
await soft_delete_steps_from_index(
|
||
db,
|
||
project=project,
|
||
start_index=1,
|
||
deleted_at=deleted_at,
|
||
config=config.flow_config,
|
||
log_module_event=log_v2_event,
|
||
)
|
||
if config.module == SHOT_REPLICATE_V2.module:
|
||
segment_result = await execute_with_lock_timeout(
|
||
db,
|
||
select(ShotReplicateSegment)
|
||
.where(ShotReplicateSegment.module_project_id == project.id)
|
||
.with_for_update(),
|
||
)
|
||
for segment in segment_result.scalars().all():
|
||
segment.module_project_id = None
|
||
segment.replicate_status = ShotSegmentReplicateStatusEnum.NOT_STARTED.value
|
||
upload_module = (
|
||
UploadResourceModuleEnum.HOT_OPENING_REPLICATE.value
|
||
if config.module == HOT_OPENING_V2.module
|
||
else UploadResourceModuleEnum.SHOT_REPLICATE.value
|
||
)
|
||
release = await release_upload_resources_by_source(
|
||
db,
|
||
source_model=UploadResourceSourceModelEnum.MODULE_GENERATION_PROJECT.value,
|
||
source_ids=[project.id],
|
||
module=upload_module,
|
||
)
|
||
await log_v2_event(
|
||
db,
|
||
project=project,
|
||
event_type=ModuleEventTypeEnum.PROJECT_DELETED.value,
|
||
message=f"软删除{config.display_name} V2 项目",
|
||
detail={
|
||
"upload_resource_release": {
|
||
key: value for key, value in release.items() if key != "released_resource_ids"
|
||
}
|
||
},
|
||
)
|
||
return {
|
||
"message": "项目已删除",
|
||
"project_id": project.id,
|
||
"deleted": True,
|
||
"released_size_bytes": int(release.get("released_size_bytes") or 0),
|
||
"upload_resource_released": int(release.get("released") or 0),
|
||
"pending_delete_resource_ids": list(release.get("released_resource_ids") or []),
|
||
}
|