Files
video-gen/video-gen-api/app/services/hot_opening_replicate_service.py
T
2026-07-09 09:10:49 +08:00

1548 lines
67 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 String, cast, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
from app.config import settings
from app.enums.common import ModuleEventTypeEnum, ModuleProjectStatusEnum, ModulePromptTypeEnum, ModuleStepStatusEnum
from app.enums.hot_opening_replicate import HotOpeningGenerationModeEnum, HotOpeningStepCodeEnum, 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.hot_opening_replicate import (
HotOpeningDeleteOut,
HotOpeningGenerateImageRequest,
HotOpeningGenerateVideoPromptRequest,
HotOpeningGenerateVideoRequest,
HotOpeningImageGenerationOut,
HotOpeningImagePromptUpdateRequest,
HotOpeningMaterialOut,
HotOpeningMaterialUpdateRequest,
HotOpeningStepOut,
HotOpeningStepUpdate,
HotOpeningTaskCreate,
HotOpeningTaskDetailOut,
HotOpeningTaskListItemOut,
HotOpeningTaskListOut,
HotOpeningVideoGenerationOut,
HotOpeningVideoPromptSchemaUpdateRequest,
)
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, 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.module_generation_flow_base_service import (
chat_tasks_by_id as _base_chat_tasks_by_id,
create_module_step as _base_create_step,
get_current_step_by_code as _base_get_current_step_by_code,
get_current_steps as _base_get_current_steps,
get_project_for_user as _base_get_project_for_user,
get_step_for_user as _base_get_step_for_user,
next_version as _base_next_version,
soft_delete_steps_from_index as _base_soft_delete_steps_from_index,
)
from app.enums.module_generation_flow import ModuleGenerationFlowConfig
from app.services.module_generation_step_common_service import (
build_file_url_or_data_uri as _common_build_file_url_or_data_uri,
build_step_input as _common_build_step_input,
build_step_output as _common_build_step_output,
force_set_json as _common_force_set_json,
is_wrapped_step_io as _common_is_wrapped_step_io,
json_dumps as _common_json,
merge_dict as _common_merge_dict,
parse_json as _common_parse_json,
snapshot_from_chat as _common_snapshot_from_chat,
step_payload as _common_step_payload,
step_result as _common_step_result,
step_usage as _common_step_usage,
unwrap_step_output as _common_unwrap_step_output,
utc_now as _common_now,
)
from app.services.module_generation_step_update_service import (
update_module_image_prompt,
update_module_material_input,
update_module_step,
update_module_video_prompt_schema,
)
from app.services.resource_signed_url_service import build_resource_signed_url
from app.enums.upload_resource import UploadResourceModuleEnum, UploadResourceSourceModelEnum
from app.services.upload_resource import release_upload_resources_by_source
from app.services.video_prompt_schema_config_service import fallback_runtime_schema_snapshot, get_runtime_schema_snapshot
from app.utils.id_gen import generate_id
MODULE = ModuleCodeEnum.HOT_OPENING_REPLICATE.value
GENERATION_MODE = HotOpeningGenerationModeEnum.HOT_OPENING_REPLICATE.value
STEP_INDEX_MAP = {
HotOpeningStepCodeEnum.MATERIAL_INPUT.value: 1,
HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value: 2,
HotOpeningStepCodeEnum.IMAGE_GENERATE.value: 3,
HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value: 4,
HotOpeningStepCodeEnum.VIDEO_GENERATE.value: 5,
}
STEP_IO_SCHEMA_VERSION = "hot_opening_step_io_v1"
FLOW_CONFIG = ModuleGenerationFlowConfig(
module=MODULE,
step_index_map=STEP_INDEX_MAP,
material_step_code=HotOpeningStepCodeEnum.MATERIAL_INPUT.value,
image_prompt_step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
image_generate_step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
video_prompt_step_code=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
video_generate_step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
project_not_found_message="爆款开头复刻项目不存在",
step_not_found_message="子任务不存在",
cancel_chat_task_error_message="爆款开头复刻步骤被重新生成或删除,旧生成任务已取消",
material_video_url_editable=True,
step_io_schema_version=STEP_IO_SCHEMA_VERSION,
)
_now = _common_now
_json = _common_json
_parse_json = _common_parse_json
_merge_dict = _common_merge_dict
_force_set_json = _common_force_set_json
def _step_input(
*,
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 _common_build_step_input(
step_code=step_code,
payload=payload,
source_step_id=source_step_id,
parent_step_id=parent_step_id,
context=context,
schema_version=STEP_IO_SCHEMA_VERSION,
)
def _step_output(
*,
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 _common_build_step_output(
step_code=step_code,
status=status,
payload=payload,
result=result,
usage=usage,
error=error,
schema_version=STEP_IO_SCHEMA_VERSION,
)
def _is_wrapped_step_io(value: Any) -> bool:
return _common_is_wrapped_step_io(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _step_payload(value: Any) -> dict[str, Any]:
return _common_step_payload(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _step_result(value: Any) -> dict[str, Any]:
return _common_step_result(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _step_usage(value: Any) -> dict[str, Any]:
return _common_step_usage(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _unwrap_step_output(value: Any) -> dict[str, Any]:
return _common_unwrap_step_output(value, schema_version=STEP_IO_SCHEMA_VERSION)
def _resolve_video_schema_config_snapshot(video_prompt_output: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None, str | None, bool]:
"""
详情接口运行时补齐历史第4步缺失的 schema_config_snapshot。
注意:这里仅用于接口返回,不修改 step.output_json,避免把历史脏数据伪装成生成当时真实快照。
"""
prompt_schema = video_prompt_output.get("prompt_schema")
if not isinstance(prompt_schema, dict) or not prompt_schema:
return None, None, None, False
raw_snapshot = video_prompt_output.get("schema_config_snapshot")
has_real_snapshot = isinstance(raw_snapshot, dict) and isinstance(raw_snapshot.get("data"), dict)
snapshot = fallback_runtime_schema_snapshot(raw_snapshot if has_real_snapshot else None)
return (
snapshot,
str(snapshot.get("source") or video_prompt_output.get("schema_config_source") or "") or None,
str(snapshot.get("version") or video_prompt_output.get("schema_config_version") or "") or None,
not has_real_snapshot,
)
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:
return await _base_get_project_for_user(
db,
project_id=project_id,
user=user,
config=FLOW_CONFIG,
for_update=for_update,
populate_existing=populate_existing,
)
async def _get_step_for_user(
db: AsyncSession,
*,
project_id: str,
step_id: str,
user: User,
for_update: bool = False,
) -> ModuleGenerationStep:
return await _base_get_step_for_user(
db,
project_id=project_id,
step_id=step_id,
user=user,
config=FLOW_CONFIG,
for_update=for_update,
)
async def _get_current_steps(db: AsyncSession, project_id: str) -> list[ModuleGenerationStep]:
return await _base_get_current_steps(db, project_id=project_id, config=FLOW_CONFIG)
async def _get_current_step_by_code(db: AsyncSession, project_id: str, step_code: str) -> ModuleGenerationStep | None:
return await _base_get_current_step_by_code(db, project_id=project_id, step_code=step_code, config=FLOW_CONFIG)
async def _next_version(db: AsyncSession, project_id: str, step_code: str) -> int:
return await _base_next_version(db, project_id=project_id, step_code=step_code, config=FLOW_CONFIG)
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:
return await _base_create_step(
db,
project=project,
step_code=step_code,
config=FLOW_CONFIG,
log_module_event=log_module_event,
status=status,
parent_step_id=parent_step_id,
source_step_id=source_step_id,
chat_task_id=chat_task_id,
input_data=input_data,
output_data=output_data,
)
async def _soft_delete_steps_from_index(
db: AsyncSession,
*,
project: ModuleGenerationProject,
start_index: int,
deleted_at: datetime | None = None,
) -> None:
await _base_soft_delete_steps_from_index(
db,
project=project,
start_index=start_index,
config=FLOW_CONFIG,
log_module_event=log_module_event,
deleted_at=deleted_at,
)
def _step_to_out(step: ModuleGenerationStep) -> HotOpeningStepOut:
return HotOpeningStepOut(
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]:
return _common_snapshot_from_chat(chat_task)
async def _chat_tasks_by_id(db: AsyncSession, steps: list[ModuleGenerationStep]) -> dict[str, ChatGenerationTask]:
return await _base_chat_tasks_by_id(db, steps)
async def project_to_detail_out(db: AsyncSession, project: ModuleGenerationProject) -> HotOpeningTaskDetailOut:
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(HotOpeningStepCodeEnum.MATERIAL_INPUT.value)
image_prompt_step = by_code.get(HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
image_generate_step = by_code.get(HotOpeningStepCodeEnum.IMAGE_GENERATE.value)
video_prompt_step = by_code.get(HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
video_generate_step = by_code.get(HotOpeningStepCodeEnum.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
schema_config_snapshot, schema_config_source, schema_config_version, schema_config_is_fallback = _resolve_video_schema_config_snapshot(video_prompt_output)
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 HotOpeningTaskDetailOut(
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=HotOpeningMaterialOut(
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=HotOpeningImageGenerationOut(
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=HotOpeningVideoGenerationOut(
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"),
schema_config_snapshot=schema_config_snapshot,
schema_config_source=schema_config_source,
schema_config_version=schema_config_version,
schema_config_is_fallback=schema_config_is_fallback,
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_hot_opening_project(db: AsyncSession, current_user: User, req: HotOpeningTaskCreate) -> 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=HotOpeningStepCodeEnum.MATERIAL_INPUT.value,
idempotency_key=req.idempotency_key,
)
db.add(project)
await db.flush()
await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.MATERIAL_INPUT.value,
status=ModuleStepStatusEnum.COMPLETED.value,
input_data={
"material_video_url": req.material_video_url,
"material_image_url": req.material_image_url,
"material_video_resource_id": req.material_video_resource_id,
"material_image_resource_id": req.material_image_resource_id,
"material_video_duration_seconds": req.material_video_duration_seconds,
"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
def _user_name_filter_subquery(value: str):
"""后台按用户名筛选时使用的子查询。
只在传入 user_name 时查询 users 表;keyword 不再关联 users,避免后台关键词搜索扩大查询范围。
"""
like = f"%{value.strip()}%"
return select(User.id).where(User.username.ilike(like))
async def _user_name_map_by_ids(db: AsyncSession, user_ids: set[str]) -> dict[str, str | None]:
"""一次性查询当前页涉及的用户,避免列表逐条查询用户表。"""
if not user_ids:
return {}
result = await db.execute(select(User.id, User.username).where(User.id.in_(list(user_ids))))
return {user_id: username for user_id, username in result.all()}
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_hot_opening_projects(
db: AsyncSession,
*,
current_user: User,
status: str | None,
keyword: str | None = None,
user_id: str | None = None,
user_name: str | None = None,
created_start: datetime | None = None,
created_end: datetime | None = None,
page: int,
page_size: int,
) -> HotOpeningTaskListOut:
"""
爆款开头列表查询。
性能策略:
1. 列表接口只查列表字段,不调用 project_to_detail_out,避免 steps/chat/user 多重 N+1。
2. 主表分页查询完成后,再按当前页 project_ids 批量查 material_input。
3. 管理员需要展示 user_name 时,按当前页 user_id 去重后一次性查 User。
4. user_name 使用 IN (SELECT users.id ...) 子查询筛选;keyword 不查询 users 表。
"""
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)
else:
if user_id and user_id.strip():
query = query.where(ModuleGenerationProject.user_id == user_id.strip())
if user_name and user_name.strip():
query = query.where(ModuleGenerationProject.user_id.in_(_user_name_filter_subquery(user_name)))
if created_start:
query = query.where(ModuleGenerationProject.created_at >= created_start)
if created_end:
query = query.where(ModuleGenerationProject.created_at <= created_end)
if status:
query = query.where(ModuleGenerationProject.status == status)
if keyword and keyword.strip():
like = f"%{keyword.strip()}%"
material_step_subquery = (
select(ModuleGenerationStep.project_id).where(
ModuleGenerationStep.step_code == HotOpeningStepCodeEnum.MATERIAL_INPUT.value,
ModuleGenerationStep.is_current.is_(True),
ModuleGenerationStep.deleted_at.is_(None),
cast(ModuleGenerationStep.input_json, String).ilike(like),
)
)
query = query.where(
or_(
ModuleGenerationProject.id.ilike(like),
ModuleGenerationProject.title.ilike(like),
ModuleGenerationProject.current_step_code.ilike(like),
ModuleGenerationProject.error_message.ilike(like),
ModuleGenerationProject.id.in_(material_step_subquery),
)
)
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=HotOpeningStepCodeEnum.MATERIAL_INPUT.value,
)
user_name_map: dict[str, str | None] = {}
if current_user.is_admin:
user_ids = {project.user_id for project in projects if project.user_id}
user_name_map = await _user_name_map_by_ids(db, user_ids)
items: list[HotOpeningTaskListItemOut] = []
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(
HotOpeningTaskListItemOut(
id=project.id,
project_id=project.id,
user_id=project.user_id,
user_name=user_name_map.get(project.user_id) if current_user.is_admin and project.user_id else None,
module=project.module,
title=project.title,
status=project.status,
current_step_code=project.current_step_code,
source_project_name=material.get("source_project_name"),
target_project_name=material.get("target_project_name"),
core_content_point=material.get("core_content_point"),
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,
created_at=project.created_at,
updated_at=project.updated_at,
completed_at=project.completed_at,
)
)
return HotOpeningTaskListOut(total=total, items=items)
async def update_hot_opening_step(
db: AsyncSession,
*,
current_user: User,
project_id: str,
step_id: str,
req: HotOpeningStepUpdate,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
return await update_module_step(
db,
current_user=current_user,
project_id=project_id,
step_id=step_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
)
async def update_hot_opening_material_input(
db: AsyncSession,
*,
current_user: User,
project_id: str,
req: HotOpeningMaterialUpdateRequest,
) -> tuple[str, str]:
"""修改第1步素材输入。
采用方案 B:软删除旧第1步及之后的当前有效步骤,然后新建第1步 version+1。
未传字段沿用旧第1步素材输入,避免前端只改一个字段时丢失其它素材信息。
"""
return await update_module_material_input(
db,
current_user=current_user,
project_id=project_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
)
async def update_hot_opening_image_prompt(
db: AsyncSession,
*,
current_user: User,
project_id: str,
step_id: str,
req: HotOpeningImagePromptUpdateRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
"""直接修改第2步图片 AI 优化提词,不调用 AI、不扣积分。
修改后软删除第3、4、5步当前有效任务,让用户从图片生成开始重新执行。
"""
return await update_module_image_prompt(
db,
current_user=current_user,
project_id=project_id,
step_id=step_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
)
async def update_hot_opening_video_prompt_schema(
db: AsyncSession,
*,
current_user: User,
project_id: str,
step_id: str,
req: HotOpeningVideoPromptSchemaUpdateRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
"""以前端 schema 为 patch 修改第4步视频 AI 提词,不调用 AI、不扣积分。
服务端已有 schema 为基准:视频规格、数组长度、时间段、合规控制、质量控制、协议字段均锁定。
最终提示词允许修改,但保存前会清洗视频时长、比例、分辨率、帧率等参数。
"""
return await update_module_video_prompt_schema(
db,
current_user=current_user,
project_id=project_id,
step_id=step_id,
req=req,
config=FLOW_CONFIG,
log_module_event=log_module_event,
patch_video_prompt_schema_from_client=patch_video_prompt_schema_from_client,
build_final_video_prompt=build_final_video_prompt,
)
async def submit_image_prompt_optimize(
db: AsyncSession,
*,
current_user: User,
project_id: str,
material_step_id: str,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
material_step = await _get_step_for_user(db, project_id=project_id, step_id=material_step_id, user=current_user, for_update=True)
if material_step.step_code != HotOpeningStepCodeEnum.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[HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value])
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.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 = HotOpeningStepCodeEnum.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, HotOpeningStepCodeEnum.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 == HotOpeningStepCodeEnum.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, HotOpeningStepCodeEnum.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=HotOpeningStepCodeEnum.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=HotOpeningStepCodeEnum.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 = HotOpeningStepCodeEnum.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,
prompt_step_id: str,
req: HotOpeningGenerateImageRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
prompt_step = await _get_step_for_user(db, project_id=project_id, step_id=prompt_step_id, user=current_user, for_update=True)
if prompt_step.step_code != HotOpeningStepCodeEnum.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[HotOpeningStepCodeEnum.IMAGE_GENERATE.value])
material_step = await _get_current_step_by_code(db, project.id, HotOpeningStepCodeEnum.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": "新产品图片"},
]
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
input_data={
"engine_id": req.engine_id,
"params": {
"image_size": req.image_size,
"image_proportion": req.image_proportion,
"image_px": req.image_px,
},
"prompt": optimized_prompt,
"media_references": refs,
},
)
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="爆款开头复刻图片生成",
billing_source_module=project.module,
billing_source_project_id=project.id,
billing_source_step_id=step.id,
billing_source_step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
)
step.chat_task_id = chat_task.id
_force_set_json(
step,
"input_json",
_step_input(
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
source_step_id=prompt_step.id,
parent_step_id=prompt_step.id,
payload={
"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 = HotOpeningStepCodeEnum.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
async def _resolve_video_prompt_config(db: AsyncSession, req: HotOpeningGenerateVideoPromptRequest) -> 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, "HOT_OPENING_DEFAULT_VIDEO_RATIO", None) or VIDEO_DEFAULT_RATIO
default_resolution = getattr(settings, "HOT_OPENING_DEFAULT_VIDEO_RESOLUTION", None) or VIDEO_DEFAULT_RESOLUTION
default_duration = int(getattr(settings, "HOT_OPENING_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,
image_step_id: str,
req: HotOpeningGenerateVideoPromptRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
image_step = await _get_step_for_user(db, project_id=project_id, step_id=image_step_id, user=current_user, for_update=True)
if image_step.step_code != HotOpeningStepCodeEnum.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[HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value])
video_config = await _resolve_video_prompt_config(db, req)
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.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, "HOT_OPENING_DEFAULT_TARGET_PLATFORM", "抖音") or "抖音",
},
)
project.status = ModuleProjectStatusEnum.PROCESSING.value
project.current_step_code = HotOpeningStepCodeEnum.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, HotOpeningStepCodeEnum.MATERIAL_INPUT.value)
image_prompt_step = await _get_current_step_by_code(db, project.id, HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
image_step = await _get_current_step_by_code(db, project.id, HotOpeningStepCodeEnum.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 == HotOpeningStepCodeEnum.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, HotOpeningStepCodeEnum.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, "HOT_OPENING_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,
)
schema_config_snapshot = await get_runtime_schema_snapshot(db)
request_log["schema_config_source"] = schema_config_snapshot.get("source")
prompt_schema, final_prompt, token_usage = await optimize_hot_opening_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,
schema_config_snapshot=schema_config_snapshot,
module=project.module,
project_id=project.id,
step_id=step.id,
trace_id=f"hot-video-prompt:{step.id}",
)
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=HotOpeningStepCodeEnum.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,
"schema_config_snapshot": schema_config_snapshot,
"schema_config_source": schema_config_snapshot.get("source"),
"schema_config_version": schema_config_snapshot.get("version"),
},
usage=usage,
),
)
project.status = ModuleProjectStatusEnum.WAITING_USER.value
project.current_step_code = HotOpeningStepCodeEnum.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, "schema_config_source": schema_config_snapshot.get("source")},
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,
prompt_step_id: str,
req: HotOpeningGenerateVideoRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
prompt_step = await _get_step_for_user(db, project_id=project_id, step_id=prompt_step_id, user=current_user, for_update=True)
if prompt_step.step_code != HotOpeningStepCodeEnum.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[HotOpeningStepCodeEnum.VIDEO_GENERATE.value])
image_step = await _get_current_step_by_code(db, project.id, HotOpeningStepCodeEnum.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.HOT_OPENING_DEFAULT_VIDEO_DURATION or 4)
aspect_ratio = prompt_params.get("aspect_ratio") or settings.HOT_OPENING_DEFAULT_VIDEO_RATIO or "9:16"
resolution = prompt_params.get("resolution") or settings.HOT_OPENING_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": "新项目图片"},
]
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
input_data={
"engine_id": req.engine_id or prompt_params.get("engine_id"),
"params": {
"duration": duration,
"aspect_ratio": aspect_ratio,
"resolution": resolution,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
)
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="爆款开头复刻视频生成",
billing_source_module=project.module,
billing_source_project_id=project.id,
billing_source_step_id=step.id,
billing_source_step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
)
step.chat_task_id = chat_task.id
_force_set_json(
step,
"input_json",
_step_input(
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
source_step_id=prompt_step.id,
parent_step_id=prompt_step.id,
payload={
"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 = HotOpeningStepCodeEnum.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
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 == HotOpeningStepCodeEnum.IMAGE_GENERATE.value:
step.status = ModuleStepStatusEnum.COMPLETED.value
step.completed_at = _now()
_force_set_json(
step,
"output_json",
_step_output(
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
status=ModuleStepStatusEnum.COMPLETED.value,
result={"result_image_url": task.image_url, "chat_task_id": task.id},
),
)
project.final_image_url = task.image_url
project.status = ModuleProjectStatusEnum.WAITING_USER.value
project.current_step_code = HotOpeningStepCodeEnum.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 == HotOpeningStepCodeEnum.VIDEO_GENERATE.value:
step.status = ModuleStepStatusEnum.COMPLETED.value
step.completed_at = _now()
_force_set_json(
step,
"output_json",
_step_output(
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
status=ModuleStepStatusEnum.COMPLETED.value,
result={"result_video_url": task.video_url, "result_video_cover_url": task.video_cover_url, "chat_task_id": task.id},
),
)
project.final_video_url = task.video_url
project.final_video_cover_url = task.video_cover_url
project.status = ModuleProjectStatusEnum.COMPLETED.value
project.current_step_code = HotOpeningStepCodeEnum.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_hot_opening_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_hot_opening_project(db: AsyncSession, *, current_user: User, project_id: str) -> HotOpeningDeleteOut:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
project_id_snapshot = project.id
deleted_at = _now()
project.deleted_at = deleted_at
await _soft_delete_steps_from_index(db, project=project, start_index=1, deleted_at=deleted_at)
upload_release = await release_upload_resources_by_source(
db,
source_model=UploadResourceSourceModelEnum.MODULE_GENERATION_PROJECT.value,
source_ids=[project_id_snapshot],
module=UploadResourceModuleEnum.HOT_OPENING_REPLICATE.value,
)
pending_ids = list(upload_release.get("released_resource_ids") or [])
released_size = int(upload_release.get("released_size_bytes") or 0)
upload_resource_released = int(upload_release.get("released") or 0)
await log_module_event(
db,
project=project,
event_type=ModuleEventTypeEnum.PROJECT_DELETED.value,
message="软删除爆款开头复刻项目",
detail={
"upload_resource_release": {k: v for k, v in upload_release.items() if k != "released_resource_ids"},
"pending_delete_resource_count": len(pending_ids),
},
)
return HotOpeningDeleteOut(
message="项目已删除",
project_id=project_id_snapshot,
deleted=True,
released_size_bytes=released_size,
upload_resource_released=upload_resource_released,
pending_delete_resource_ids=pending_ids,
)
def _build_file_url_or_data_uri(file_url: str) -> str:
return _common_build_file_url_or_data_uri(file_url)