1601 lines
70 KiB
Python
1601 lines
70 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from 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.resource_signed_url_service import build_resource_signed_url
|
|
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"
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _json(data: Any) -> str | None:
|
|
if data is None:
|
|
return None
|
|
return json.dumps(data, ensure_ascii=False, default=str)
|
|
|
|
|
|
def _parse_json(value: Any, fallback: Any = None) -> Any:
|
|
if value is None or value == "":
|
|
return fallback
|
|
if isinstance(value, (dict, list)):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
return json.loads(value)
|
|
except Exception:
|
|
return fallback
|
|
return fallback
|
|
|
|
|
|
def _step_input(
|
|
*,
|
|
step_code: str,
|
|
payload: dict[str, Any] | None = None,
|
|
source_step_id: str | None = None,
|
|
parent_step_id: str | None = None,
|
|
context: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": STEP_IO_SCHEMA_VERSION,
|
|
"step_code": step_code,
|
|
"source": {
|
|
"source_step_id": source_step_id,
|
|
"parent_step_id": parent_step_id,
|
|
},
|
|
"payload": payload or {},
|
|
"context": context or {},
|
|
}
|
|
|
|
|
|
def _step_output(
|
|
*,
|
|
step_code: str,
|
|
status: str,
|
|
payload: dict[str, Any] | None = None,
|
|
result: dict[str, Any] | None = None,
|
|
usage: dict[str, Any] | None = None,
|
|
error: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": STEP_IO_SCHEMA_VERSION,
|
|
"step_code": step_code,
|
|
"status": status,
|
|
"payload": payload or {},
|
|
"result": result or {},
|
|
"usage": usage or {},
|
|
"error": error or {},
|
|
}
|
|
|
|
|
|
def _is_wrapped_step_io(value: Any) -> bool:
|
|
return isinstance(value, dict) and value.get("schema_version") == STEP_IO_SCHEMA_VERSION
|
|
|
|
|
|
def _step_payload(value: Any) -> dict[str, Any]:
|
|
data = _parse_json(value, {}) or {}
|
|
if _is_wrapped_step_io(data):
|
|
payload = data.get("payload")
|
|
return payload if isinstance(payload, dict) else {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _step_result(value: Any) -> dict[str, Any]:
|
|
data = _parse_json(value, {}) or {}
|
|
if _is_wrapped_step_io(data):
|
|
result = data.get("result")
|
|
if isinstance(result, dict) and result:
|
|
return result
|
|
payload = data.get("payload")
|
|
return payload if isinstance(payload, dict) else {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _step_usage(value: Any) -> dict[str, Any]:
|
|
data = _parse_json(value, {}) or {}
|
|
if _is_wrapped_step_io(data):
|
|
usage = data.get("usage")
|
|
return usage if isinstance(usage, dict) else {}
|
|
usage = data.get("token_usage") if isinstance(data, dict) else {}
|
|
return usage if isinstance(usage, dict) else {}
|
|
|
|
|
|
def _unwrap_step_output(value: Any) -> dict[str, Any]:
|
|
data = _parse_json(value, {}) or {}
|
|
if not _is_wrapped_step_io(data):
|
|
return data if isinstance(data, dict) else {}
|
|
merged: dict[str, Any] = {}
|
|
payload = data.get("payload")
|
|
result = data.get("result")
|
|
usage = data.get("usage")
|
|
if isinstance(payload, dict):
|
|
merged.update(payload)
|
|
if isinstance(result, dict):
|
|
merged.update(result)
|
|
if isinstance(usage, dict) and usage:
|
|
merged["token_usage"] = usage
|
|
return merged
|
|
|
|
|
|
def _merge_dict(old: dict[str, Any] | None, new: dict[str, Any] | None) -> dict[str, Any]:
|
|
merged = dict(old or {})
|
|
for key, value in (new or {}).items():
|
|
if value is not None:
|
|
merged[key] = value
|
|
return merged
|
|
|
|
|
|
async def log_module_event(
|
|
db: AsyncSession,
|
|
*,
|
|
project: ModuleGenerationProject,
|
|
event_type: str,
|
|
step: ModuleGenerationStep | None = None,
|
|
message: str | None = None,
|
|
detail: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""模块事件日志只落盘，不再写 module_generation_events 表。"""
|
|
_ = db
|
|
log_module_event_file(
|
|
module=project.module,
|
|
event_type=event_type,
|
|
project_id=project.id,
|
|
step_id=step.id if step else None,
|
|
user_id=project.user_id,
|
|
message=message,
|
|
detail=detail,
|
|
)
|
|
|
|
|
|
|
|
def _log_project_error(
|
|
*,
|
|
project: ModuleGenerationProject | None,
|
|
event_type: str,
|
|
message: str,
|
|
exc: BaseException | None = None,
|
|
step: ModuleGenerationStep | None = None,
|
|
detail: dict[str, Any] | None = None,
|
|
) -> None:
|
|
log_module_error(
|
|
module=(project.module if project else MODULE),
|
|
event_type=event_type,
|
|
project_id=(project.id if project else None),
|
|
step_id=(step.id if step else None),
|
|
user_id=(project.user_id if project else None),
|
|
message=message,
|
|
detail=detail,
|
|
exc=exc,
|
|
)
|
|
|
|
|
|
async def _get_project_for_user(
|
|
db: AsyncSession,
|
|
*,
|
|
project_id: str,
|
|
user: User,
|
|
for_update: bool = False,
|
|
populate_existing: bool = False,
|
|
) -> ModuleGenerationProject:
|
|
query = select(ModuleGenerationProject).where(
|
|
ModuleGenerationProject.id == project_id,
|
|
ModuleGenerationProject.module == MODULE,
|
|
ModuleGenerationProject.deleted_at.is_(None),
|
|
)
|
|
if not user.is_admin:
|
|
query = query.where(ModuleGenerationProject.user_id == user.id)
|
|
if populate_existing:
|
|
query = query.execution_options(populate_existing=True)
|
|
if for_update:
|
|
query = query.with_for_update()
|
|
result = await db.execute(query.limit(1))
|
|
project = result.scalar_one_or_none()
|
|
if not project:
|
|
raise HTTPException(status_code=404, detail="爆款开头复刻项目不存在")
|
|
return project
|
|
|
|
|
|
async def _get_step_for_user(
|
|
db: AsyncSession,
|
|
*,
|
|
project_id: str,
|
|
step_id: str,
|
|
user: User,
|
|
for_update: bool = False,
|
|
) -> ModuleGenerationStep:
|
|
await _get_project_for_user(db, project_id=project_id, user=user, for_update=for_update)
|
|
query = select(ModuleGenerationStep).where(
|
|
ModuleGenerationStep.id == step_id,
|
|
ModuleGenerationStep.project_id == project_id,
|
|
ModuleGenerationStep.module == MODULE,
|
|
ModuleGenerationStep.deleted_at.is_(None),
|
|
ModuleGenerationStep.is_current == True,
|
|
)
|
|
if not user.is_admin:
|
|
query = query.where(ModuleGenerationStep.user_id == user.id)
|
|
if for_update:
|
|
query = query.with_for_update()
|
|
result = await db.execute(query.limit(1))
|
|
step = result.scalar_one_or_none()
|
|
if not step:
|
|
raise HTTPException(status_code=404, detail="子任务不存在")
|
|
return step
|
|
|
|
|
|
async def _get_current_steps(db: AsyncSession, project_id: str) -> list[ModuleGenerationStep]:
|
|
result = await db.execute(
|
|
select(ModuleGenerationStep)
|
|
.where(
|
|
ModuleGenerationStep.project_id == project_id,
|
|
ModuleGenerationStep.module == MODULE,
|
|
ModuleGenerationStep.deleted_at.is_(None),
|
|
ModuleGenerationStep.is_current == True,
|
|
)
|
|
.order_by(ModuleGenerationStep.step_index.asc(), ModuleGenerationStep.created_at.asc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _get_current_step_by_code(db: AsyncSession, project_id: str, step_code: str) -> ModuleGenerationStep | None:
|
|
result = await db.execute(
|
|
select(ModuleGenerationStep)
|
|
.where(
|
|
ModuleGenerationStep.project_id == project_id,
|
|
ModuleGenerationStep.module == MODULE,
|
|
ModuleGenerationStep.step_code == step_code,
|
|
ModuleGenerationStep.is_current == True,
|
|
ModuleGenerationStep.deleted_at.is_(None),
|
|
)
|
|
.order_by(ModuleGenerationStep.version.desc(), ModuleGenerationStep.created_at.desc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def _next_version(db: AsyncSession, project_id: str, step_code: str) -> int:
|
|
result = await db.execute(
|
|
select(func.max(ModuleGenerationStep.version)).where(
|
|
ModuleGenerationStep.project_id == project_id,
|
|
ModuleGenerationStep.module == MODULE,
|
|
ModuleGenerationStep.step_code == step_code,
|
|
)
|
|
)
|
|
return int(result.scalar_one_or_none() or 0) + 1
|
|
|
|
|
|
async def _create_step(
|
|
db: AsyncSession,
|
|
*,
|
|
project: ModuleGenerationProject,
|
|
step_code: str,
|
|
status: str = ModuleStepStatusEnum.PENDING.value,
|
|
parent_step_id: str | None = None,
|
|
source_step_id: str | None = None,
|
|
chat_task_id: str | None = None,
|
|
input_data: dict[str, Any] | None = None,
|
|
output_data: dict[str, Any] | None = None,
|
|
) -> ModuleGenerationStep:
|
|
version = await _next_version(db, project.id, step_code)
|
|
step = ModuleGenerationStep(
|
|
id=generate_id(),
|
|
project_id=project.id,
|
|
user_id=project.user_id,
|
|
module=project.module,
|
|
step_index=STEP_INDEX_MAP[step_code],
|
|
step_code=step_code,
|
|
status=status,
|
|
version=version,
|
|
is_current=True,
|
|
parent_step_id=parent_step_id,
|
|
source_step_id=source_step_id,
|
|
chat_task_id=chat_task_id,
|
|
input_json=_step_input(
|
|
step_code=step_code,
|
|
payload=input_data,
|
|
source_step_id=source_step_id,
|
|
parent_step_id=parent_step_id,
|
|
) if input_data is not None else None,
|
|
output_json=_step_output(
|
|
step_code=step_code,
|
|
status=status,
|
|
result=output_data,
|
|
) if output_data is not None else None,
|
|
started_at=_now() if status == ModuleStepStatusEnum.PROCESSING.value else None,
|
|
completed_at=_now() if status == ModuleStepStatusEnum.COMPLETED.value else None,
|
|
)
|
|
db.add(step)
|
|
project.current_step_code = step_code
|
|
await db.flush()
|
|
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.STEP_CREATED.value, detail={"step_code": step_code, "version": version})
|
|
return step
|
|
|
|
|
|
async def _soft_delete_steps_from_index(
|
|
db: AsyncSession,
|
|
*,
|
|
project: ModuleGenerationProject,
|
|
start_index: int,
|
|
deleted_at: datetime | None = None,
|
|
) -> None:
|
|
deleted_at = deleted_at or _now()
|
|
result = await db.execute(
|
|
select(ModuleGenerationStep)
|
|
.where(
|
|
ModuleGenerationStep.project_id == project.id,
|
|
ModuleGenerationStep.module == MODULE,
|
|
ModuleGenerationStep.is_current == True,
|
|
ModuleGenerationStep.deleted_at.is_(None),
|
|
ModuleGenerationStep.step_index >= start_index,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
steps = list(result.scalars().all())
|
|
for step in steps:
|
|
step.is_current = False
|
|
step.deleted_at = deleted_at
|
|
if step.chat_task_id:
|
|
chat_result = await db.execute(
|
|
select(ChatGenerationTask)
|
|
.where(ChatGenerationTask.id == step.chat_task_id, ChatGenerationTask.deleted_at.is_(None))
|
|
.with_for_update()
|
|
.limit(1)
|
|
)
|
|
chat_task = chat_result.scalar_one_or_none()
|
|
if chat_task:
|
|
if chat_task.status == "completed":
|
|
await soft_delete_chat_task_resources(db, chat_task.id, deleted_at=deleted_at)
|
|
elif chat_task.status != "failed":
|
|
await mark_chat_generation_task_failed_and_refund_once(
|
|
db,
|
|
task=chat_task,
|
|
error_message="爆款开头复刻步骤被重新生成或删除，旧生成任务已取消",
|
|
pipeline_stage="failed",
|
|
)
|
|
chat_task.deleted_at = deleted_at
|
|
if steps:
|
|
await log_module_event(
|
|
db,
|
|
project=project,
|
|
event_type=ModuleEventTypeEnum.SOFT_DELETE_STEPS.value,
|
|
message=f"软删除第 {start_index} 步及之后的旧子任务",
|
|
detail={"step_ids": [step.id for step in steps]},
|
|
)
|
|
|
|
|
|
def _step_to_out(step: ModuleGenerationStep) -> 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]:
|
|
if not chat_task:
|
|
return {}
|
|
return _parse_json(chat_task.engine_snapshot_json, {}) or {}
|
|
|
|
|
|
async def _chat_tasks_by_id(db: AsyncSession, steps: list[ModuleGenerationStep]) -> dict[str, ChatGenerationTask]:
|
|
ids = [step.chat_task_id for step in steps if step.chat_task_id]
|
|
if not ids:
|
|
return {}
|
|
result = await db.execute(select(ChatGenerationTask).where(ChatGenerationTask.id.in_(ids)))
|
|
return {task.id: task for task in result.scalars().all()}
|
|
|
|
|
|
async def project_to_detail_out(db: AsyncSession, project: ModuleGenerationProject) -> 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
|
|
|
|
return HotOpeningTaskDetailOut(
|
|
id=project.id,
|
|
project_id=project.id,
|
|
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"),
|
|
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,
|
|
"source_project_name": req.source_project_name,
|
|
"target_project_name": req.target_project_name,
|
|
"core_content_point": req.core_content_point,
|
|
},
|
|
output_data={"message": "素材输入已提交，后端不做素材文件校验。下一步请手动生成图片AI提词。"},
|
|
)
|
|
await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_CREATED.value, message="创建爆款开头复刻项目")
|
|
return project
|
|
|
|
|
|
async def list_hot_opening_projects(
|
|
db: AsyncSession,
|
|
*,
|
|
current_user: User,
|
|
status: str | None,
|
|
page: int,
|
|
page_size: int,
|
|
) -> HotOpeningTaskListOut:
|
|
query = select(ModuleGenerationProject).where(
|
|
ModuleGenerationProject.module == MODULE,
|
|
ModuleGenerationProject.deleted_at.is_(None),
|
|
)
|
|
if not current_user.is_admin:
|
|
query = query.where(ModuleGenerationProject.user_id == current_user.id)
|
|
if status:
|
|
query = query.where(ModuleGenerationProject.status == status)
|
|
|
|
total = (await db.execute(select(func.count()).select_from(query.subquery()))).scalar_one()
|
|
result = await db.execute(query.order_by(ModuleGenerationProject.created_at.desc()).offset((page - 1) * page_size).limit(page_size))
|
|
projects = list(result.scalars().all())
|
|
|
|
items: list[HotOpeningTaskListItemOut] = []
|
|
for project in projects:
|
|
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)
|
|
items.append(
|
|
HotOpeningTaskListItemOut(
|
|
id=project.id,
|
|
project_id=project.id,
|
|
module=project.module,
|
|
title=project.title,
|
|
status=project.status,
|
|
current_step_code=project.current_step_code,
|
|
target_project_name=material.get("target_project_name"),
|
|
final_image_url=build_resource_signed_url(project.final_image_url) if project.final_image_url else None,
|
|
final_video_url=build_resource_signed_url(project.final_video_url) if project.final_video_url else None,
|
|
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]:
|
|
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
|
step = await _get_step_for_user(db, project_id=project_id, step_id=step_id, user=current_user, for_update=True)
|
|
if step.status == ModuleStepStatusEnum.PROCESSING.value:
|
|
raise HTTPException(status_code=400, detail="当前子任务正在处理中，暂不能修改")
|
|
|
|
input_data = _step_payload(step.input_json)
|
|
output_data = _unwrap_step_output(step.output_json)
|
|
|
|
if step.step_code == HotOpeningStepCodeEnum.MATERIAL_INPUT.value:
|
|
input_data = _merge_dict(
|
|
input_data,
|
|
{
|
|
"material_video_url": req.material_video_url,
|
|
"material_image_url": req.material_image_url,
|
|
"source_project_name": req.source_project_name,
|
|
"target_project_name": req.target_project_name,
|
|
"core_content_point": req.core_content_point,
|
|
},
|
|
)
|
|
if req.target_project_name:
|
|
project.title = req.target_project_name
|
|
elif step.step_code == HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
|
|
if req.prompt is not None:
|
|
output_data["optimized_prompt"] = req.prompt
|
|
output_data["prompt"] = req.prompt
|
|
elif step.step_code == HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
|
|
if req.prompt_schema is not None:
|
|
output_data["prompt_schema"] = req.prompt_schema
|
|
if req.prompt is not None:
|
|
output_data["final_prompt"] = req.prompt
|
|
else:
|
|
if req.input_json:
|
|
input_data = _merge_dict(input_data, req.input_json)
|
|
if req.output_json:
|
|
output_data = _merge_dict(output_data, req.output_json)
|
|
|
|
if req.input_json:
|
|
input_data = _merge_dict(input_data, req.input_json)
|
|
if req.output_json:
|
|
output_data = _merge_dict(output_data, req.output_json)
|
|
|
|
step.input_json = _step_input(step_code=step.step_code, payload=input_data, source_step_id=step.source_step_id, parent_step_id=step.parent_step_id)
|
|
step.output_json = _step_output(step_code=step.step_code, status=ModuleStepStatusEnum.COMPLETED.value, payload=output_data)
|
|
step.status = ModuleStepStatusEnum.COMPLETED.value
|
|
step.error_message = None
|
|
step.completed_at = _now()
|
|
project.status = ModuleProjectStatusEnum.WAITING_USER.value
|
|
project.current_step_code = step.step_code
|
|
project.error_message = None
|
|
|
|
await _soft_delete_steps_from_index(db, project=project, start_index=step.step_index + 1)
|
|
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.STEP_UPDATED.value, message="用户修改子任务内容")
|
|
return project, step
|
|
|
|
|
|
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步素材输入，避免前端只改一个字段时丢失其它素材信息。
|
|
"""
|
|
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
|
old_material_step = await _get_current_step_by_code(db, project.id, HotOpeningStepCodeEnum.MATERIAL_INPUT.value)
|
|
old_material = _step_payload(old_material_step.input_json if old_material_step else None)
|
|
|
|
material = {
|
|
"material_video_url": req.material_video_url if req.material_video_url is not None else old_material.get("material_video_url"),
|
|
"material_image_url": req.material_image_url if req.material_image_url is not None else old_material.get("material_image_url"),
|
|
"source_project_name": req.source_project_name if req.source_project_name is not None else old_material.get("source_project_name"),
|
|
"target_project_name": req.target_project_name if req.target_project_name is not None else old_material.get("target_project_name"),
|
|
"core_content_point": req.core_content_point if req.core_content_point is not None else old_material.get("core_content_point"),
|
|
}
|
|
|
|
missing_fields = [key for key, value in material.items() if value is None or str(value).strip() == ""]
|
|
if missing_fields:
|
|
raise HTTPException(status_code=400, detail=f"素材输入缺少必要字段: {', '.join(missing_fields)}")
|
|
|
|
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[HotOpeningStepCodeEnum.MATERIAL_INPUT.value])
|
|
|
|
project.title = str(material["target_project_name"])
|
|
project.status = ModuleProjectStatusEnum.WAITING_USER.value
|
|
project.current_step_code = HotOpeningStepCodeEnum.MATERIAL_INPUT.value
|
|
project.final_image_url = None
|
|
project.final_video_url = None
|
|
project.final_video_cover_url = None
|
|
project.error_message = None
|
|
project.completed_at = None
|
|
|
|
new_step = await _create_step(
|
|
db,
|
|
project=project,
|
|
step_code=HotOpeningStepCodeEnum.MATERIAL_INPUT.value,
|
|
status=ModuleStepStatusEnum.COMPLETED.value,
|
|
input_data=material,
|
|
output_data={"message": "素材输入已修改，旧步骤已软删除。下一步请重新生成图片AI提词。"},
|
|
)
|
|
await log_module_event(
|
|
db,
|
|
project=project,
|
|
step=new_step,
|
|
event_type=ModuleEventTypeEnum.STEP_UPDATED.value,
|
|
message="用户修改素材输入并重建第1步新版本",
|
|
detail={
|
|
"old_material_step_id": old_material_step.id if old_material_step else None,
|
|
"new_material_step_id": new_step.id,
|
|
"version": new_step.version,
|
|
},
|
|
)
|
|
return project.id, new_step.id
|
|
|
|
|
|
async def update_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步当前有效任务，让用户从图片生成开始重新执行。
|
|
"""
|
|
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
|
step = await _get_step_for_user(db, project_id=project_id, step_id=step_id, user=current_user, for_update=True)
|
|
if step.step_code != HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
|
|
raise HTTPException(status_code=400, detail="只能修改第2步图片 AI 提词子任务")
|
|
if step.status != ModuleStepStatusEnum.COMPLETED.value:
|
|
raise HTTPException(status_code=400, detail="图片 AI 提词未完成，不能直接修改")
|
|
|
|
output_data = _step_payload(step.output_json)
|
|
usage = _step_usage(step.output_json)
|
|
new_prompt = req.prompt.strip()
|
|
output_data["optimized_prompt"] = new_prompt
|
|
output_data["prompt"] = new_prompt
|
|
output_data["manual_edited"] = True
|
|
output_data["manual_edited_at"] = _now().isoformat()
|
|
|
|
step.output_json = _step_output(
|
|
step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
|
|
status=ModuleStepStatusEnum.COMPLETED.value,
|
|
payload=output_data,
|
|
usage=usage,
|
|
)
|
|
step.status = ModuleStepStatusEnum.COMPLETED.value
|
|
step.error_message = None
|
|
step.completed_at = _now()
|
|
|
|
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[HotOpeningStepCodeEnum.IMAGE_GENERATE.value])
|
|
|
|
project.status = ModuleProjectStatusEnum.WAITING_USER.value
|
|
project.current_step_code = HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
|
|
project.final_image_url = None
|
|
project.final_video_url = None
|
|
project.final_video_cover_url = None
|
|
project.completed_at = None
|
|
project.error_message = None
|
|
|
|
await log_module_event(
|
|
db,
|
|
project=project,
|
|
step=step,
|
|
event_type=ModuleEventTypeEnum.STEP_UPDATED.value,
|
|
message="用户直接修改图片 AI 优化提词，已软删除后续步骤",
|
|
detail={"start_deleted_step_index": STEP_INDEX_MAP[HotOpeningStepCodeEnum.IMAGE_GENERATE.value]},
|
|
)
|
|
return project, step
|
|
|
|
|
|
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 为基准：视频规格、数组长度、时间段、合规控制、质量控制、协议字段均锁定。
|
|
最终提示词允许修改，但保存前会清洗视频时长、比例、分辨率、帧率等参数。
|
|
"""
|
|
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
|
|
step = await _get_step_for_user(db, project_id=project_id, step_id=step_id, user=current_user, for_update=True)
|
|
if step.step_code != HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
|
|
raise HTTPException(status_code=400, detail="只能修改第4步视频 AI 提词 JSON schema 子任务")
|
|
if step.status != ModuleStepStatusEnum.COMPLETED.value:
|
|
raise HTTPException(status_code=400, detail="视频 AI 提词未完成，不能直接修改")
|
|
|
|
output_data = _step_payload(step.output_json)
|
|
usage = _step_usage(step.output_json)
|
|
input_data = _step_payload(step.input_json)
|
|
server_schema = output_data.get("prompt_schema") if isinstance(output_data.get("prompt_schema"), dict) else {}
|
|
video_config = output_data.get("params_used_for_prompt") or input_data.get("video_config") or {}
|
|
if not isinstance(video_config, dict) or not video_config.get("duration") or not video_config.get("aspect_ratio") or not video_config.get("resolution"):
|
|
raise HTTPException(status_code=400, detail="缺少第4步视频参数快照，不能安全修改视频 schema")
|
|
|
|
patched_schema = patch_video_prompt_schema_from_client(
|
|
server_schema=server_schema,
|
|
client_schema=req.prompt_schema,
|
|
video_config=video_config,
|
|
)
|
|
final_prompt = build_final_video_prompt(patched_schema)
|
|
|
|
output_data["prompt_schema"] = patched_schema
|
|
output_data["final_prompt"] = final_prompt
|
|
output_data["params_used_for_prompt"] = video_config
|
|
output_data["manual_edited"] = True
|
|
output_data["manual_edited_at"] = _now().isoformat()
|
|
|
|
step.output_json = _step_output(
|
|
step_code=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
|
|
status=ModuleStepStatusEnum.COMPLETED.value,
|
|
payload=output_data,
|
|
usage=usage,
|
|
)
|
|
step.status = ModuleStepStatusEnum.COMPLETED.value
|
|
step.error_message = None
|
|
step.completed_at = _now()
|
|
|
|
await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[HotOpeningStepCodeEnum.VIDEO_GENERATE.value])
|
|
|
|
project.status = ModuleProjectStatusEnum.WAITING_USER.value
|
|
project.current_step_code = HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value
|
|
project.final_video_url = None
|
|
project.final_video_cover_url = None
|
|
project.completed_at = None
|
|
project.error_message = None
|
|
|
|
await log_module_event(
|
|
db,
|
|
project=project,
|
|
step=step,
|
|
event_type=ModuleEventTypeEnum.STEP_UPDATED.value,
|
|
message="用户修改视频 AI 提词 schema，已软删除视频生成步骤",
|
|
detail={
|
|
"start_deleted_step_index": STEP_INDEX_MAP[HotOpeningStepCodeEnum.VIDEO_GENERATE.value],
|
|
"locked_fields": [
|
|
"schema_version",
|
|
"schema_usage",
|
|
"画面属性.视频时长",
|
|
"画面属性.视频比例",
|
|
"画面属性.清晰度",
|
|
"画面属性.帧率",
|
|
"画面属性.推荐分辨率",
|
|
"动作流程[*].时间段",
|
|
"镜头流程[*].时间段",
|
|
"动态时间规划",
|
|
"输出规格限制",
|
|
"质量控制",
|
|
"合规控制",
|
|
],
|
|
},
|
|
)
|
|
return project, step
|
|
|
|
|
|
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()
|
|
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": "新产品图片"},
|
|
]
|
|
|
|
chat_task = await create_chat_generation_task_for_module(
|
|
db,
|
|
current_user=current_user,
|
|
generation_mode=GENERATION_MODE,
|
|
gen_type="image",
|
|
original_prompt=prompt_output.get("original_prompt") or optimized_prompt,
|
|
optimized_prompt=optimized_prompt,
|
|
engine_id=req.engine_id,
|
|
media_references=refs,
|
|
image_size=req.image_size,
|
|
image_proportion=req.image_proportion,
|
|
image_px=req.image_px,
|
|
billing_project_name=project.title or "爆款开头复刻",
|
|
billing_description_prefix="爆款开头复刻图片生成",
|
|
)
|
|
step = await _create_step(
|
|
db,
|
|
project=project,
|
|
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
|
|
status=ModuleStepStatusEnum.PROCESSING.value,
|
|
parent_step_id=prompt_step.id,
|
|
source_step_id=prompt_step.id,
|
|
chat_task_id=chat_task.id,
|
|
input_data={
|
|
"engine_id": chat_task.engine_id,
|
|
"params": {"image_size": chat_task.image_size, "image_proportion": chat_task.image_proportion, "image_px": chat_task.image_px},
|
|
"prompt": optimized_prompt,
|
|
"media_references": refs,
|
|
},
|
|
)
|
|
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
|
project.current_step_code = 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,
|
|
)
|
|
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,
|
|
)
|
|
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()
|
|
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,
|
|
},
|
|
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},
|
|
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": "新项目图片"},
|
|
]
|
|
|
|
chat_task = await create_chat_generation_task_for_module(
|
|
db,
|
|
current_user=current_user,
|
|
generation_mode=GENERATION_MODE,
|
|
gen_type="video",
|
|
original_prompt=prompt_schema_str or final_prompt,
|
|
optimized_prompt=prompt_schema_str or final_prompt,
|
|
engine_id=req.engine_id or prompt_params.get("engine_id"),
|
|
media_references=refs,
|
|
duration=duration,
|
|
aspect_ratio=aspect_ratio,
|
|
resolution=resolution,
|
|
billing_project_name=project.title or "爆款开头复刻",
|
|
billing_description_prefix="爆款开头复刻视频生成",
|
|
)
|
|
step = await _create_step(
|
|
db,
|
|
project=project,
|
|
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
|
|
status=ModuleStepStatusEnum.PROCESSING.value,
|
|
parent_step_id=prompt_step.id,
|
|
source_step_id=prompt_step.id,
|
|
chat_task_id=chat_task.id,
|
|
input_data={
|
|
"engine_id": chat_task.engine_id,
|
|
"params": {
|
|
"duration": chat_task.duration,
|
|
"aspect_ratio": chat_task.aspect_ratio,
|
|
"resolution": chat_task.resolution,
|
|
"image_size": chat_task.image_size,
|
|
"image_proportion": chat_task.image_proportion,
|
|
"image_px": chat_task.image_px,
|
|
},
|
|
"prompt_schema": prompt_schema,
|
|
"final_prompt": final_prompt,
|
|
"media_references": refs,
|
|
},
|
|
)
|
|
project.status = ModuleProjectStatusEnum.PROCESSING.value
|
|
project.current_step_code = 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()
|
|
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()
|
|
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)
|
|
deleted_at = _now()
|
|
project.deleted_at = deleted_at
|
|
await _soft_delete_steps_from_index(db, project=project, start_index=1, deleted_at=deleted_at)
|
|
await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_DELETED.value, message="软删除爆款开头复刻项目")
|
|
return HotOpeningDeleteOut(message="项目已删除", project_id=project.id, deleted=True)
|
|
|
|
def _build_file_url_or_data_uri(file_url: str) -> str:
|
|
"""
|
|
Convert local upload path to base64 data URI.
|
|
Keep remote http/https/data URLs as-is.
|
|
"""
|
|
if file_url.startswith(("http://", "https://", "data:")):
|
|
return file_url
|
|
file_url_sign = build_resource_signed_url(resource_url=file_url, expire_seconds=86400)
|
|
return f"{settings.BASE_URL}{file_url_sign}" |