积分冻结释放

2026-07-24 09:18:05 +08:00
parent 920d884e92
commit 68e902b4a4
38 changed files with 4743 additions and 391 deletions
@@ -2,6 +2,7 @@ from __future__ import annotations

 from datetime import datetime
 from types import SimpleNamespace
+from typing import Any

 from fastapi import APIRouter, Body, Depends, File, HTTPException, Path, Query, UploadFile
 from sqlalchemy import inspect as sa_inspect
@@ -13,6 +14,12 @@ from app.dependencies import get_current_user, get_db
 from app.models.user import User
 from app.enums.common import ModuleEventTypeEnum
 from app.enums.generation_task import GenerationOwnerType
+from app.enums.credit_record import (
+    CreditRecordBillingScene,
+    CreditRecordChargeKind,
+    CreditRecordOwnerType,
+)
+from app.enums.llm_billing import LlmBillingConfigKey
 from app.enums.shot_replicate import (
    ModuleCodeEnum,
    ShotAnalysisStatusEnum,
@@ -65,6 +72,7 @@ from app.services.shot_replicate_flow_service import (
    update_shot_replicate_video_prompt_schema,
 )
 from app.services.shot_replicate_taskset_service import (
+    build_task_set_analysis_billing_context,
    create_custom_segment,
    create_segments_by_ai,
    create_task_set,
@@ -72,6 +80,9 @@ from app.services.shot_replicate_taskset_service import (
    delete_task_set,
    list_segments,
    list_task_sets,
+    mark_custom_segment_split_dispatch_failed,
+    mark_segment_analysis_dispatch_failed,
+    mark_task_set_analysis_dispatch_failed,
    prepare_reanalyze_segment,
    prepare_reanalyze_task_set,
    prepare_retry_split_segment,
@@ -79,10 +90,20 @@ from app.services.shot_replicate_taskset_service import (
    task_set_detail,
 )
 from app.services.module_generation_log_service import log_module_error, log_module_event_file
+from app.services.llm_billing import (
+    LlmBillingContext,
+    log_celery_dispatch_compensated,
+    log_celery_dispatch_failure,
+    log_celery_dispatch_start,
+    log_celery_dispatch_success,
+)
 from app.services.module_async_recovery_service import (
+    OBJECT_MODULE_STEP,
    TASK_SHOT_IMAGE_PROMPT,
    TASK_SHOT_VIDEO_PROMPT,
+    has_live_object_lock,
    register_module_step_task,
+    remove_active_task,
 )
 from app.tasks.celery_app import celery_app
 from app.enums.upload_resource import UploadResourceEventEnum, UploadResourceModuleEnum, UploadResourceSourceModelEnum, UploadResourceTypeEnum
@@ -187,6 +208,83 @@ def _ensure_celery_enabled(*, current_user: User | None = None, project_id: str
    )
    raise HTTPException(status_code=503, detail=message)

+def _prompt_dispatch_billing_context(
+    *,
+    user_id: str,
+    project_id: str,
+    step_id: str,
+    step_code: str,
+    attempt_no: int,
+    celery_task_id: str,
+) -> LlmBillingContext:
+    is_image = step_code == ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value
+    return LlmBillingContext(
+        user_id=user_id,
+        owner_type=CreditRecordOwnerType.MODULE_GENERATION_STEP.value,
+        owner_id=step_id,
+        attempt_no=attempt_no,
+        charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value,
+        billing_scene=(
+            CreditRecordBillingScene.SHOT_IMAGE_PROMPT_OPTIMIZE.value
+            if is_image
+            else CreditRecordBillingScene.SHOT_VIDEO_PROMPT_OPTIMIZE.value
+        ),
+        source_module=MODULE,
+        source_project_id=project_id,
+        source_step_id=step_id,
+        source_step_code=step_code,
+        related_id=step_id,
+        hold_config_key=(
+            LlmBillingConfigKey.HOLD_MODULE_IMAGE_PROMPT.value
+            if is_image
+            else LlmBillingConfigKey.HOLD_MODULE_VIDEO_PROMPT.value
+        ),
+        description_prefix=(
+            "拆镜复刻图片AI提词优化" if is_image else "拆镜复刻视频提词优化"
+        ),
+        trace_id=f"shot-replicate-prompt:{step_id}:attempt:{attempt_no}",
+        celery_task_id=celery_task_id,
+    )
+
+
+def _analysis_dispatch_billing_context(
+    *,
+    user_id: str,
+    owner_id: str,
+    attempt_no: int,
+    task_set_id: str,
+    is_segment: bool,
+    celery_task_id: str,
+) -> LlmBillingContext:
+    return LlmBillingContext(
+        user_id=user_id,
+        owner_type=(
+            CreditRecordOwnerType.SHOT_REPLICATE_SEGMENT.value
+            if is_segment
+            else CreditRecordOwnerType.SHOT_REPLICATE_TASK_SET.value
+        ),
+        owner_id=owner_id,
+        attempt_no=attempt_no,
+        charge_kind=CreditRecordChargeKind.VIDEO_ANALYSIS.value,
+        billing_scene=(
+            CreditRecordBillingScene.SHOT_SEGMENT_VIDEO_ANALYSIS.value
+            if is_segment
+            else CreditRecordBillingScene.SHOT_ORIGINAL_VIDEO_ANALYSIS.value
+        ),
+        source_module=MODULE,
+        source_project_id=task_set_id,
+        source_step_id=owner_id,
+        source_step_code=ShotReplicateStepCodeEnum.VIDEO_ANALYSIS.value,
+        related_id=owner_id,
+        hold_config_key=LlmBillingConfigKey.HOLD_SHOT_VIDEO_ANALYSIS.value,
+        description_prefix=(
+            "拆镜复刻片段视频AI分析" if is_segment else "拆镜复刻原视频AI分析"
+        ),
+        trace_id=f"shot-analysis:{owner_id}:attempt:{attempt_no}",
+        celery_task_id=celery_task_id,
+    )
+
+
 async def _reload_project_detail(db: AsyncSession, current_user: User, project_id: str) -> ShotReplicateTaskDetailOut:
    project = await _get_project_for_user(
        db,
@@ -205,9 +303,25 @@ async def _mark_dispatch_failed_and_raise(
    project_id: str,
    step_id: str | None,
    message: str,
+    billing_context: LlmBillingContext | None = None,
 ) -> None:
+    if billing_context is not None:
+        log_celery_dispatch_failure(billing_context, error=message)
+    compensated = False
    if step_id:
        try:
+            if await has_live_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id):
+                log_module_error(
+                    module=MODULE,
+                    event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
+                    project_id=project_id,
+                    step_id=step_id,
+                    user_id=_safe_user_id(current_user),
+                    message="Celery 投递返回异常，但 worker 已领取任务，跳过失败补偿",
+                    detail={"reason": "uncertain_dispatch_worker_started", "dispatch_error": message},
+                    error=message,
+                )
+                raise HTTPException(status_code=503, detail=f"{message}；任务可能已被 worker 接收，请勿重复提交")
            await mark_shot_replicate_step_dispatch_failed(
                db,
                current_user=_user_context(current_user),
@@ -216,6 +330,23 @@ async def _mark_dispatch_failed_and_raise(
                error_message=message,
            )
            await db.commit()
+            compensated = True
+            if billing_context is not None:
+                log_celery_dispatch_compensated(billing_context, error=message)
+            try:
+                await remove_active_task(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+            except Exception as cleanup_exc:
+                _log_api_error(
+                    event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value,
+                    current_user=current_user,
+                    project_id=project_id,
+                    step_id=step_id,
+                    message="Celery 投递补偿完成，但清理 active registry 失败",
+                    detail={"dispatch_error": message},
+                    exc=cleanup_exc,
+                )
+        except HTTPException:
+            raise
        except Exception as exc:
            await db.rollback()
            _log_api_error(
@@ -234,12 +365,89 @@ async def _mark_dispatch_failed_and_raise(
        step_id=step_id,
        user_id=_safe_user_id(current_user),
        message=message,
-        detail={"reason": "celery_dispatch_failed"},
+        detail={"reason": "celery_dispatch_failed", "compensated": compensated},
        error=message,
    )
    raise HTTPException(status_code=503, detail=message)


+async def _dispatch_prompt_task(
+    db: AsyncSession,
+    *,
+    current_user: User,
+    project_id: str,
+    step_id: str,
+    step_code: str,
+    task_name: str,
+    celery_task: Any,
+    celery_task_id: str,
+    billing_context: LlmBillingContext,
+    error_prefix: str,
+) -> None:
+    """Redis 注册与 Celery 直投任一成功即视为可恢复投递。"""
+    registry_error: Exception | None = None
+    try:
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id,
+            step_id=step_id,
+            step_code=step_code,
+            task_name=task_name,
+        )
+    except Exception as exc:
+        registry_error = exc
+        log_module_error(
+            module=MODULE,
+            event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
+            project_id=project_id,
+            step_id=step_id,
+            user_id=_safe_user_id(current_user),
+            message="提词任务 Redis 活跃注册失败，将继续尝试 Celery 直投",
+            detail={"channel": "active_registry"},
+            exc=exc,
+        )
+
+    celery_error: Exception | None = None
+    try:
+        celery_task.apply_async(
+            args=[project_id, step_id],
+            queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
+            countdown=0,
+            task_id=celery_task_id,
+        )
+    except Exception as exc:
+        celery_error = exc
+
+    if celery_error is None:
+        log_celery_dispatch_success(billing_context)
+        return
+    if registry_error is None:
+        log_celery_dispatch_failure(
+            billing_context,
+            error=f"Celery 直投失败，已保留 active registry 等待恢复：{celery_error}",
+        )
+        log_module_event_file(
+            module=MODULE,
+            event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
+            project_id=project_id,
+            step_id=step_id,
+            user_id=_safe_user_id(current_user),
+            message="Celery 直投失败，任务将由 active registry 恢复投递",
+            detail={"recoverable": True, "celery_task_id": celery_task_id},
+            error=str(celery_error),
+        )
+        return
+
+    await _mark_dispatch_failed_and_raise(
+        db,
+        current_user=current_user,
+        project_id=project_id,
+        step_id=step_id,
+        message=f"{error_prefix}: Redis 注册失败（{registry_error}）；Celery 投递失败（{celery_error}）",
+        billing_context=billing_context,
+    )
+
+
@router.get(
    "/spec",
    response_model=ShotReplicateSpecOut,
@@ -327,8 +535,16 @@ async def create_shot_task_set(
 ):
    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
    try:
-        task_set = await create_task_set(db, current_user=current_user, req=req)
-        task_set_id = task_set.id
+        task_set, created_new = await create_task_set(db, current_user=current_user, req=req)
+        task_set_id = str(task_set.id)
+        if not created_new:
+            # 幂等重复请求不重复预扣和投递；已有 pending 任务由原投递或恢复任务继续处理。
+            await db.rollback()
+            return await task_set_detail(db, current_user=_user_context(current_user), task_set_id=task_set_id)
+        analysis_attempt_no = max(1, int(task_set.analysis_attempt_no or 1))
+        celery_task_id = f"shot-analysis:task-set:{task_set_id}:attempt:{analysis_attempt_no}"
+        billing_context = build_task_set_analysis_billing_context(task_set)
+        billing_context.celery_task_id = celery_task_id
        await bind_upload_resources(
            db,
            user_id=current_user.id,
@@ -348,11 +564,19 @@ async def create_shot_task_set(
        _log_api_exception_from_locals(exc, locals(), f"创建拆镜总任务集失败: {exc}")
        raise HTTPException(status_code=500, detail=f"创建拆镜总任务集失败: {exc}")

+    log_celery_dispatch_start(billing_context)
    try:
        from app.tasks.shot_replicate_tasks import analyze_original_video

-        analyze_original_video.apply_async(args=[task_set_id], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0)
+        analyze_original_video.apply_async(
+            args=[task_set_id],
+            queue=CeleryQueue.GEN_SHOT_ANALYSIS.value,
+            countdown=0,
+            task_id=celery_task_id,
+        )
+        log_celery_dispatch_success(billing_context)
    except Exception as exc:
+        log_celery_dispatch_failure(billing_context, error=str(exc))
        _log_api_error(
            event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
            current_user=current_user,
@@ -361,6 +585,26 @@ async def create_shot_task_set(
            detail={"task_set_id": task_set_id, "task": "analyze_original_video"},
            exc=exc,
        )
+        try:
+            compensated = await mark_task_set_analysis_dispatch_failed(
+                db,
+                current_user=_user_context(current_user),
+                task_set_id=task_set_id,
+                error_message=f"拆镜分析任务投递失败: {exc}",
+            )
+            await db.commit()
+            if compensated:
+                log_celery_dispatch_compensated(billing_context, error=str(exc))
+        except Exception as mark_exc:
+            await db.rollback()
+            _log_api_error(
+                event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value,
+                current_user=current_user,
+                project_id=task_set_id,
+                message="拆镜分析任务投递失败后补偿失败",
+                detail={"task_set_id": task_set_id, "task": "analyze_original_video"},
+                exc=mark_exc,
+            )
        raise HTTPException(status_code=503, detail=f"拆镜分析任务投递失败: {exc}")

    return await task_set_detail(db, current_user=_user_context(current_user), task_set_id=task_set_id)
@@ -445,6 +689,16 @@ async def reanalyze_task_set(
            force=req.force,
            reason=req.reason,
        )
+        analysis_attempt_no = int(out.analysis_attempt_no)
+        celery_task_id = f"shot-analysis:task-set:{task_set_id}:attempt:{analysis_attempt_no}"
+        billing_context = _analysis_dispatch_billing_context(
+            user_id=str(current_user.id),
+            owner_id=task_set_id,
+            attempt_no=analysis_attempt_no,
+            task_set_id=task_set_id,
+            is_segment=False,
+            celery_task_id=celery_task_id,
+        )
        await db.commit()
    except HTTPException as exc:
        await db.rollback()
@@ -470,10 +724,17 @@ async def reanalyze_task_set(
        )
        raise HTTPException(status_code=500, detail=f"原视频再次分析状态重置失败: {exc}")

+    log_celery_dispatch_start(billing_context)
    try:
        from app.tasks.shot_replicate_tasks import analyze_original_video

-        analyze_original_video.apply_async(args=[task_set_id], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0)
+        analyze_original_video.apply_async(
+            args=[task_set_id],
+            queue=CeleryQueue.GEN_SHOT_ANALYSIS.value,
+            countdown=0,
+            task_id=celery_task_id,
+        )
+        log_celery_dispatch_success(billing_context)
        log_module_event_file(
            module=MODULE,
            event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_SUBMITTED.value,
@@ -483,6 +744,7 @@ async def reanalyze_task_set(
            detail={"task_set_id": task_set_id, "task": "analyze_original_video", "request": req.model_dump()},
        )
    except Exception as exc:
+        log_celery_dispatch_failure(billing_context, error=str(exc))
        _log_api_error(
            event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
            current_user=current_user,
@@ -491,6 +753,26 @@ async def reanalyze_task_set(
            detail={"task_set_id": task_set_id, "task": "analyze_original_video"},
            exc=exc,
        )
+        try:
+            compensated = await mark_task_set_analysis_dispatch_failed(
+                db,
+                current_user=_user_context(current_user),
+                task_set_id=task_set_id,
+                error_message=f"原视频再次分析任务投递失败: {exc}",
+            )
+            await db.commit()
+            if compensated:
+                log_celery_dispatch_compensated(billing_context, error=str(exc))
+        except Exception as mark_exc:
+            await db.rollback()
+            _log_api_error(
+                event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value,
+                current_user=current_user,
+                project_id=task_set_id,
+                message="原视频再次分析任务投递失败后补偿失败",
+                detail={"task_set_id": task_set_id, "task": "analyze_original_video"},
+                exc=mark_exc,
+            )
        raise HTTPException(status_code=503, detail=f"原视频再次分析任务投递失败: {exc}")
    out.message = "原视频再次分析任务已提交"
    return out
@@ -558,7 +840,38 @@ async def split_custom(

    from app.tasks.shot_replicate_tasks import split_one_segment

-    split_one_segment.apply_async(args=[segment_id], queue=CeleryQueue.GEN_SHOT_SPLIT.value, countdown=0)
+    try:
+        split_one_segment.apply_async(args=[segment_id], queue=CeleryQueue.GEN_SHOT_SPLIT.value, countdown=0)
+    except Exception as exc:
+        _log_api_error(
+            event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
+            current_user=current_user,
+            project_id=task_set_id,
+            step_id=segment_id,
+            message=f"自定义拆镜切片任务投递失败: {exc}",
+            detail={"segment_id": segment_id, "task_set_id": task_set_id, "task": "split_one_segment"},
+            exc=exc,
+        )
+        try:
+            await mark_custom_segment_split_dispatch_failed(
+                db,
+                current_user=_user_context(current_user),
+                segment_id=segment_id,
+                error_message=f"自定义拆镜切片任务投递失败: {exc}",
+            )
+            await db.commit()
+        except Exception as mark_exc:
+            await db.rollback()
+            _log_api_error(
+                event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value,
+                current_user=current_user,
+                project_id=task_set_id,
+                step_id=segment_id,
+                message="自定义拆镜切片投递失败后补偿失败",
+                detail={"segment_id": segment_id, "task_set_id": task_set_id},
+                exc=mark_exc,
+            )
+        raise HTTPException(status_code=503, detail=f"自定义拆镜切片任务投递失败: {exc}")
    return out


@@ -627,7 +940,17 @@ async def reanalyze_segment(
            force=req.force,
            reason=req.reason,
        )
-        task_set_id = out.task_set_id
+        task_set_id = str(out.task_set_id)
+        analysis_attempt_no = int(out.analysis_attempt_no)
+        celery_task_id = f"shot-analysis:segment:{segment_id}:attempt:{analysis_attempt_no}"
+        billing_context = _analysis_dispatch_billing_context(
+            user_id=str(current_user.id),
+            owner_id=segment_id,
+            attempt_no=analysis_attempt_no,
+            task_set_id=task_set_id,
+            is_segment=True,
+            celery_task_id=celery_task_id,
+        )
        await db.commit()
    except HTTPException as exc:
        await db.rollback()
@@ -653,10 +976,17 @@ async def reanalyze_segment(
        )
        raise HTTPException(status_code=500, detail=f"切片视频再次分析状态重置失败: {exc}")

+    log_celery_dispatch_start(billing_context)
    try:
        from app.tasks.shot_replicate_tasks import analyze_custom_segment_video

-        analyze_custom_segment_video.apply_async(args=[segment_id], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0)
+        analyze_custom_segment_video.apply_async(
+            args=[segment_id],
+            queue=CeleryQueue.GEN_SHOT_ANALYSIS.value,
+            countdown=0,
+            task_id=celery_task_id,
+        )
+        log_celery_dispatch_success(billing_context)
        log_module_event_file(
            module=MODULE,
            event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_SUBMITTED.value,
@@ -667,6 +997,7 @@ async def reanalyze_segment(
            detail={"segment_id": segment_id, "task_set_id": task_set_id, "task": "analyze_custom_segment_video", "request": req.model_dump()},
        )
    except Exception as exc:
+        log_celery_dispatch_failure(billing_context, error=str(exc))
        _log_api_error(
            event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value,
            current_user=current_user,
@@ -676,6 +1007,27 @@ async def reanalyze_segment(
            detail={"segment_id": segment_id, "task_set_id": task_set_id, "task": "analyze_custom_segment_video"},
            exc=exc,
        )
+        try:
+            compensated = await mark_segment_analysis_dispatch_failed(
+                db,
+                current_user=_user_context(current_user),
+                segment_id=segment_id,
+                error_message=f"切片视频再次分析任务投递失败: {exc}",
+            )
+            await db.commit()
+            if compensated:
+                log_celery_dispatch_compensated(billing_context, error=str(exc))
+        except Exception as mark_exc:
+            await db.rollback()
+            _log_api_error(
+                event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_MARK_FAILED.value,
+                current_user=current_user,
+                project_id=task_set_id,
+                step_id=segment_id,
+                message="切片视频再次分析任务投递失败后补偿失败",
+                detail={"segment_id": segment_id, "task_set_id": task_set_id, "task": "analyze_custom_segment_video"},
+                exc=mark_exc,
+            )
        raise HTTPException(status_code=503, detail=f"切片视频再次分析任务投递失败: {exc}")
    out.message = "切片视频再次分析任务已提交"
    return out
@@ -985,7 +1337,18 @@ async def generate_image_prompt(
    _ensure_celery_enabled(current_user=current_user, project_id=project_id, step_id=step_id)
    try:
        project, step = await submit_image_prompt_optimize(db, current_user=current_user, project_id=project_id, material_step_id=step_id, req=req)
-        project_id_value, step_id_value = project.id, step.id
+        project_id_value, step_id_value = str(project.id), str(step.id)
+        user_id_value = str(project.user_id)
+        attempt_no_value = int(step.version or 1)
+        celery_task_id = f"shot-replicate:image-prompt:{step_id_value}"
+        billing_context = _prompt_dispatch_billing_context(
+            user_id=user_id_value,
+            project_id=project_id_value,
+            step_id=step_id_value,
+            step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            attempt_no=attempt_no_value,
+            celery_task_id=celery_task_id,
+        )
        await db.commit()
    except HTTPException:
        await db.rollback()
@@ -995,19 +1358,21 @@ async def generate_image_prompt(
        _log_api_exception_from_locals(exc, locals(), f"提交图片 AI 提词失败: {exc}")
        raise HTTPException(status_code=500, detail=f"提交图片 AI 提词失败: {exc}")

-    try:
-        from app.tasks.shot_replicate_flow_tasks import start_image_prompt_optimize
+    from app.tasks.shot_replicate_flow_tasks import start_image_prompt_optimize

-        await register_module_step_task(
-            module=MODULE,
-            project_id=project_id_value,
-            step_id=step_id_value,
-            step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
-            task_name=TASK_SHOT_IMAGE_PROMPT,
-        )
-        start_image_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue=CeleryQueue.GEN_CHATAPI_CREATE.value, countdown=0)
-    except Exception as exc:
-        await _mark_dispatch_failed_and_raise(db, current_user=current_user, project_id=project_id_value, step_id=step_id_value, message=f"图片 AI 提词任务投递失败: {exc}")
+    log_celery_dispatch_start(billing_context)
+    await _dispatch_prompt_task(
+        db,
+        current_user=current_user,
+        project_id=project_id_value,
+        step_id=step_id_value,
+        step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+        task_name=TASK_SHOT_IMAGE_PROMPT,
+        celery_task=start_image_prompt_optimize,
+        celery_task_id=celery_task_id,
+        billing_context=billing_context,
+        error_prefix="图片 AI 提词任务投递失败",
+    )

    return ShotReplicateActionOut(message="图片 AI 提词任务已提交", project_id=project_id_value, step_id=step_id_value, detail=await _reload_project_detail(db, current_user, project_id_value))

@@ -1085,7 +1450,18 @@ async def generate_video_prompt(
    _ensure_celery_enabled(current_user=current_user, project_id=project_id, step_id=step_id)
    try:
        project, step = await submit_video_prompt_optimize(db, current_user=current_user, project_id=project_id, image_step_id=step_id, req=req)
-        project_id_value, step_id_value = project.id, step.id
+        project_id_value, step_id_value = str(project.id), str(step.id)
+        user_id_value = str(project.user_id)
+        attempt_no_value = int(step.version or 1)
+        celery_task_id = f"shot-replicate:video-prompt:{step_id_value}"
+        billing_context = _prompt_dispatch_billing_context(
+            user_id=user_id_value,
+            project_id=project_id_value,
+            step_id=step_id_value,
+            step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            attempt_no=attempt_no_value,
+            celery_task_id=celery_task_id,
+        )
        await db.commit()
    except HTTPException:
        await db.rollback()
@@ -1095,19 +1471,21 @@ async def generate_video_prompt(
        _log_api_exception_from_locals(exc, locals(), f"提交视频 AI 提词失败: {exc}")
        raise HTTPException(status_code=500, detail=f"提交视频 AI 提词失败: {exc}")

-    try:
-        from app.tasks.shot_replicate_flow_tasks import start_video_prompt_optimize
+    from app.tasks.shot_replicate_flow_tasks import start_video_prompt_optimize

-        await register_module_step_task(
-            module=MODULE,
-            project_id=project_id_value,
-            step_id=step_id_value,
-            step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
-            task_name=TASK_SHOT_VIDEO_PROMPT,
-        )
-        start_video_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue=CeleryQueue.GEN_CHATAPI_CREATE.value, countdown=0)
-    except Exception as exc:
-        await _mark_dispatch_failed_and_raise(db, current_user=current_user, project_id=project_id_value, step_id=step_id_value, message=f"视频 AI 提词任务投递失败: {exc}")
+    log_celery_dispatch_start(billing_context)
+    await _dispatch_prompt_task(
+        db,
+        current_user=current_user,
+        project_id=project_id_value,
+        step_id=step_id_value,
+        step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+        task_name=TASK_SHOT_VIDEO_PROMPT,
+        celery_task=start_video_prompt_optimize,
+        celery_task_id=celery_task_id,
+        billing_context=billing_context,
+        error_prefix="视频 AI 提词任务投递失败",
+    )

    return ShotReplicateActionOut(message="视频 AI 提词任务已提交", project_id=project_id_value, step_id=step_id_value, detail=await _reload_project_detail(db, current_user, project_id_value))