from __future__ import annotations

import json
from copy import deepcopy
from datetime import datetime, timezone
from typing import Any

from fastapi import HTTPException
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified

from app.config import settings
from app.enums.common import ModuleEventTypeEnum, ModuleProjectStatusEnum, ModulePromptTypeEnum, ModuleStepStatusEnum
from app.enums.shot_replicate import ShotReplicateGenerationModeEnum, ShotReplicateStepCodeEnum, ModuleCodeEnum
from app.models.chat_generation_task import ChatGenerationTask
from app.models.module_generation_project import ModuleGenerationProject
from app.models.module_generation_step import ModuleGenerationStep
from app.models.user import User
from app.schemas.shot_replicate import (
    ShotReplicateDeleteOut,
    ShotReplicateGenerateImageRequest,
    ShotReplicateGenerateVideoPromptRequest,
    ShotReplicateGenerateVideoRequest,
    ShotReplicateImageGenerationOut,
    ShotReplicateImagePromptUpdateRequest,
    ShotReplicateMaterialOut,
    ShotReplicateMaterialUpdateRequest,
    ShotReplicateStepOut,
    ShotReplicateStepUpdate,
    ShotReplicateTaskCreate,
    ShotReplicateTaskDetailOut,
    ShotReplicateTaskListItemOut,
    ShotReplicateTaskListOut,
    ShotReplicateVideoGenerationOut,
    ShotReplicateVideoPromptSchemaUpdateRequest,
)
from app.services.generation_ai_service import (
    VIDEO_DEFAULT_DURATION,
    VIDEO_DEFAULT_RATIO,
    VIDEO_DEFAULT_RESOLUTION,
    _get_video_engine,
    _parse_list,
)
from app.services.generation_billing_service import charge_module_prompt_usage
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
from app.services.generation_task_factory_service import create_chat_generation_task_for_module
from app.services.hot_opening_video_prompt_service import (
    build_final_video_prompt,
    optimize_hot_opening_video_prompt as optimize_shot_replicate_video_prompt,
    patch_video_prompt_schema_from_client,
)
from app.services.module_generation_log_service import log_module_error, log_module_event_file, log_module_prompt_event
from app.services.llm import optimize_prompt
from app.services.resource_accounting_service import soft_delete_chat_task_resources
from app.services.module_generation_flow_base_service import (
    chat_tasks_by_id as _base_chat_tasks_by_id,
    create_module_step as _base_create_step,
    get_current_step_by_code as _base_get_current_step_by_code,
    get_current_steps as _base_get_current_steps,
    get_project_for_user as _base_get_project_for_user,
    get_step_for_user as _base_get_step_for_user,
    next_version as _base_next_version,
    soft_delete_steps_from_index as _base_soft_delete_steps_from_index,
)
from app.enums.module_generation_flow import ModuleGenerationFlowConfig
from app.services.module_generation_step_common_service import (
    build_file_url_or_data_uri as _common_build_file_url_or_data_uri,
    build_step_input as _common_build_step_input,
    build_step_output as _common_build_step_output,
    force_set_json as _common_force_set_json,
    is_wrapped_step_io as _common_is_wrapped_step_io,
    json_dumps as _common_json,
    merge_dict as _common_merge_dict,
    parse_json as _common_parse_json,
    snapshot_from_chat as _common_snapshot_from_chat,
    step_payload as _common_step_payload,
    step_result as _common_step_result,
    step_usage as _common_step_usage,
    unwrap_step_output as _common_unwrap_step_output,
    utc_now as _common_now,
)
from app.services.module_generation_step_update_service import (
    update_module_image_prompt,
    update_module_material_input,
    update_module_step,
    update_module_video_prompt_schema,
)
from app.services.resource_signed_url_service import build_resource_signed_url
from app.services.video_prompt_schema_config_service import fallback_runtime_schema_snapshot, get_runtime_schema_snapshot
from app.utils.id_gen import generate_id
from app.models.shot_replicate_segment import ShotReplicateSegment
from app.enums.shot_replicate import ShotSegmentReplicateStatusEnum, ShotSplitStatusEnum
from app.schemas.shot_replicate import ShotSegmentReplicationCreateRequest

MODULE = ModuleCodeEnum.SHOT_REPLICATE.value
GENERATION_MODE = ShotReplicateGenerationModeEnum.SHOT_REPLICATE.value

STEP_INDEX_MAP = {
    ShotReplicateStepCodeEnum.MATERIAL_INPUT.value: 1,
    ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value: 2,
    ShotReplicateStepCodeEnum.IMAGE_GENERATE.value: 3,
    ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value: 4,
    ShotReplicateStepCodeEnum.VIDEO_GENERATE.value: 5,
}

STEP_IO_SCHEMA_VERSION = "shot_replicate_step_io_v1"

FLOW_CONFIG = ModuleGenerationFlowConfig(
    module=MODULE,
    step_index_map=STEP_INDEX_MAP,
    material_step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
    image_prompt_step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
    image_generate_step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
    video_prompt_step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
    video_generate_step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
    project_not_found_message="拆镜复刻项目不存在",
    step_not_found_message="子任务不存在",
    cancel_chat_task_error_message="拆镜复刻步骤被重新生成或删除，旧生成任务已取消",
    material_video_url_editable=False,
    step_io_schema_version=STEP_IO_SCHEMA_VERSION,
)


_now = _common_now
_json = _common_json
_parse_json = _common_parse_json
_merge_dict = _common_merge_dict
_force_set_json = _common_force_set_json


def _step_input(
    *,
    step_code: str,
    payload: dict[str, Any] | None = None,
    source_step_id: str | None = None,
    parent_step_id: str | None = None,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return _common_build_step_input(
        step_code=step_code,
        payload=payload,
        source_step_id=source_step_id,
        parent_step_id=parent_step_id,
        context=context,
        schema_version=STEP_IO_SCHEMA_VERSION,
    )


def _step_output(
    *,
    step_code: str,
    status: str,
    payload: dict[str, Any] | None = None,
    result: dict[str, Any] | None = None,
    usage: dict[str, Any] | None = None,
    error: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return _common_build_step_output(
        step_code=step_code,
        status=status,
        payload=payload,
        result=result,
        usage=usage,
        error=error,
        schema_version=STEP_IO_SCHEMA_VERSION,
    )


def _is_wrapped_step_io(value: Any) -> bool:
    return _common_is_wrapped_step_io(value, schema_version=STEP_IO_SCHEMA_VERSION)


def _step_payload(value: Any) -> dict[str, Any]:
    return _common_step_payload(value, schema_version=STEP_IO_SCHEMA_VERSION)


def _step_result(value: Any) -> dict[str, Any]:
    return _common_step_result(value, schema_version=STEP_IO_SCHEMA_VERSION)


def _step_usage(value: Any) -> dict[str, Any]:
    return _common_step_usage(value, schema_version=STEP_IO_SCHEMA_VERSION)


def _unwrap_step_output(value: Any) -> dict[str, Any]:
    return _common_unwrap_step_output(value, schema_version=STEP_IO_SCHEMA_VERSION)


def _resolve_video_schema_config_snapshot(video_prompt_output: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None, str | None, bool]:
    """
    详情接口运行时补齐历史第4步缺失的 schema_config_snapshot。

    注意：这里仅用于接口返回，不修改 step.output_json，避免把历史脏数据伪装成生成当时真实快照。
    """
    prompt_schema = video_prompt_output.get("prompt_schema")
    if not isinstance(prompt_schema, dict) or not prompt_schema:
        return None, None, None, False

    raw_snapshot = video_prompt_output.get("schema_config_snapshot")
    has_real_snapshot = isinstance(raw_snapshot, dict) and isinstance(raw_snapshot.get("data"), dict)
    snapshot = fallback_runtime_schema_snapshot(raw_snapshot if has_real_snapshot else None)
    return (
        snapshot,
        str(snapshot.get("source") or video_prompt_output.get("schema_config_source") or "") or None,
        str(snapshot.get("version") or video_prompt_output.get("schema_config_version") or "") or None,
        not has_real_snapshot,
    )




async def log_module_event(
    db: AsyncSession,
    *,
    project: ModuleGenerationProject,
    event_type: str,
    step: ModuleGenerationStep | None = None,
    message: str | None = None,
    detail: dict[str, Any] | None = None,
) -> None:
    """模块事件日志只落盘，不再写 module_generation_events 表。"""
    _ = db
    log_module_event_file(
        module=project.module,
        event_type=event_type,
        project_id=project.id,
        step_id=step.id if step else None,
        user_id=project.user_id,
        message=message,
        detail=detail,
    )



def _log_project_error(
    *,
    project: ModuleGenerationProject | None,
    event_type: str,
    message: str,
    exc: BaseException | None = None,
    step: ModuleGenerationStep | None = None,
    detail: dict[str, Any] | None = None,
) -> None:
    log_module_error(
        module=(project.module if project else MODULE),
        event_type=event_type,
        project_id=(project.id if project else None),
        step_id=(step.id if step else None),
        user_id=(project.user_id if project else None),
        message=message,
        detail=detail,
        exc=exc,
    )


async def _get_project_for_user(
    db: AsyncSession,
    *,
    project_id: str,
    user: User,
    for_update: bool = False,
    populate_existing: bool = False,
) -> ModuleGenerationProject:
    return await _base_get_project_for_user(
        db,
        project_id=project_id,
        user=user,
        config=FLOW_CONFIG,
        for_update=for_update,
        populate_existing=populate_existing,
    )


async def _get_step_for_user(
    db: AsyncSession,
    *,
    project_id: str,
    step_id: str,
    user: User,
    for_update: bool = False,
) -> ModuleGenerationStep:
    return await _base_get_step_for_user(
        db,
        project_id=project_id,
        step_id=step_id,
        user=user,
        config=FLOW_CONFIG,
        for_update=for_update,
    )


async def _get_current_steps(db: AsyncSession, project_id: str) -> list[ModuleGenerationStep]:
    return await _base_get_current_steps(db, project_id=project_id, config=FLOW_CONFIG)


async def _get_current_step_by_code(db: AsyncSession, project_id: str, step_code: str) -> ModuleGenerationStep | None:
    return await _base_get_current_step_by_code(db, project_id=project_id, step_code=step_code, config=FLOW_CONFIG)


async def _next_version(db: AsyncSession, project_id: str, step_code: str) -> int:
    return await _base_next_version(db, project_id=project_id, step_code=step_code, config=FLOW_CONFIG)


async def _create_step(
    db: AsyncSession,
    *,
    project: ModuleGenerationProject,
    step_code: str,
    status: str = ModuleStepStatusEnum.PENDING.value,
    parent_step_id: str | None = None,
    source_step_id: str | None = None,
    chat_task_id: str | None = None,
    input_data: dict[str, Any] | None = None,
    output_data: dict[str, Any] | None = None,
) -> ModuleGenerationStep:
    return await _base_create_step(
        db,
        project=project,
        step_code=step_code,
        config=FLOW_CONFIG,
        log_module_event=log_module_event,
        status=status,
        parent_step_id=parent_step_id,
        source_step_id=source_step_id,
        chat_task_id=chat_task_id,
        input_data=input_data,
        output_data=output_data,
    )


async def _soft_delete_steps_from_index(
    db: AsyncSession,
    *,
    project: ModuleGenerationProject,
    start_index: int,
    deleted_at: datetime | None = None,
) -> None:
    await _base_soft_delete_steps_from_index(
        db,
        project=project,
        start_index=start_index,
        config=FLOW_CONFIG,
        log_module_event=log_module_event,
        deleted_at=deleted_at,
    )


def _step_to_out(step: ModuleGenerationStep) -> ShotReplicateStepOut:
    return ShotReplicateStepOut(
        id=step.id,
        project_id=step.project_id,
        module=step.module,
        step_index=step.step_index,
        step_code=step.step_code,
        status=step.status,
        version=step.version,
        is_current=step.is_current,
        parent_step_id=step.parent_step_id,
        source_step_id=step.source_step_id,
        chat_task_id=step.chat_task_id,
        input=_parse_json(step.input_json, {}),
        output=_parse_json(step.output_json, {}),
        error_message=step.error_message,
        created_at=step.created_at,
        updated_at=step.updated_at,
        completed_at=step.completed_at,
    )


def _snapshot_from_chat(chat_task: ChatGenerationTask | None) -> dict[str, Any]:
    return _common_snapshot_from_chat(chat_task)


async def _chat_tasks_by_id(db: AsyncSession, steps: list[ModuleGenerationStep]) -> dict[str, ChatGenerationTask]:
    return await _base_chat_tasks_by_id(db, steps)


async def project_to_detail_out(db: AsyncSession, project: ModuleGenerationProject) -> ShotReplicateTaskDetailOut:
    steps = await _get_current_steps(db, project.id)
    by_code = {step.step_code: step for step in steps}
    chats = await _chat_tasks_by_id(db, steps)

    material_step = by_code.get(ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
    image_prompt_step = by_code.get(ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
    image_generate_step = by_code.get(ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
    video_prompt_step = by_code.get(ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
    video_generate_step = by_code.get(ShotReplicateStepCodeEnum.VIDEO_GENERATE.value)

    material_input = _step_payload(material_step.input_json if material_step else None)
    image_prompt_output = _unwrap_step_output(image_prompt_step.output_json if image_prompt_step else None)
    image_generate_input = _step_payload(image_generate_step.input_json if image_generate_step else None)
    image_generate_output = _unwrap_step_output(image_generate_step.output_json if image_generate_step else None)
    video_prompt_input = _step_payload(video_prompt_step.input_json if video_prompt_step else None)
    video_prompt_output = _unwrap_step_output(video_prompt_step.output_json if video_prompt_step else None)
    video_generate_input = _step_payload(video_generate_step.input_json if video_generate_step else None)
    video_generate_output = _unwrap_step_output(video_generate_step.output_json if video_generate_step else None)

    image_chat = chats.get(image_generate_step.chat_task_id) if image_generate_step and image_generate_step.chat_task_id else None
    video_chat = chats.get(video_generate_step.chat_task_id) if video_generate_step and video_generate_step.chat_task_id else None
    image_snapshot = _snapshot_from_chat(image_chat)
    video_snapshot = _snapshot_from_chat(video_chat)

    image_url = image_generate_output.get("result_image_url") or (image_chat.image_url if image_chat else None) or project.final_image_url
    video_url = video_generate_output.get("result_video_url") or (video_chat.video_url if video_chat else None) or project.final_video_url
    cover_url = video_generate_output.get("result_video_cover_url") or (video_chat.video_cover_url if video_chat else None) or project.final_video_cover_url

    schema_config_snapshot, schema_config_source, schema_config_version, schema_config_is_fallback = _resolve_video_schema_config_snapshot(video_prompt_output)

    user_name: str | None = None
    if project.user_id:
        user_result = await db.execute(select(User.username).where(User.id == project.user_id).limit(1))
        user_name = user_result.scalar_one_or_none()

    return ShotReplicateTaskDetailOut(
        id=project.id,
        project_id=project.id,
        user_id=project.user_id,
        user_name=user_name,
        module=project.module,
        title=project.title,
        status=project.status,
        current_step_code=project.current_step_code,
        final_image_url=build_resource_signed_url(project.final_image_url) if project.final_image_url else None,
        final_video_url=build_resource_signed_url(project.final_video_url) if project.final_video_url else None,
        final_video_cover_url=build_resource_signed_url(project.final_video_cover_url) if project.final_video_cover_url else None,
        error_message=project.error_message,
        material=ShotReplicateMaterialOut(
            material_step_id=material_step.id if material_step else None,
            material_video_url=material_input.get("material_video_url"),
            material_image_url=material_input.get("material_image_url"),
            source_project_name=material_input.get("source_project_name"),
            target_project_name=material_input.get("target_project_name"),
            core_content_point=material_input.get("core_content_point"),
        ),
        image_generation=ShotReplicateImageGenerationOut(
            prompt_step_id=image_prompt_step.id if image_prompt_step else None,
            generate_step_id=image_generate_step.id if image_generate_step else None,
            prompt=image_prompt_output.get("optimized_prompt") or image_prompt_output.get("prompt"),
            engine_id=image_snapshot.get("id") or image_generate_input.get("engine_id"),
            engine_name=image_snapshot.get("name") or image_generate_input.get("engine_name"),
            params=image_generate_input.get("params") or image_generate_input,
            chat_task_id=image_generate_step.chat_task_id if image_generate_step else None,
            status=image_chat.status if image_chat else (image_generate_step.status if image_generate_step else None),
            result_image_url=build_resource_signed_url(image_url) if image_url else None,
            error_message=image_chat.error_message if image_chat else (image_generate_step.error_message if image_generate_step else None),
        ),
        video_generation=ShotReplicateVideoGenerationOut(
            prompt_step_id=video_prompt_step.id if video_prompt_step else None,
            generate_step_id=video_generate_step.id if video_generate_step else None,
            prompt_schema=video_prompt_output.get("prompt_schema"),
            final_prompt=video_prompt_output.get("final_prompt"),
            prompt_params=video_prompt_output.get("params_used_for_prompt") or video_prompt_input.get("video_config"),
            schema_config_snapshot=schema_config_snapshot,
            schema_config_source=schema_config_source,
            schema_config_version=schema_config_version,
            schema_config_is_fallback=schema_config_is_fallback,
            engine_id=video_snapshot.get("id") or video_generate_input.get("engine_id"),
            engine_name=video_snapshot.get("name") or video_generate_input.get("engine_name"),
            params=video_generate_input.get("params") or video_generate_input,
            chat_task_id=video_generate_step.chat_task_id if video_generate_step else None,
            status=video_chat.status if video_chat else (video_generate_step.status if video_generate_step else None),
            result_video_url=build_resource_signed_url(video_url) if video_url else None,
            result_video_cover_url=build_resource_signed_url(cover_url) if cover_url else None,
            error_message=video_chat.error_message if video_chat else (video_generate_step.error_message if video_generate_step else None),
        ),
        steps=[_step_to_out(step) for step in steps],
        created_at=project.created_at,
        updated_at=project.updated_at,
        completed_at=project.completed_at,
    )


async def create_shot_replicate_project(db: AsyncSession, current_user: User, req: ShotReplicateTaskCreate) -> ModuleGenerationProject:
    if req.idempotency_key:
        result = await db.execute(
            select(ModuleGenerationProject)
            .where(
                ModuleGenerationProject.user_id == current_user.id,
                ModuleGenerationProject.module == MODULE,
                ModuleGenerationProject.idempotency_key == req.idempotency_key,
                ModuleGenerationProject.deleted_at.is_(None),
            )
            .order_by(ModuleGenerationProject.created_at.desc())
            .limit(1)
        )
        existing = result.scalar_one_or_none()
        if existing:
            return existing

    project = ModuleGenerationProject(
        id=generate_id(),
        user_id=current_user.id,
        module=MODULE,
        title=req.target_project_name,
        status=ModuleProjectStatusEnum.WAITING_USER.value,
        current_step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
        idempotency_key=req.idempotency_key,
    )
    db.add(project)
    await db.flush()

    await _create_step(
        db,
        project=project,
        step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
        status=ModuleStepStatusEnum.COMPLETED.value,
        input_data={
            "material_video_url": req.material_video_url,
            "material_image_url": req.material_image_url,
            "source_project_name": req.source_project_name,
            "target_project_name": req.target_project_name,
            "core_content_point": req.core_content_point,
        },
        output_data={"message": "素材输入已提交，后端不做素材文件校验。下一步请手动生成图片AI提词。"},
    )
    await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_CREATED.value, message="创建拆镜复刻项目")
    return project


async def _current_step_map_by_project_ids(
    db: AsyncSession,
    *,
    project_ids: set[str],
    step_code: str,
) -> dict[str, ModuleGenerationStep]:
    """一次性查询当前页项目的指定步骤，避免列表逐条查 material_input。"""
    if not project_ids:
        return {}
    result = await db.execute(
        select(ModuleGenerationStep).where(
            ModuleGenerationStep.project_id.in_(list(project_ids)),
            ModuleGenerationStep.step_code == step_code,
            ModuleGenerationStep.is_current.is_(True),
            ModuleGenerationStep.deleted_at.is_(None),
        )
    )
    return {step.project_id: step for step in result.scalars().all()}


async def list_shot_replicate_projects(
    db: AsyncSession,
    *,
    current_user: User,
    status: str | None,
    page: int,
    page_size: int,
) -> ShotReplicateTaskListOut:
    """
    拆镜复刻项目列表查询。

    这个列表目前不是后台主入口，但仍避免逐条查 material_input，保持和爆款开头列表一致的批量查询策略。
    """
    query = select(ModuleGenerationProject).where(
        ModuleGenerationProject.module == MODULE,
        ModuleGenerationProject.deleted_at.is_(None),
    )
    if not current_user.is_admin:
        query = query.where(ModuleGenerationProject.user_id == current_user.id)
    if status:
        query = query.where(ModuleGenerationProject.status == status)

    total = int((await db.execute(select(func.count()).select_from(query.subquery()))).scalar() or 0)
    result = await db.execute(
        query.order_by(ModuleGenerationProject.created_at.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    projects = list(result.scalars().unique().all())
    project_ids = {project.id for project in projects if project.id}
    material_step_map = await _current_step_map_by_project_ids(
        db,
        project_ids=project_ids,
        step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
    )

    items: list[ShotReplicateTaskListItemOut] = []
    for project in projects:
        material_step = material_step_map.get(project.id)
        material = _step_payload(material_step.input_json if material_step else None)
        items.append(
            ShotReplicateTaskListItemOut(
                id=project.id,
                project_id=project.id,
                module=project.module,
                title=project.title,
                status=project.status,
                current_step_code=project.current_step_code,
                target_project_name=material.get("target_project_name"),
                final_image_url=build_resource_signed_url(project.final_image_url) if project.final_image_url else None,
                final_video_url=build_resource_signed_url(project.final_video_url) if project.final_video_url else None,
                error_message=project.error_message,
                created_at=project.created_at,
                updated_at=project.updated_at,
                completed_at=project.completed_at,
            )
        )
    return ShotReplicateTaskListOut(total=total, items=items)


async def update_shot_replicate_step(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    step_id: str,
    req: ShotReplicateStepUpdate,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
    return await update_module_step(
        db,
        current_user=current_user,
        project_id=project_id,
        step_id=step_id,
        req=req,
        config=FLOW_CONFIG,
        log_module_event=log_module_event,
    )


async def update_shot_replicate_material_input(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    req: ShotReplicateMaterialUpdateRequest,
) -> tuple[str, str]:
    """修改第1步素材输入。

    采用方案 B：软删除旧第1步及之后的当前有效步骤，然后新建第1步 version+1。
    未传字段沿用旧第1步素材输入，避免前端只改一个字段时丢失其它素材信息。
    """
    return await update_module_material_input(
        db,
        current_user=current_user,
        project_id=project_id,
        req=req,
        config=FLOW_CONFIG,
        log_module_event=log_module_event,
    )


async def update_shot_replicate_image_prompt(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    step_id: str,
    req: ShotReplicateImagePromptUpdateRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
    """直接修改第2步图片 AI 优化提词，不调用 AI、不扣积分。

    修改后软删除第3、4、5步当前有效任务，让用户从图片生成开始重新执行。
    """
    return await update_module_image_prompt(
        db,
        current_user=current_user,
        project_id=project_id,
        step_id=step_id,
        req=req,
        config=FLOW_CONFIG,
        log_module_event=log_module_event,
    )


async def update_shot_replicate_video_prompt_schema(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    step_id: str,
    req: ShotReplicateVideoPromptSchemaUpdateRequest,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
    """以前端 schema 为 patch 修改第4步视频 AI 提词，不调用 AI、不扣积分。

    服务端已有 schema 为基准：视频规格、数组长度、时间段、合规控制、质量控制、协议字段均锁定。
    最终提示词允许修改，但保存前会清洗视频时长、比例、分辨率、帧率等参数。
    """
    return await update_module_video_prompt_schema(
        db,
        current_user=current_user,
        project_id=project_id,
        step_id=step_id,
        req=req,
        config=FLOW_CONFIG,
        log_module_event=log_module_event,
        patch_video_prompt_schema_from_client=patch_video_prompt_schema_from_client,
        build_final_video_prompt=build_final_video_prompt,
    )


async def submit_image_prompt_optimize(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    material_step_id: str | None = None,
    req: Any | None = None,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
    _ = req
    project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
    if material_step_id:
        material_step = await _get_step_for_user(db, project_id=project_id, step_id=material_step_id, user=current_user, for_update=True)
    else:
        material_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
        if not material_step:
            raise HTTPException(status_code=400, detail="缺少第1步素材输入子任务")
    if material_step.step_code != ShotReplicateStepCodeEnum.MATERIAL_INPUT.value:
        raise HTTPException(status_code=400, detail="请基于第1步素材输入子任务生成图片 AI 提词")
    if material_step.status != ModuleStepStatusEnum.COMPLETED.value:
        raise HTTPException(status_code=400, detail="素材输入子任务未完成，不能生成图片 AI 提词")

    await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value])
    step = await _create_step(
        db,
        project=project,
        step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
        status=ModuleStepStatusEnum.PROCESSING.value,
        parent_step_id=material_step.id,
        source_step_id=material_step.id,
        input_data={"source_step_id": material_step.id},
    )
    project.status = ModuleProjectStatusEnum.PROCESSING.value
    project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
    project.error_message = None
    await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_PROMPT_SUBMITTED.value, message="图片 AI 提词任务已提交")
    return project, step


async def run_image_prompt_optimize(db: AsyncSession, *, project_id: str, step_id: str | None = None) -> ModuleGenerationStep | None:
    project_result = await db.execute(
        select(ModuleGenerationProject)
        .where(ModuleGenerationProject.id == project_id, ModuleGenerationProject.module == MODULE, ModuleGenerationProject.deleted_at.is_(None))
        .with_for_update()
        .limit(1)
    )
    project = project_result.scalar_one_or_none()
    if not project:
        return None

    material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
    if not material_step:
        project.status = ModuleProjectStatusEnum.FAILED.value
        project.error_message = "缺少素材输入子任务"
        return None

    if step_id:
        result = await db.execute(
            select(ModuleGenerationStep)
            .where(
                ModuleGenerationStep.id == step_id,
                ModuleGenerationStep.project_id == project.id,
                ModuleGenerationStep.step_code == ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
                ModuleGenerationStep.deleted_at.is_(None),
                ModuleGenerationStep.is_current == True,
            )
            .with_for_update()
            .limit(1)
        )
        step = result.scalar_one_or_none()
    else:
        step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
    if not step:
        if step_id:
            # 用户重复提交后，旧 Celery 消息对应的 step 可能已被软删。
            # 指定 step_id 查不到时必须静默忽略，不能重新创建步骤导致旧任务复活。
            return None
        step = await _create_step(
            db,
            project=project,
            step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
            status=ModuleStepStatusEnum.PROCESSING.value,
            parent_step_id=material_step.id,
            source_step_id=material_step.id,
            input_data={"source_step_id": material_step.id},
        )
    else:
        step.status = ModuleStepStatusEnum.PROCESSING.value
        step.started_at = _now()
        step.error_message = None

    material = _step_payload(material_step.input_json)
    prompt_text = (
        "请基于参考素材复刻拆镜视觉风格，用于生成新项目图片。\n"
        f"视频素材内容项目名称：{material.get('source_project_name')}\n"
        f"生成项目名称：{material.get('target_project_name')}\n"
        f"生成项目核心内容点：{material.get('core_content_point')}\n"
        "要求：参考素材视频的开头构图、主体位置、节奏和风格；结合新产品图片生成新项目推广图片；不要照抄原素材品牌、文字、水印；适合作为后续图生视频首帧。"
    )
    references = [
        {"type": "video", "url": material.get("material_video_url"), "name": "参考素材视频"},
        {"type": "image", "url": material.get("material_image_url"), "name": "新产品图片"},
    ]
    try:
        request_log = {"original_prompt": prompt_text, "references": references, "gen_type": "image"}
        log_module_prompt_event(
            event_type="module_prompt_request",
            project_id=project.id,
            step_id=step.id,
            user_id=project.user_id,
            module=project.module,
            prompt_type=ModulePromptTypeEnum.IMAGE_PROMPT.value,
            request=request_log,
        )
        optimized, token_usage = await optimize_prompt(
            db,
            original_prompt=prompt_text,
            user_id=project.user_id,
            references=references,
            gen_type="image",
        )
        billing = await charge_module_prompt_usage(
            db,
            user_id=project.user_id,
            step_id=step.id,
            usage=token_usage,
            description="拆镜复刻-图片AI提词优化",
        )
        usage = dict(token_usage or {})
        usage.update({
            "text_credits_cost": (billing.items[0].amount if billing.items else billing.total_charged),
            "credit_biz_key": billing.items[0].biz_key if billing.items else None,
        })
        step.status = ModuleStepStatusEnum.COMPLETED.value
        step.completed_at = _now()
        _force_set_json(
            step,
            "output_json",
            _step_output(
                step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
                status=ModuleStepStatusEnum.COMPLETED.value,
                payload={
                    "optimized_prompt": optimized,
                    "prompt": optimized,
                    "original_prompt": prompt_text,
                    "references": references,
                },
                usage=usage,
            ),
        )
        project.status = ModuleProjectStatusEnum.WAITING_USER.value
        project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
        project.error_message = None
        log_module_prompt_event(
            event_type="module_prompt_response",
            project_id=project.id,
            step_id=step.id,
            user_id=project.user_id,
            module=project.module,
            prompt_type=ModulePromptTypeEnum.IMAGE_PROMPT.value,
            request=request_log,
            response={"optimized_prompt": optimized},
            token_usage=usage,
        )
        await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_PROMPT_SUCCESS.value, message="图片 AI 提词生成成功")
    except Exception as exc:
        step.status = ModuleStepStatusEnum.FAILED.value
        step.error_message = str(exc)
        step.completed_at = _now()
        project.status = ModuleProjectStatusEnum.FAILED.value
        project.error_message = f"图片 AI 提词生成失败: {exc}"
        log_module_prompt_event(
            event_type="module_prompt_error",
            project_id=project.id,
            step_id=step.id,
            user_id=project.user_id,
            module=project.module,
            prompt_type=ModulePromptTypeEnum.IMAGE_PROMPT.value,
            request=locals().get("request_log", {}),
            error=str(exc),
        )
        _log_project_error(project=project, step=step, event_type="IMAGE_PROMPT_FAILED", message=project.error_message, exc=exc)
        await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_PROMPT_FAILED.value, message=project.error_message)
    return step


async def generate_image_from_prompt(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    req: ShotReplicateGenerateImageRequest,
    prompt_step_id: str | None = None,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep, ChatGenerationTask]:
    project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
    if prompt_step_id:
        prompt_step = await _get_step_for_user(db, project_id=project_id, step_id=prompt_step_id, user=current_user, for_update=True)
    else:
        prompt_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
        if not prompt_step:
            raise HTTPException(status_code=400, detail="缺少第2步图片 AI 提词子任务")
    if prompt_step.step_code != ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
        raise HTTPException(status_code=400, detail="请基于第2步图片 AI 提词子任务生成图片")
    if prompt_step.status != ModuleStepStatusEnum.COMPLETED.value:
        raise HTTPException(status_code=400, detail="图片 AI 提词未完成，不能生成图片")

    await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.IMAGE_GENERATE.value])

    material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
    material = _step_payload(material_step.input_json if material_step else None)
    prompt_output = _unwrap_step_output(prompt_step.output_json)
    optimized_prompt = prompt_output.get("optimized_prompt") or prompt_output.get("prompt") or ""
    refs = [
        {"type": "image", "url": material.get("material_image_url"), "name": "新产品图片"},
    ]

    step = await _create_step(
        db,
        project=project,
        step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
        status=ModuleStepStatusEnum.PROCESSING.value,
        parent_step_id=prompt_step.id,
        source_step_id=prompt_step.id,
        input_data={
            "engine_id": req.engine_id,
            "params": {
                "image_size": req.image_size,
                "image_proportion": req.image_proportion,
                "image_px": req.image_px,
            },
            "prompt": optimized_prompt,
            "media_references": refs,
        },
    )
    chat_task = await create_chat_generation_task_for_module(
        db,
        current_user=current_user,
        generation_mode=GENERATION_MODE,
        gen_type="image",
        original_prompt=prompt_output.get("original_prompt") or optimized_prompt,
        optimized_prompt=optimized_prompt,
        engine_id=req.engine_id,
        media_references=refs,
        image_size=req.image_size,
        image_proportion=req.image_proportion,
        image_px=req.image_px,
        billing_project_name=project.title or "拆镜复刻",
        billing_description_prefix="拆镜复刻图片生成",
        billing_source_module=project.module,
        billing_source_project_id=project.id,
        billing_source_step_id=step.id,
        billing_source_step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
    )
    step.chat_task_id = chat_task.id
    _force_set_json(
        step,
        "input_json",
        _step_input(
            step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
            source_step_id=prompt_step.id,
            parent_step_id=prompt_step.id,
            payload={
                "engine_id": chat_task.engine_id,
                "params": {
                    "image_size": chat_task.image_size,
                    "image_proportion": chat_task.image_proportion,
                    "image_px": chat_task.image_px,
                },
                "prompt": optimized_prompt,
                "media_references": refs,
            },
        ),
    )
    project.status = ModuleProjectStatusEnum.PROCESSING.value
    project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_GENERATE.value
    project.error_message = None
    await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_GENERATE_SUBMITTED.value, message="图片生成任务已提交", detail={"chat_task_id": chat_task.id})
    return project, step, chat_task


async def _resolve_video_prompt_config(db: AsyncSession, req: ShotReplicateGenerateVideoPromptRequest) -> dict[str, Any]:
    engine = await _get_video_engine(db, req.engine_id)
    supported_ratios = _parse_list(engine.supported_ratios, [])
    supported_resolutions = _parse_list(engine.supported_resolutions, [])
    supported_durations = _parse_list(engine.supported_durations, [])

    default_ratio = getattr(settings, "SHOT_REPLICATE_DEFAULT_VIDEO_RATIO", None) or VIDEO_DEFAULT_RATIO
    default_resolution = getattr(settings, "SHOT_REPLICATE_DEFAULT_VIDEO_RESOLUTION", None) or VIDEO_DEFAULT_RESOLUTION
    default_duration = int(getattr(settings, "SHOT_REPLICATE_DEFAULT_VIDEO_DURATION", None) or VIDEO_DEFAULT_DURATION)

    selected_ratio = req.aspect_ratio or (default_ratio if not supported_ratios or default_ratio in supported_ratios else supported_ratios[0])
    selected_resolution = req.resolution or (default_resolution if not supported_resolutions or default_resolution in supported_resolutions else supported_resolutions[0])
    selected_duration = req.duration or (default_duration if not supported_durations or default_duration in supported_durations else supported_durations[0])

    if supported_ratios and selected_ratio not in supported_ratios:
        raise HTTPException(status_code=400, detail=f"视频比例不支持: {selected_ratio}")
    if supported_resolutions and selected_resolution not in supported_resolutions:
        raise HTTPException(status_code=400, detail=f"视频分辨率不支持: {selected_resolution}")
    if supported_durations and selected_duration not in supported_durations:
        raise HTTPException(status_code=400, detail=f"视频时长不支持: {selected_duration}")
    if engine.max_duration and int(selected_duration) > int(engine.max_duration):
        raise HTTPException(status_code=400, detail=f"视频时长不能超过 {engine.max_duration} 秒")

    return {
        "engine_id": engine.id,
        "engine_name": engine.name,
        "duration": int(selected_duration),
        "aspect_ratio": selected_ratio,
        "resolution": selected_resolution,
        "supported_ratios": supported_ratios,
        "supported_resolutions": supported_resolutions,
        "supported_durations": supported_durations,
        "max_duration": engine.max_duration,
        "frame_rate": "30fps",
        "reference_video_fps": max(1, int(settings.CHATAPI_VIDEO_FPS or 1)),
    }


async def submit_video_prompt_optimize(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    req: ShotReplicateGenerateVideoPromptRequest,
    image_step_id: str | None = None,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep]:
    project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
    if image_step_id:
        image_step = await _get_step_for_user(db, project_id=project_id, step_id=image_step_id, user=current_user, for_update=True)
    else:
        image_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
        if not image_step:
            raise HTTPException(status_code=400, detail="缺少第3步图片生成子任务")
    if image_step.step_code != ShotReplicateStepCodeEnum.IMAGE_GENERATE.value:
        raise HTTPException(status_code=400, detail="请基于第3步图片生成子任务生成视频 AI 提词")
    if image_step.status != ModuleStepStatusEnum.COMPLETED.value:
        raise HTTPException(status_code=400, detail="图片生成子任务未完成，不能生成视频 AI 提词")

    await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value])
    video_config = await _resolve_video_prompt_config(db, req)
    step = await _create_step(
        db,
        project=project,
        step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
        status=ModuleStepStatusEnum.PROCESSING.value,
        parent_step_id=image_step.id,
        source_step_id=image_step.id,
        input_data={
            "source_step_id": image_step.id,
            "video_config": video_config,
            "target_platform": req.target_platform or getattr(settings, "SHOT_REPLICATE_DEFAULT_TARGET_PLATFORM", "抖音") or "抖音",
        },
    )
    project.status = ModuleProjectStatusEnum.PROCESSING.value
    project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value
    project.error_message = None
    await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_SUBMITTED.value, message="视频 AI 提词任务已提交")
    return project, step


async def run_video_prompt_optimize(db: AsyncSession, *, project_id: str, step_id: str | None = None) -> ModuleGenerationStep | None:
    project_result = await db.execute(
        select(ModuleGenerationProject)
        .where(ModuleGenerationProject.id == project_id, ModuleGenerationProject.module == MODULE, ModuleGenerationProject.deleted_at.is_(None))
        .with_for_update()
        .limit(1)
    )
    project = project_result.scalar_one_or_none()
    if not project:
        return None

    material_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.MATERIAL_INPUT.value)
    image_prompt_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value)
    image_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
    if not material_step or not image_step:
        project.status = ModuleProjectStatusEnum.FAILED.value
        project.error_message = "生成视频提词失败：缺少素材输入或图片生成结果"
        return None

    if step_id:
        result = await db.execute(
            select(ModuleGenerationStep)
            .where(
                ModuleGenerationStep.id == step_id,
                ModuleGenerationStep.project_id == project.id,
                ModuleGenerationStep.step_code == ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
                ModuleGenerationStep.deleted_at.is_(None),
                ModuleGenerationStep.is_current == True,
            )
            .with_for_update()
            .limit(1)
        )
        step = result.scalar_one_or_none()
    else:
        step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
    if not step:
        if step_id:
            # 用户重复提交后，旧 Celery 消息对应的 step 可能已被软删。
            # 指定 step_id 查不到时必须静默忽略，不能把当前项目标记失败。
            return None
        project.status = ModuleProjectStatusEnum.FAILED.value
        project.error_message = "缺少视频 AI 提词子任务，请先手动提交视频提词生成"
        return None

    step.status = ModuleStepStatusEnum.PROCESSING.value
    step.started_at = _now()
    step.error_message = None

    material = _step_payload(material_step.input_json)
    image_output = _unwrap_step_output(image_step.output_json)
    step_input = _step_payload(step.input_json)
    video_config = step_input.get("video_config") or {}
    target_platform = step_input.get("target_platform") or getattr(settings, "SHOT_REPLICATE_DEFAULT_TARGET_PLATFORM", "抖音") or "抖音"
    generated_image_url = image_output.get("result_image_url") or project.final_image_url
    if not generated_image_url:
        step.status = ModuleStepStatusEnum.FAILED.value
        step.error_message = "缺少新项目图片结果，不能生成视频提词"
        project.status = ModuleProjectStatusEnum.FAILED.value
        project.error_message = step.error_message
        return step

    try:
        request_log = {
            "source_project_name": material.get("source_project_name") or "无",
            "target_project_name": material.get("target_project_name") or "无",
            "core_content_point": material.get("core_content_point") or "无",
            "material_video_url": material.get("material_video_url") or "",
            "generated_image_url": generated_image_url,
            "video_config": video_config,
            "target_platform": target_platform,
        }
        log_module_prompt_event(
            event_type="module_prompt_request",
            project_id=project.id,
            step_id=step.id,
            user_id=project.user_id,
            module=project.module,
            prompt_type=ModulePromptTypeEnum.VIDEO_PROMPT.value,
            request=request_log,
        )
        schema_config_snapshot = await get_runtime_schema_snapshot(db)
        request_log["schema_config_source"] = schema_config_snapshot.get("source")
        prompt_schema, final_prompt, token_usage = await optimize_shot_replicate_video_prompt(
            db,
            user_id=project.user_id,
            source_project_name=request_log["source_project_name"],
            target_project_name=request_log["target_project_name"],
            core_content_point=request_log["core_content_point"],
            material_video_url=request_log["material_video_url"],
            generated_image_url=generated_image_url,
            video_config=video_config,
            target_platform=target_platform,
            schema_config_snapshot=schema_config_snapshot,
        )
        billing = await charge_module_prompt_usage(
            db,
            user_id=project.user_id,
            step_id=step.id,
            usage=token_usage,
            description="拆镜复刻-视频AI提词优化",
        )
        usage = dict(token_usage or {})
        usage.update({
            "text_credits_cost": (billing.items[0].amount if billing.items else billing.total_charged),
            "credit_biz_key": billing.items[0].biz_key if billing.items else None,
        })
        step.status = ModuleStepStatusEnum.COMPLETED.value
        step.completed_at = _now()
        _force_set_json(
            step,
            "output_json",
            _step_output(
                step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
                status=ModuleStepStatusEnum.COMPLETED.value,
                payload={
                    "prompt_schema": prompt_schema,
                    "final_prompt": final_prompt,
                    "params_used_for_prompt": video_config,
                    "target_platform": target_platform,
                    "schema_config_snapshot": schema_config_snapshot,
                    "schema_config_source": schema_config_snapshot.get("source"),
                    "schema_config_version": schema_config_snapshot.get("version"),
                },
                usage=usage,
            ),
        )
        project.status = ModuleProjectStatusEnum.WAITING_USER.value
        project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value
        project.error_message = None
        log_module_prompt_event(
            event_type="module_prompt_response",
            project_id=project.id,
            step_id=step.id,
            user_id=project.user_id,
            module=project.module,
            prompt_type=ModulePromptTypeEnum.VIDEO_PROMPT.value,
            request=request_log,
            response={"prompt_schema": prompt_schema, "final_prompt": final_prompt, "schema_config_source": schema_config_snapshot.get("source")},
            token_usage=usage,
        )
        await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_SUCCESS.value, message="视频 AI 提词生成成功")
    except Exception as exc:
        step.status = ModuleStepStatusEnum.FAILED.value
        step.error_message = str(exc)
        step.completed_at = _now()
        project.status = ModuleProjectStatusEnum.FAILED.value
        project.error_message = f"视频 AI 提词生成失败: {exc}"
        log_module_prompt_event(
            event_type="module_prompt_error",
            project_id=project.id,
            step_id=step.id,
            user_id=project.user_id,
            module=project.module,
            prompt_type=ModulePromptTypeEnum.VIDEO_PROMPT.value,
            request=locals().get("request_log", {}),
            error=str(exc),
        )
        _log_project_error(project=project, step=step, event_type="VIDEO_PROMPT_FAILED", message=project.error_message, exc=exc)
        await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_PROMPT_FAILED.value, message=project.error_message)
    return step


async def generate_video_from_prompt(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    req: ShotReplicateGenerateVideoRequest,
    prompt_step_id: str | None = None,
) -> tuple[ModuleGenerationProject, ModuleGenerationStep, ChatGenerationTask]:
    project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
    if prompt_step_id:
        prompt_step = await _get_step_for_user(db, project_id=project_id, step_id=prompt_step_id, user=current_user, for_update=True)
    else:
        prompt_step = await _get_current_step_by_code(db, project_id, ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value)
        if not prompt_step:
            raise HTTPException(status_code=400, detail="缺少第4步视频 AI 提词子任务")
    if prompt_step.step_code != ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
        raise HTTPException(status_code=400, detail="请基于第4步视频 AI 提词子任务生成视频")
    if prompt_step.status != ModuleStepStatusEnum.COMPLETED.value:
        raise HTTPException(status_code=400, detail="视频 AI 提词未完成，不能生成视频")

    await _soft_delete_steps_from_index(db, project=project, start_index=STEP_INDEX_MAP[ShotReplicateStepCodeEnum.VIDEO_GENERATE.value])

    image_step = await _get_current_step_by_code(db, project.id, ShotReplicateStepCodeEnum.IMAGE_GENERATE.value)
    image_output = _unwrap_step_output(image_step.output_json if image_step else None)
    prompt_output = _unwrap_step_output(prompt_step.output_json)
    final_prompt = prompt_output.get("final_prompt") or ""
    prompt_schema = prompt_output.get("prompt_schema") or {}
    prompt_schema_str = json.dumps(prompt_schema, ensure_ascii=False, default=str) if prompt_schema else ""
    prompt_input = _step_payload(prompt_step.input_json)
    prompt_params = prompt_output.get("params_used_for_prompt") or prompt_input.get("video_config") or {}
    duration = int(prompt_params.get("duration") or settings.SHOT_REPLICATE_DEFAULT_VIDEO_DURATION or 4)
    aspect_ratio = prompt_params.get("aspect_ratio") or settings.SHOT_REPLICATE_DEFAULT_VIDEO_RATIO or "9:16"
    resolution = prompt_params.get("resolution") or settings.SHOT_REPLICATE_DEFAULT_VIDEO_RESOLUTION or "480p"
    generated_image_url = image_output.get("result_image_url") or project.final_image_url
    if not generated_image_url:
        raise HTTPException(status_code=400, detail="缺少新项目图片结果，不能生成视频")

    refs = [
        {"type": "image", "url": _build_file_url_or_data_uri(generated_image_url), "name": "新项目图片"},
    ]

    step = await _create_step(
        db,
        project=project,
        step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
        status=ModuleStepStatusEnum.PROCESSING.value,
        parent_step_id=prompt_step.id,
        source_step_id=prompt_step.id,
        input_data={
            "engine_id": req.engine_id or prompt_params.get("engine_id"),
            "params": {
                "duration": duration,
                "aspect_ratio": aspect_ratio,
                "resolution": resolution,
            },
            "prompt_schema": prompt_schema,
            "final_prompt": final_prompt,
            "media_references": refs,
        },
    )
    chat_task = await create_chat_generation_task_for_module(
        db,
        current_user=current_user,
        generation_mode=GENERATION_MODE,
        gen_type="video",
        original_prompt=prompt_schema_str or final_prompt,
        optimized_prompt=prompt_schema_str or final_prompt,
        engine_id=req.engine_id or prompt_params.get("engine_id"),
        media_references=refs,
        duration=duration,
        aspect_ratio=aspect_ratio,
        resolution=resolution,
        billing_project_name=project.title or "拆镜复刻",
        billing_description_prefix="拆镜复刻视频生成",
        billing_source_module=project.module,
        billing_source_project_id=project.id,
        billing_source_step_id=step.id,
        billing_source_step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
    )
    step.chat_task_id = chat_task.id
    _force_set_json(
        step,
        "input_json",
        _step_input(
            step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
            source_step_id=prompt_step.id,
            parent_step_id=prompt_step.id,
            payload={
                "engine_id": chat_task.engine_id,
                "params": {
                    "duration": chat_task.duration,
                    "aspect_ratio": chat_task.aspect_ratio,
                    "resolution": chat_task.resolution,
                    "image_size": chat_task.image_size,
                    "image_proportion": chat_task.image_proportion,
                    "image_px": chat_task.image_px,
                },
                "prompt_schema": prompt_schema,
                "final_prompt": final_prompt,
                "media_references": refs,
            },
        ),
    )
    project.status = ModuleProjectStatusEnum.PROCESSING.value
    project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_GENERATE.value
    project.error_message = None
    await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_GENERATE_SUBMITTED.value, message="视频生成任务已提交", detail={"chat_task_id": chat_task.id})
    return project, step, chat_task


async def handle_chat_generation_task_completed(db: AsyncSession, task: ChatGenerationTask) -> None:
    if not task or task.generation_mode != GENERATION_MODE:
        return
    result = await db.execute(
        select(ModuleGenerationStep)
        .where(
            ModuleGenerationStep.chat_task_id == task.id,
            ModuleGenerationStep.module == MODULE,
            ModuleGenerationStep.is_current == True,
            ModuleGenerationStep.deleted_at.is_(None),
        )
        .with_for_update()
        .limit(1)
    )
    step = result.scalar_one_or_none()
    if not step:
        return
    project_result = await db.execute(
        select(ModuleGenerationProject)
        .where(ModuleGenerationProject.id == step.project_id, ModuleGenerationProject.deleted_at.is_(None))
        .with_for_update()
        .limit(1)
    )
    project = project_result.scalar_one_or_none()
    if not project:
        return

    if step.step_code == ShotReplicateStepCodeEnum.IMAGE_GENERATE.value:
        step.status = ModuleStepStatusEnum.COMPLETED.value
        step.completed_at = _now()
        _force_set_json(
            step,
            "output_json",
            _step_output(
                step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
                status=ModuleStepStatusEnum.COMPLETED.value,
                result={"result_image_url": task.image_url, "chat_task_id": task.id},
            ),
        )
        project.final_image_url = task.image_url
        project.status = ModuleProjectStatusEnum.WAITING_USER.value
        project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_GENERATE.value
        await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.IMAGE_GENERATE_SUCCESS.value, message="图片生成完成，等待用户手动生成视频 AI 提词")
    elif step.step_code == ShotReplicateStepCodeEnum.VIDEO_GENERATE.value:
        step.status = ModuleStepStatusEnum.COMPLETED.value
        step.completed_at = _now()
        _force_set_json(
            step,
            "output_json",
            _step_output(
                step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
                status=ModuleStepStatusEnum.COMPLETED.value,
                result={"result_video_url": task.video_url, "result_video_cover_url": task.video_cover_url, "chat_task_id": task.id},
            ),
        )
        project.final_video_url = task.video_url
        project.final_video_cover_url = task.video_cover_url
        project.status = ModuleProjectStatusEnum.COMPLETED.value
        project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_GENERATE.value
        project.completed_at = _now()
        await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.VIDEO_GENERATE_SUCCESS.value, message="视频生成完成，总任务完成")


async def handle_chat_generation_task_failed(db: AsyncSession, task: ChatGenerationTask) -> None:
    if not task or task.generation_mode != GENERATION_MODE:
        return
    result = await db.execute(
        select(ModuleGenerationStep)
        .where(ModuleGenerationStep.chat_task_id == task.id, ModuleGenerationStep.module == MODULE, ModuleGenerationStep.is_current == True, ModuleGenerationStep.deleted_at.is_(None))
        .with_for_update()
        .limit(1)
    )
    step = result.scalar_one_or_none()
    if not step:
        return
    project_result = await db.execute(select(ModuleGenerationProject).where(ModuleGenerationProject.id == step.project_id).with_for_update().limit(1))
    project = project_result.scalar_one_or_none()
    if not project:
        return
    step.status = ModuleStepStatusEnum.FAILED.value
    step.error_message = task.error_message
    step.completed_at = _now()
    project.status = ModuleProjectStatusEnum.FAILED.value
    project.error_message = task.error_message or "生成失败"
    await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.CHAT_TASK_FAILED.value, message=project.error_message, detail={"chat_task_id": task.id})


async def mark_shot_replicate_step_dispatch_failed(
    db: AsyncSession,
    *,
    current_user: User,
    project_id: str,
    step_id: str,
    error_message: str,
) -> None:
    project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
    result = await db.execute(
        select(ModuleGenerationStep)
        .where(
            ModuleGenerationStep.id == step_id,
            ModuleGenerationStep.project_id == project.id,
            ModuleGenerationStep.module == MODULE,
            ModuleGenerationStep.deleted_at.is_(None),
            ModuleGenerationStep.is_current == True,
        )
        .with_for_update()
        .limit(1)
    )
    step = result.scalar_one_or_none()
    if not step:
        return
    if step.chat_task_id:
        await mark_chat_generation_task_failed_and_refund_once(
            db,
            task_id=step.chat_task_id,
            error_message=error_message,
            pipeline_stage="failed",
        )
    step.status = ModuleStepStatusEnum.FAILED.value
    step.error_message = error_message
    step.completed_at = _now()
    project.status = ModuleProjectStatusEnum.FAILED.value
    project.error_message = error_message
    log_module_error(
        module=project.module,
        event_type="CELERY_DISPATCH_FAILED",
        project_id=project.id,
        step_id=step.id,
        user_id=project.user_id,
        message=error_message,
        detail={"reason": "celery_dispatch_failed", "chat_task_id": step.chat_task_id},
        error=error_message,
    )
    await log_module_event(
        db,
        project=project,
        step=step,
        event_type=ModuleEventTypeEnum.CHAT_TASK_FAILED.value,
        message=error_message,
        detail={"reason": "celery_dispatch_failed"},
    )


async def delete_shot_replicate_project(db: AsyncSession, *, current_user: User, project_id: str) -> ShotReplicateDeleteOut:
    project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
    deleted_at = _now()
    project.deleted_at = deleted_at
    await _soft_delete_steps_from_index(db, project=project, start_index=1, deleted_at=deleted_at)
    await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_DELETED.value, message="软删除拆镜复刻项目")
    return ShotReplicateDeleteOut(message="项目已删除", project_id=project.id, deleted=True)


async def create_shot_replicate_project_from_segment(
    db: AsyncSession,
    *,
    current_user: User,
    segment: ShotReplicateSegment,
    req: ShotSegmentReplicationCreateRequest,
) -> ModuleGenerationProject:
    """从拆镜片段创建拆镜复刻项目。

    素材视频固定取 segment.segment_video_url，不允许前端传入或后续修改。
    """
    if segment.user_id != current_user.id and not current_user.is_admin:
        raise HTTPException(status_code=404, detail="拆镜片段不存在")
    if segment.split_status != ShotSplitStatusEnum.COMPLETED.value or not segment.segment_video_url:
        raise HTTPException(status_code=400, detail="拆镜片段视频未完成，不能进入复刻流程")
    if segment.module_project_id:
        existing = await _get_project_for_user(
            db,
            project_id=segment.module_project_id,
            user=current_user,
            for_update=False,
        )
        return existing

    if req.idempotency_key:
        existing_result = await db.execute(
            select(ModuleGenerationProject).where(
                ModuleGenerationProject.user_id == current_user.id,
                ModuleGenerationProject.module == MODULE,
                ModuleGenerationProject.idempotency_key == req.idempotency_key,
                ModuleGenerationProject.deleted_at.is_(None),
            ).limit(1)
        )
        existing = existing_result.scalar_one_or_none()
        if existing:
            segment.module_project_id = existing.id
            segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
            await db.flush()
            return existing

    project = ModuleGenerationProject(
        id=generate_id(),
        user_id=current_user.id,
        module=MODULE,
        title=req.target_project_name,
        status=ModuleProjectStatusEnum.WAITING_USER.value,
        current_step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
        idempotency_key=req.idempotency_key,
    )
    db.add(project)
    await db.flush()

    payload = {
        "material_video_url": segment.segment_video_url,
        "material_video_locked": True,
        "material_image_url": req.material_image_url,
        "source_project_name": segment.segment_category or segment.original_video_category or "拆镜片段",
        "target_project_name": req.target_project_name,
        "core_content_point": req.core_content_point,
        "source_shot_task_set_id": segment.task_set_id,
        "source_segment_id": segment.id,
        "source_segment_index": segment.segment_index,
        "segment_time_node": segment.time_node,
    }
    context = {
        "shot_segment": {
            "id": segment.id,
            "task_set_id": segment.task_set_id,
            "segment_index": segment.segment_index,
            "time_node": segment.time_node,
            "start_second": segment.start_second,
            "end_second": segment.end_second,
            "source_mode": segment.source_mode,
        },
        "analysis": {
            "original_video_content": segment.original_video_content,
            "original_video_category": segment.original_video_category,
            "original_video_audience": segment.original_video_audience,
            "segment_content": segment.segment_content,
            "segment_category": segment.segment_category,
            "segment_audience": segment.segment_audience,
        },
    }

    step = await _create_step(
        db,
        project=project,
        step_code=ShotReplicateStepCodeEnum.MATERIAL_INPUT.value,
        status=ModuleStepStatusEnum.COMPLETED.value,
        input_data={**payload, "source_context": context},
        output_data={
            "accepted": True,
            "message": "拆镜片段素材输入已提交，素材视频已锁定",
            "next_step_code": ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
            **context["analysis"],
        },
    )
    project.current_step_code = ShotReplicateStepCodeEnum.MATERIAL_INPUT.value
    project.status = ModuleProjectStatusEnum.WAITING_USER.value
    segment.module_project_id = project.id
    segment.replicate_status = ShotSegmentReplicateStatusEnum.PROJECT_CREATED.value
    await log_module_event(
        db,
        project=project,
        step=step,
        event_type=ModuleEventTypeEnum.PROJECT_CREATED.value,
        message="从拆镜片段创建拆镜复刻项目",
        detail={"segment_id": segment.id, "task_set_id": segment.task_set_id},
    )
    await db.flush()
    return project


def _build_file_url_or_data_uri(file_url: str) -> str:
    return _common_build_file_url_or_data_uri(file_url)