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}"