修复冻结积分BUG | 拆镜状态异常BUG

video-gen-api/app/services/generation/prompt_optimize_service.py

@@ -0,0 +1,647 @@
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from fastapi import HTTPException
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.enums.common import LogEventStatusEnum
+from app.enums.credit_record import (
+    CreditRecordBillingScene,
+    CreditRecordChargeKind,
+    CreditRecordSourceModule,
+)
+from app.enums.generation_record import (
+    GenerationRecordConfigSourceEnum,
+    GenerationRecordEventTypeEnum,
+)
+from app.enums.generation_status import (
+    ASPECT_RATIOS,
+    DURATIONS,
+    IMAGE_SIZES,
+    RESOLUTIONS,
+    GenerationStatus,
+    GenerationType,
+)
+from app.enums.llm_billing import LlmBillingConfigKey
+from app.models.generation_record import GenerationRecord
+from app.models.project import Project
+from app.schemas.generation import OptimizeParams
+from app.services.error_codes import extract_error_message
+from app.services.generation.ai.engine_service import (
+    get_image_engine,
+    get_video_engine,
+    image_supported_sizes,
+    parse_json_list,
+)
+from app.services.generation.billing_service import OWNER_GENERATION_RECORD
+from app.services.generation.media_reference_service import (
+    calculate_media_reference_usage,
+    validate_media_reference_usage_for_engine,
+)
+from app.services.generation.pipeline.generation_record_config_service import (
+    freeze_generation_record_config_with_log,
+    is_generation_record_config_complete,
+)
+from app.services.llm import optimize_prompt
+from app.services.llm_billing import (
+    LlmBillingContext,
+    log_provider_failure,
+    log_provider_start,
+    log_provider_success,
+    release_on_failure,
+    settle_success,
+    start_hold,
+)
+from app.services.operation_log_service import log_operation_error, log_operation_event
+from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
+from app.utils.id_gen import generate_id
+
+logger = logging.getLogger("videogen")
+
+_PROMPT_ATTEMPT_NO = 1
+_LOG_DOMAIN = "generation_record"
+_LOG_MODULE = "generation_record"
+_LOG_SOURCE = GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE.value
+
+
+@dataclass(slots=True, frozen=True)
+class PromptOptimizeServiceResult:
+    record_id: str
+    idempotent: bool = False
+
+
+def _log_event(
+    event: GenerationRecordEventTypeEnum,
+    *,
+    status: LogEventStatusEnum = LogEventStatusEnum.SUCCESS,
+    user_id: str,
+    project_id: str | None,
+    record_id: str | None,
+    detail: dict[str, Any] | None = None,
+    error: str | None = None,
+) -> None:
+    log_operation_event(
+        domain=_LOG_DOMAIN,
+        module=_LOG_MODULE,
+        event_type=event.value,
+        event_status=status.value,
+        source=_LOG_SOURCE,
+        user_id=user_id,
+        project_id=project_id,
+        task_id=record_id,
+        detail=detail,
+        error=error,
+    )
+
+
+def _billing_context(*, user_id: str, record_id: str, request_id: str | None) -> LlmBillingContext:
+    return LlmBillingContext(
+        user_id=user_id,
+        owner_type=OWNER_GENERATION_RECORD,
+        owner_id=record_id,
+        attempt_no=_PROMPT_ATTEMPT_NO,
+        charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value,
+        billing_scene=CreditRecordBillingScene.GENERATION_RECORD_TEXT_PROMPT_OPTIMIZE.value,
+        source_module=CreditRecordSourceModule.GENERATION_RECORD.value,
+        related_id=record_id,
+        hold_config_key=LlmBillingConfigKey.HOLD_GENERATION_RECORD_PROMPT.value,
+        description_prefix="AI创作提示词优化",
+        trace_id=f"generation-optimize:{record_id}",
+        request_id=request_id,
+    )
+
+
+def _canonical_references(value: object) -> str:
+    return json.dumps(value or [], ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _idempotency_config_matches(record: GenerationRecord, req: OptimizeParams) -> bool:
+    try:
+        existing_references = json.loads(record.media_references) if record.media_references else []
+    except (TypeError, json.JSONDecodeError):
+        return False
+    if (
+        str(record.project_id) != str(req.project_id)
+        or record.original_prompt != req.prompt
+        or record.gen_type != req.gen_type.value
+        or str(record.engine_id or "") != str(req.engine_id)
+        or bool(record.include_media_references) != bool(req.include_media_references)
+        or _canonical_references(existing_references) != _canonical_references(req.references)
+    ):
+        return False
+    if req.gen_type == GenerationType.video:
+        return (
+            record.duration == req.duration
+            and record.aspect_ratio == req.aspect_ratio
+            and record.resolution == req.resolution
+        )
+    return (
+        record.image_size == req.image_size
+        and record.image_proportion == req.image_proportion
+        and record.image_px == req.image_px
+    )
+
+
+def _validate_video_engine_selection(engine: Any, *, aspect_ratio: str, resolution: str, duration: int) -> None:
+    ratios = [str(item) for item in parse_json_list(engine.supported_ratios, [])]
+    resolutions = [str(item) for item in parse_json_list(engine.supported_resolutions, [])]
+    durations = [int(item) for item in parse_json_list(engine.supported_durations, []) if str(item).isdigit()]
+    if ratios and aspect_ratio not in ratios:
+        raise HTTPException(status_code=400, detail="当前视频引擎不支持所选画面比例")
+    if resolutions and resolution not in resolutions:
+        raise HTTPException(status_code=400, detail="当前视频引擎不支持所选分辨率")
+    if durations and duration not in durations:
+        raise HTTPException(status_code=400, detail="当前视频引擎不支持所选时长")
+    if int(engine.max_duration or 0) > 0 and duration > int(engine.max_duration):
+        raise HTTPException(status_code=400, detail="生成时长超过当前视频引擎上限")
+
+
+def _validate_image_engine_selection(engine: Any, *, image_size: str) -> None:
+    sizes = image_supported_sizes(engine)
+    if sizes and image_size not in sizes:
+        raise HTTPException(status_code=400, detail="当前图片引擎不支持所选画面分辨率")
+
+
+async def _find_idempotency_record(
+    db: AsyncSession,
+    *,
+    user_id: str,
+    idempotency_key: str | None,
+) -> GenerationRecord | None:
+    if not idempotency_key:
+        return None
+    stmt = (
+        select(GenerationRecord)
+        .where(
+            GenerationRecord.user_id == user_id,
+            GenerationRecord.idempotency_key == idempotency_key,
+            GenerationRecord.deleted_at.is_(None),
+        )
+        .order_by(GenerationRecord.created_at.desc(), GenerationRecord.id.desc())
+        .limit(1)
+    )
+    result = await db.execute(stmt)
+    return result.scalar_one_or_none()
+
+
+async def _settle_staged_result(
+    db: AsyncSession,
+    *,
+    record_id: str,
+    user_id: str,
+    project_name: str,
+    request_id: str | None,
+) -> PromptOptimizeServiceResult:
+    result = await db.execute(
+        select(GenerationRecord)
+        .where(
+            GenerationRecord.id == record_id,
+            GenerationRecord.user_id == user_id,
+            GenerationRecord.deleted_at.is_(None),
+        )
+        .with_for_update()
+        .limit(1)
+    )
+    record = result.scalar_one_or_none()
+    if record is None:
+        raise HTTPException(status_code=404, detail="生成记录不存在")
+    if record.status in {
+        GenerationStatus.prompt_optimized.value,
+        GenerationStatus.generating.value,
+        GenerationStatus.completed.value,
+    }:
+        await db.rollback()
+        return PromptOptimizeServiceResult(record_id=record_id, idempotent=True)
+    if record.status != GenerationStatus.settlement_pending.value:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail=f"当前提词状态不可结算：{record.status}")
+    if not record.optimized_prompt or not record.prompt_usage_snapshot_json:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="提词结果或计费快照缺失，需人工排查")
+
+    try:
+        usage = json.loads(record.prompt_usage_snapshot_json)
+    except (TypeError, json.JSONDecodeError) as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="提词计费快照损坏，需人工排查") from exc
+    if not isinstance(usage, dict):
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="提词计费快照格式错误，需人工排查")
+
+    project_id_snapshot = str(record.project_id)
+    status_snapshot = str(record.status)
+    ctx = _billing_context(user_id=user_id, record_id=record_id, request_id=request_id)
+    _log_event(
+        GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_SETTLEMENT_PENDING,
+        status=LogEventStatusEnum.STARTED,
+        user_id=user_id,
+        project_id=project_id_snapshot,
+        record_id=record_id,
+        detail={"status": status_snapshot, "attempt_no": _PROMPT_ATTEMPT_NO},
+    )
+    try:
+        billing = await settle_success(
+            db,
+            ctx,
+            usage=usage,
+            description=f"提示词优化 - {project_name}",
+        )
+        charge_item = next(
+            (item for item in billing.items if item.biz_key == ctx.charge_biz_key),
+            None,
+        )
+        record.text_credits_cost = round(float(charge_item.amount if charge_item else 0.0), 2)
+        record.text_tokens_used = int(usage.get("total_tokens", 0) or 0)
+        record.status = GenerationStatus.prompt_optimized.value
+        record.pipeline_stage = None
+        record.error_message = None
+        credits_snapshot = float(record.text_credits_cost or 0.0)
+        tokens_snapshot = int(record.text_tokens_used or 0)
+        await db.commit()
+    except Exception as exc:
+        await db.rollback()
+        _log_event(
+            GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_SETTLEMENT_PENDING,
+            status=LogEventStatusEnum.FAILED,
+            user_id=user_id,
+            project_id=project_id_snapshot,
+            record_id=record_id,
+            detail={"attempt_no": _PROMPT_ATTEMPT_NO, "error_type": type(exc).__name__},
+            error=str(exc),
+        )
+        raise HTTPException(status_code=503, detail="提词已生成，积分结算暂未完成，请使用相同幂等键重试") from exc
+
+    _log_event(
+        GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_SETTLEMENT_SUCCESS,
+        user_id=user_id,
+        project_id=project_id_snapshot,
+        record_id=record_id,
+        detail={
+            "attempt_no": _PROMPT_ATTEMPT_NO,
+            "text_credits_cost": credits_snapshot,
+            "text_tokens_used": tokens_snapshot,
+        },
+    )
+    return PromptOptimizeServiceResult(record_id=record_id, idempotent=False)
+
+
+async def _handle_existing_record(
+    db: AsyncSession,
+    *,
+    record: GenerationRecord,
+    req: OptimizeParams,
+    user_id: str,
+    project_name: str,
+) -> PromptOptimizeServiceResult:
+    record_id = str(record.id)
+    project_id = str(record.project_id)
+    status = str(record.status)
+    if not _idempotency_config_matches(record, req):
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="幂等键已绑定其他生成配置，请重新提交")
+
+    _log_event(
+        GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_IDEMPOTENCY_HIT,
+        status=LogEventStatusEnum.SKIPPED,
+        user_id=user_id,
+        project_id=project_id,
+        record_id=record_id,
+        detail={"record_status": status, "idempotency_key": req.idempotency_key},
+    )
+    if status == GenerationStatus.settlement_pending.value:
+        await db.rollback()
+        return await _settle_staged_result(
+            db,
+            record_id=record_id,
+            user_id=user_id,
+            project_name=project_name,
+            request_id=req.idempotency_key,
+        )
+    if status == GenerationStatus.optimizing.value:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="相同幂等请求正在处理，请勿重复提交")
+    if status == GenerationStatus.failed.value:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="该幂等请求已失败，请使用新的幂等键重新提交")
+    if record.optimized_prompt and is_generation_record_config_complete(record):
+        await db.rollback()
+        return PromptOptimizeServiceResult(record_id=record_id, idempotent=True)
+    await db.rollback()
+    raise HTTPException(status_code=409, detail="幂等记录配置不完整，需人工排查或使用新的幂等键")
+
+
+async def optimize_generation_prompt(
+    db: AsyncSession,
+    *,
+    req: OptimizeParams,
+    user_id: str,
+) -> PromptOptimizeServiceResult:
+    project_result = await db.execute(
+        select(Project).where(
+            Project.id == req.project_id,
+            Project.user_id == user_id,
+            Project.deleted_at.is_(None),
+        ).limit(1)
+    )
+    project = project_result.scalar_one_or_none()
+    if project is None:
+        raise HTTPException(status_code=404, detail="项目不存在")
+    project_name = str(project.name)
+    project_industry = str(project.industry or "")
+    project_id = str(project.id)
+
+    existing = await _find_idempotency_record(
+        db,
+        user_id=user_id,
+        idempotency_key=req.idempotency_key,
+    )
+    if existing is not None:
+        return await _handle_existing_record(
+            db,
+            record=existing,
+            req=req,
+            user_id=user_id,
+            project_name=project_name,
+        )
+
+    if req.gen_type == GenerationType.video:
+        if req.duration not in DURATIONS:
+            raise HTTPException(status_code=400, detail=f"视频时长必须为{DURATIONS}秒之一")
+        if req.aspect_ratio not in ASPECT_RATIOS:
+            raise HTTPException(status_code=400, detail="不支持的画面比例")
+        if req.resolution not in RESOLUTIONS:
+            raise HTTPException(status_code=400, detail="不支持的分辨率")
+        engine = await get_video_engine(db, req.engine_id)
+        _validate_video_engine_selection(
+            engine,
+            aspect_ratio=str(req.aspect_ratio),
+            resolution=str(req.resolution),
+            duration=int(req.duration),
+        )
+    else:
+        if req.image_size not in IMAGE_SIZES:
+            raise HTTPException(status_code=400, detail=f"图片分辨率必须为{IMAGE_SIZES}之一")
+        if not req.image_proportion or not req.image_px:
+            raise HTTPException(status_code=400, detail="图片生成需要指定比例和像素尺寸")
+        engine = await get_image_engine(db, req.engine_id)
+        _validate_image_engine_selection(engine, image_size=str(req.image_size))
+
+    reference_usage = calculate_media_reference_usage(
+        json.dumps(req.references, ensure_ascii=False) if req.references else None,
+        include=bool(req.include_media_references),
+    )
+    validate_media_reference_usage_for_engine(
+        reference_usage,
+        gen_type=req.gen_type.value,
+        engine=engine,
+    )
+
+    record_id = generate_id()
+    record = GenerationRecord(
+        id=record_id,
+        user_id=user_id,
+        project_id=project_id,
+        original_prompt=req.prompt,
+        optimized_prompt=None,
+        prompt_usage_snapshot_json=None,
+        gen_type=req.gen_type.value,
+        duration=req.duration if req.gen_type == GenerationType.video else None,
+        aspect_ratio=req.aspect_ratio if req.gen_type == GenerationType.video else None,
+        resolution=req.resolution if req.gen_type == GenerationType.video else None,
+        image_size=req.image_size if req.gen_type == GenerationType.image else None,
+        image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None,
+        image_px=req.image_px if req.gen_type == GenerationType.image else None,
+        status=GenerationStatus.optimizing.value,
+        pipeline_stage=None,
+        credits_cost=0,
+        text_credits_cost=0,
+        text_tokens_used=0,
+        media_references=json.dumps(req.references, ensure_ascii=False) if req.references else None,
+        include_media_references=bool(req.include_media_references),
+        idempotency_key=req.idempotency_key,
+        engine_id=req.engine_id,
+    )
+    if req.gen_type == GenerationType.video:
+        provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot(
+            db,
+            target_resolution=str(req.resolution),
+            aspect_ratio=str(req.aspect_ratio),
+            supported_provider_resolutions=parse_json_list(engine.supported_resolutions, []),
+        )
+        record.provider_generation_resolution = provider_resolution
+        record.video_upscale_enabled_snapshot = upscale_enabled
+        record.video_upscale_snapshot_json = upscale_snapshot_json
+    else:
+        record.provider_generation_resolution = None
+        record.video_upscale_enabled_snapshot = False
+        record.video_upscale_snapshot_json = None
+    freeze_generation_record_config_with_log(
+        record,
+        engine=engine,
+        source=GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE,
+    )
+    db.add(record)
+    try:
+        await db.flush()
+    except IntegrityError:
+        await db.rollback()
+        conflicting = await _find_idempotency_record(
+            db,
+            user_id=user_id,
+            idempotency_key=req.idempotency_key,
+        )
+        if conflicting is None:
+            raise
+        return await _handle_existing_record(
+            db,
+            record=conflicting,
+            req=req,
+            user_id=user_id,
+            project_name=project_name,
+        )
+
+    ctx = _billing_context(user_id=user_id, record_id=record_id, request_id=req.idempotency_key)
+    try:
+        await start_hold(db, ctx)
+        await db.commit()
+    except Exception:
+        await db.rollback()
+        raise
+
+    _log_event(
+        GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_PLACEHOLDER_CREATED,
+        user_id=user_id,
+        project_id=project_id,
+        record_id=record_id,
+        detail={
+            "idempotency_key": req.idempotency_key,
+            "gen_type": req.gen_type.value,
+            "engine_id": req.engine_id,
+            "attempt_no": _PROMPT_ATTEMPT_NO,
+            "reference_count": len(req.references or []),
+        },
+    )
+
+    log_provider_start(ctx, detail={"gen_type": req.gen_type.value})
+    try:
+        optimized_prompt, usage = await optimize_prompt(
+            db,
+            req.prompt,
+            user_id=user_id,
+            industry_key=project_industry,
+            duration=req.duration if req.gen_type == GenerationType.video else None,
+            image_size=req.image_size if req.gen_type == GenerationType.image else None,
+            image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None,
+            image_px=req.image_px if req.gen_type == GenerationType.image else None,
+            references=req.references,
+            gen_type=req.gen_type.value,
+            log_module="generation_record",
+            log_step="prompt_optimize",
+            log_project_id=project_id,
+            log_owner_type=OWNER_GENERATION_RECORD,
+            log_owner_id=record_id,
+            generation_attempt_no=_PROMPT_ATTEMPT_NO,
+        )
+        log_provider_success(ctx, usage=usage)
+    except Exception as exc:
+        await db.rollback()
+        log_provider_failure(ctx, error=str(exc))
+        compensated = False
+        try:
+            failed_result = await db.execute(
+                select(GenerationRecord)
+                .where(GenerationRecord.id == record_id, GenerationRecord.deleted_at.is_(None))
+                .with_for_update()
+                .limit(1)
+            )
+            failed_record = failed_result.scalar_one_or_none()
+            if failed_record is not None and failed_record.status == GenerationStatus.optimizing.value:
+                failed_record.status = GenerationStatus.failed.value
+                failed_record.error_message = extract_error_message(exc, "提示词")
+                await release_on_failure(db, ctx, error=str(exc))
+                compensated = True
+            await db.commit()
+        except Exception:
+            await db.rollback()
+            logger.exception("prompt optimize failure compensation failed: record_id=%s", record_id)
+            raise
+        _log_event(
+            GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_FAILED_RELEASED,
+            status=(LogEventStatusEnum.FAILED if compensated else LogEventStatusEnum.SKIPPED),
+            user_id=user_id,
+            project_id=project_id,
+            record_id=record_id,
+            detail={
+                "attempt_no": _PROMPT_ATTEMPT_NO,
+                "error_type": type(exc).__name__,
+                "compensated": compensated,
+            },
+            error=str(exc),
+        )
+        raise HTTPException(
+            status_code=502,
+            detail=f"AI模型调用失败: {extract_error_message(exc, '提示词')}",
+        ) from exc
+
+    usage_snapshot = dict(usage or {})
+    try:
+        input_tokens = int(usage_snapshot.get("input_tokens", 0) or 0)
+        output_tokens = int(usage_snapshot.get("output_tokens", 0) or 0)
+    except (TypeError, ValueError):
+        # 保留原始 usage 交给统一账务校验拒绝；这里只避免展示字段写入异常。
+        input_tokens = 0
+        output_tokens = 0
+    if input_tokens >= 0 and output_tokens >= 0:
+        reported_total = usage_snapshot.get("total_tokens")
+        normalized_total = input_tokens + output_tokens
+        if reported_total not in (None, ""):
+            try:
+                if int(reported_total) != normalized_total:
+                    usage_snapshot["reported_total_tokens"] = int(reported_total)
+            except (TypeError, ValueError):
+                usage_snapshot["reported_total_tokens"] = reported_total
+        usage_snapshot["total_tokens"] = normalized_total
+    usage_snapshot.setdefault("source_module", "generation_record")
+    usage_snapshot.setdefault("source_step_code", "prompt_optimize")
+    staged = False
+    last_stage_error: Exception | None = None
+    for _ in range(2):
+        try:
+            await db.rollback()
+            stage_result = await db.execute(
+                select(GenerationRecord)
+                .where(
+                    GenerationRecord.id == record_id,
+                    GenerationRecord.user_id == user_id,
+                    GenerationRecord.deleted_at.is_(None),
+                )
+                .with_for_update()
+                .limit(1)
+            )
+            staged_record = stage_result.scalar_one_or_none()
+            if staged_record is None:
+                raise RuntimeError("prompt optimize owner record missing")
+            if staged_record.status in {
+                GenerationStatus.prompt_optimized.value,
+                GenerationStatus.generating.value,
+                GenerationStatus.completed.value,
+            }:
+                await db.rollback()
+                return PromptOptimizeServiceResult(record_id=record_id, idempotent=True)
+            staged_record.optimized_prompt = optimized_prompt
+            staged_record.prompt_usage_snapshot_json = json.dumps(
+                usage_snapshot,
+                ensure_ascii=False,
+                sort_keys=True,
+                default=str,
+            )
+            staged_record.text_tokens_used = int(usage_snapshot.get("total_tokens", 0) or 0)
+            staged_record.status = GenerationStatus.settlement_pending.value
+            staged_record.pipeline_stage = None
+            staged_record.error_message = None
+            await db.commit()
+            staged = True
+            break
+        except Exception as exc:
+            last_stage_error = exc
+            await db.rollback()
+            logger.exception("prompt optimize provider result staging failed: record_id=%s", record_id)
+    if not staged:
+        log_operation_error(
+            domain=_LOG_DOMAIN,
+            event_type=GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_PROVIDER_RESULT_STAGED.value,
+            module=_LOG_MODULE,
+            source=_LOG_SOURCE,
+            user_id=user_id,
+            project_id=project_id,
+            task_id=record_id,
+            detail={"attempt_no": _PROMPT_ATTEMPT_NO, "stage": "provider_result_persistence"},
+            exc=last_stage_error or RuntimeError("unknown staging failure"),
+        )
+        raise HTTPException(status_code=503, detail="提词已生成但本地暂存失败，请联系管理员根据模型日志处理")
+
+    _log_event(
+        GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_PROVIDER_RESULT_STAGED,
+        user_id=user_id,
+        project_id=project_id,
+        record_id=record_id,
+        detail={
+            "attempt_no": _PROMPT_ATTEMPT_NO,
+            "input_tokens": usage_snapshot.get("input_tokens"),
+            "output_tokens": usage_snapshot.get("output_tokens"),
+            "total_tokens": usage_snapshot.get("total_tokens"),
+        },
+    )
+    return await _settle_staged_result(
+        db,
+        record_id=record_id,
+        user_id=user_id,
+        project_name=project_name,
+        request_id=req.idempotency_key,
+    )

video-gen-api/app/services/generation/refund_service.py

@@ -8,7 +8,7 @@ from app.models.chat_generation_task import ChatGenerationTask
 from app.models.credit_record import CreditRecord
 from app.models.generation_record import GenerationRecord
 from app.services.generation.pipeline.db_lock_service import execute_with_lock_timeout
-from app.services.credits import refund_credits
+from app.services.credits import add_credits_result
 from app.services.credit_record_meta_service import build_refund_meta_from_charge
 from app.services.generation.billing_service import (
     CHARGE_MEDIA,
@@ -113,17 +113,19 @@ async def refund_unrefunded_media_charges(
         amount = abs(_round2(charge.amount))
         if amount <= 0:
             continue
-        await refund_credits(
+        mutation = await add_credits_result(
             db,
             user_id=user_id,
             amount=amount,
             description=f"{description_prefix}失败积分回退",
             related_id=owner_id,
+            record_type="refund",
             biz_key=refund_biz_key,
             refund_for_biz_key=charge.biz_key,
             record_meta=build_refund_meta_from_charge(charge, attempt_no=attempt_no),
         )
-        total_refunded = round(total_refunded + amount, 2)
+        if mutation.created:
+            total_refunded = round(total_refunded + mutation.amount, 2)
     return total_refunded
 
 

video-gen-api/app/services/hot_opening_video_prompt_service.py

@@ -25,9 +25,8 @@ from app.enums.common import (
 )
 from app.enums.video_prompt_schema import PromptSchemaVersionEnum, VideoPromptSchemaUsageEnum
 from app.models.model_config import ModelConfig
-from app.models.token_usage import TokenUsage
-from app.utils.id_gen import generate_id
 from app.services.resource_signed_url_service import build_resource_signed_url
+from app.utils.id_gen import generate_id
 
 DEFAULT_FRAME_RATE = "30fps"
 DEFAULT_REFERENCE_VIDEO_FPS = 1
@@ -1599,7 +1598,13 @@ async def optimize_hot_opening_video_prompt(
     await db.rollback()
     if not config:
         result = normalize_video_prompt_schema_from_ai(_mock_result(video_config, target_platform), video_config, schema_config_snapshot)
-        return result, build_final_video_prompt(result), {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+        return result, build_final_video_prompt(result), {
+            "input_tokens": 0,
+            "output_tokens": 0,
+            "total_tokens": 0,
+            "usage_reported": True,
+            "billing_free": True,
+        }
 
     if use_base64:
         video_url_final = await media_to_base64(material_video_url, "video/mp4")
@@ -1739,27 +1744,20 @@ async def optimize_hot_opening_video_prompt(
             error=str(exc),
         )
         raise
-    usage = data.get("usage", {}) or {}
+    raw_usage = data.get("usage")
+    usage_reported = bool(
+        isinstance(raw_usage, dict)
+        and any(key in raw_usage for key in ("prompt_tokens", "completion_tokens", "total_tokens"))
+    )
+    usage = raw_usage if isinstance(raw_usage, dict) else {}
     token_usage = {
         "input_tokens": int(usage.get("prompt_tokens") or 0),
         "output_tokens": int(usage.get("completion_tokens") or 0),
         "total_tokens": int(usage.get("total_tokens") or 0),
+        "usage_reported": usage_reported,
         # "log_user_message": log_user_message,
     }
-    token_usage_id = generate_id()
-    db.add(
-        TokenUsage(
-            id=token_usage_id,
-            model_config_id=config.id,
-            user_id=user_id,
-            input_tokens=token_usage["input_tokens"],
-            output_tokens=token_usage["output_tokens"],
-            total_tokens=token_usage["total_tokens"],
-        )
-    )
-    await db.flush()
     token_usage.update({
-        "token_usage_id": token_usage_id,
         "model_config_id": config.id,
         "model_config_name": config.name,
         "model_provider": config.provider,

video-gen-api/app/services/llm.py

@@ -8,9 +8,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
 from app.models.model_config import ModelConfig
-from app.models.token_usage import TokenUsage
-from app.utils.id_gen import generate_id
 from app.services.operation_log_service import build_exception_detail, log_ai_model_event
+from app.utils.id_gen import generate_id
 
 
 
@@ -47,7 +46,13 @@ def _get_default_prompt(prompt: str, gen_type: str = "video") -> tuple[str, dict
             f"smooth camera movements. Theme: {prompt}. Cinematic shooting techniques with "
             f"rich lighting layers and strong visual impact, suitable for commercial distribution."
         )
-    return optimized, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+    return optimized, {
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "total_tokens": 0,
+        "usage_reported": True,
+        "billing_free": True,
+    }
 
 
 async def optimize_prompt(
@@ -135,7 +140,13 @@ def _mock_optimize(prompt: str, gen_type: str = "video") -> tuple[str, dict]:
     """Return a keyword-matched mock optimized prompt."""
     for keyword, optimized in MOCK_OPTIMIZED_PROMPTS.items():
         if keyword in prompt:
-            return optimized, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+            return optimized, {
+                "input_tokens": 0,
+                "output_tokens": 0,
+                "total_tokens": 0,
+                "usage_reported": True,
+                "billing_free": True,
+            }
     return _get_default_prompt(prompt, gen_type)
 
 
@@ -425,7 +436,15 @@ async def _call_openai_compatible(
             raise LLMProviderCallError(f"{type(exc).__name__}: {exc}") from exc
 
         try:
-            usage = data.get("usage", {})
+            raw_usage = data.get("usage")
+            usage_reported = bool(
+                isinstance(raw_usage, dict)
+                and any(
+                    key in raw_usage
+                    for key in ("prompt_tokens", "completion_tokens", "total_tokens")
+                )
+            )
+            usage = raw_usage if isinstance(raw_usage, dict) else {}
             input_tokens = int(usage.get("prompt_tokens", 0) or 0)
             output_tokens = int(usage.get("completion_tokens", 0) or 0)
             total_tokens = int(usage.get("total_tokens", input_tokens + output_tokens) or 0)
@@ -444,40 +463,7 @@ async def _call_openai_compatible(
             )
             raise LLMProviderCallError(f"模型响应解析失败: {exc}") from exc
 
-        token_usage_id = None
-        if db is not None:
-            try:
-                token_usage_id = generate_id()
-                record = TokenUsage(
-                    id=token_usage_id,
-                    model_config_id=config.id,
-                    user_id=user_id,
-                    input_tokens=input_tokens,
-                    output_tokens=output_tokens,
-                    total_tokens=total_tokens,
-                    source_module=log_module,
-                    source_step_code=log_step,
-                    owner_type=log_owner_type,
-                    owner_id=log_owner_id,
-                )
-                db.add(record)
-                await db.flush()
-            except Exception as exc:
-                log_ai_model_event(
-                    event_type="ERROR",
-                    event_phase="ERROR",
-                    event_status="failed",
-                    latency_ms=int((time.perf_counter() - started) * 1000),
-                    detail=build_exception_detail(exc, {"stage": "token_usage_persistence"}),
-                    error=str(exc),
-                    **common_log,
-                )
-                # A local transaction failure must not call a second provider after
-                # the first provider has already returned a valid response.
-                raise
-
         token_usage = {
-            "token_usage_id": token_usage_id,
             "model_config_id": config.id,
             "model_config_name": config.name,
             "model_provider": config.provider,
@@ -487,5 +473,6 @@ async def _call_openai_compatible(
             "input_tokens": input_tokens,
             "output_tokens": output_tokens,
             "total_tokens": total_tokens,
+            "usage_reported": usage_reported,
         }
         return content, token_usage

video-gen-api/app/services/llm_billing/__init__.py

@@ -19,6 +19,7 @@ from app.services.llm_billing.service import (
     release_on_failure,
     settle_success,
     start_hold,
+    validate_retryable_previous_attempt,
 )
 
 __all__ = [
@@ -31,6 +32,7 @@ __all__ = [
     "start_hold",
     "ensure_hold_exists",
     "get_llm_ledger_states",
+    "validate_retryable_previous_attempt",
     "log_provider_start",
     "log_provider_success",
     "log_provider_failure",

video-gen-api/app/services/llm_billing/service.py

@@ -4,6 +4,7 @@ from dataclasses import dataclass
 from typing import Any, Iterable, Mapping
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.enums.credit_record import (
@@ -210,8 +211,21 @@ def _action_valid(record: CreditRecord | None, expected: CreditRecordAction) ->
 
 
 
+def _record_meta_matches_context(record: CreditRecord | None, ctx: LlmBillingContext) -> bool:
+    if record is None:
+        return True
+    checks = (
+        (record.owner_type, ctx.owner_type),
+        (record.owner_id, ctx.owner_id),
+        (record.attempt_no, ctx.attempt_no),
+        (record.charge_kind, ctx.charge_kind),
+    )
+    return all(actual is None or str(actual) == str(expected) for actual, expected in checks)
+
+
 def _classify_ledger(
     *,
+    ctx: LlmBillingContext,
     hold: CreditRecord | None,
     release: CreditRecord | None,
     charge: CreditRecord | None,
@@ -222,17 +236,39 @@ def _classify_ledger(
         return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_action_mismatch")
     if not _action_valid(charge, CreditRecordAction.CHARGE):
         return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_action_mismatch")
+    if not _record_meta_matches_context(hold, ctx):
+        return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_context_mismatch")
+    if not _record_meta_matches_context(release, ctx):
+        return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_context_mismatch")
+    if not _record_meta_matches_context(charge, ctx):
+        return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_context_mismatch")
     if hold is None:
         if release is not None or charge is not None:
             return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_missing_with_followup")
         return _LedgerRecords(LlmBillingLedgerState.MISSING)
     if release is None and charge is None:
+        if hold.type != "consume" or _round2(float(hold.amount or 0)) >= 0:
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_type_or_sign_invalid")
         if _round2(abs(float(hold.amount or 0))) <= 0:
             return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_amount_not_positive")
         return _LedgerRecords(LlmBillingLedgerState.ACTIVE, hold)
     if release is not None and charge is None:
+        if release.type != "refund" or _round2(float(release.amount or 0)) <= 0:
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_type_or_sign_invalid")
+        if str(release.refund_for_biz_key or "") != str(ctx.hold_biz_key):
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_target_mismatch")
+        if _round2(release.amount) != _round2(abs(float(hold.amount or 0))):
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_amount_mismatch")
         return _LedgerRecords(LlmBillingLedgerState.RELEASED, hold, release)
     if release is not None and charge is not None:
+        if release.type != "refund" or _round2(float(release.amount or 0)) <= 0:
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_type_or_sign_invalid")
+        if str(release.refund_for_biz_key or "") != str(ctx.hold_biz_key):
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_target_mismatch")
+        if _round2(release.amount) != _round2(abs(float(hold.amount or 0))):
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_amount_mismatch")
+        if charge.type != "consume" or _round2(float(charge.amount or 0)) > 0:
+            return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_type_or_sign_invalid")
         return _LedgerRecords(LlmBillingLedgerState.CHARGED, hold, release, charge)
     return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_without_release")
 
@@ -264,7 +300,7 @@ async def _load_ledgers(
         hold = record_map.get((ctx.user_id, ctx.hold_biz_key))
         release = record_map.get((ctx.user_id, ctx.hold_release_biz_key))
         charge = record_map.get((ctx.user_id, ctx.charge_biz_key))
-        output[ctx.hold_biz_key] = _classify_ledger(hold=hold, release=release, charge=charge)
+        output[ctx.hold_biz_key] = _classify_ledger(ctx=ctx, hold=hold, release=release, charge=charge)
     return output
 
 
@@ -291,6 +327,104 @@ async def get_llm_ledger_states(
     }
 
 
+async def validate_retryable_previous_attempt(
+    db: AsyncSession,
+    ctx: LlmBillingContext,
+) -> LlmHoldValidation:
+    """校验失败 attempt 的账务是否已关闭，供业务创建下一 attempt 前调用。
+
+    只有已释放 HOLD，或当前计费明确关闭且旧 attempt 没有流水时，才允许创建
+    新 attempt。这里不主动退款，避免重试入口承担失败补偿职责。
+    """
+    _log(
+        ctx,
+        LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_VALIDATE_START,
+        status="started",
+    )
+    ledger = await _load_ledger(db, ctx)
+
+    if ledger.state == LlmBillingLedgerState.RELEASED:
+        result = LlmHoldValidation(
+            True,
+            ledger.hold_amount,
+            ledger.state,
+            hold_record_id=ledger.hold.id if ledger.hold else None,
+        )
+        _log(
+            ctx,
+            LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_VALIDATE_SUCCESS,
+            detail=_context_detail(
+                ctx,
+                ledger_state=result.state.value,
+                hold_credits=result.amount,
+                hold_record_id=result.hold_record_id,
+            ),
+        )
+        return result
+
+    if ledger.state == LlmBillingLedgerState.MISSING:
+        policy = await get_llm_billing_policy(
+            db,
+            config_key=ctx.hold_config_key,
+            explicit_hold_credits=None,
+        )
+        if policy.bypassed:
+            result = LlmHoldValidation(
+                True,
+                0.0,
+                LlmBillingLedgerState.BILLING_BYPASSED,
+                "billing_disabled",
+            )
+            _log(
+                ctx,
+                LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_VALIDATE_SUCCESS,
+                status="skipped",
+                detail=_context_detail(
+                    ctx,
+                    ledger_state=result.state.value,
+                    billing_bypassed=True,
+                ),
+            )
+            return result
+        if not policy.valid:
+            result = LlmHoldValidation(
+                False,
+                0.0,
+                LlmBillingLedgerState.INVALID,
+                policy.error or "billing_config_invalid",
+            )
+        else:
+            result = LlmHoldValidation(
+                False,
+                0.0,
+                ledger.state,
+                "previous_attempt_hold_missing",
+            )
+    else:
+        result = LlmHoldValidation(
+            False,
+            ledger.hold_amount,
+            ledger.state,
+            ledger.reason or f"previous_attempt_ledger_{ledger.state.value}",
+            ledger.hold.id if ledger.hold else None,
+        )
+
+    _log(
+        ctx,
+        LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_BLOCKED,
+        status="failed",
+        detail=_context_detail(
+            ctx,
+            ledger_state=result.state.value,
+            hold_credits=result.amount,
+            hold_record_id=result.hold_record_id,
+            skip_reason=result.reason,
+        ),
+        error="旧attempt账务尚未关闭，拒绝创建新的分析attempt",
+    )
+    return result
+
+
 async def start_hold(db: AsyncSession, ctx: LlmBillingContext) -> LlmHoldResult:
     # 幂等/异常 attempt 优先由已落库流水判定；只有全新 attempt 才读取配置。
     ledger = await _load_ledger(db, ctx)
@@ -659,6 +793,147 @@ async def release_on_failure(db: AsyncSession, ctx: LlmBillingContext, *, error:
     return BillingSummary(record_id=ctx.owner_id, user_id=ctx.user_id, items=[item])
 
 
+def _normalize_settlement_usage(
+    ctx: LlmBillingContext,
+    usage: Mapping[str, Any],
+) -> dict[str, Any]:
+    snapshot = dict(usage or {})
+    try:
+        if snapshot.get("usage_reported") is False and not bool(snapshot.get("billing_free")):
+            raise ValueError("provider_usage_missing")
+        if "input_tokens" not in snapshot or "output_tokens" not in snapshot:
+            raise ValueError("input_or_output_tokens_missing")
+        input_tokens = int(snapshot.get("input_tokens"))
+        output_tokens = int(snapshot.get("output_tokens"))
+        if input_tokens < 0 or output_tokens < 0:
+            raise ValueError("token_count_negative")
+        normalized_total = input_tokens + output_tokens
+        raw_total = snapshot.get("total_tokens")
+        if raw_total not in (None, ""):
+            total_tokens = int(raw_total)
+            if total_tokens < 0:
+                raise ValueError("total_tokens_negative")
+        else:
+            total_tokens = normalized_total
+        if total_tokens != normalized_total:
+            snapshot["reported_total_tokens"] = total_tokens
+            total_tokens = normalized_total
+        snapshot["input_tokens"] = input_tokens
+        snapshot["output_tokens"] = output_tokens
+        snapshot["total_tokens"] = total_tokens
+        return snapshot
+    except (TypeError, ValueError) as exc:
+        _log(
+            ctx,
+            LlmBillingEvent.USAGE_INVALID,
+            status="failed",
+            detail=_context_detail(ctx, usage=snapshot, reason=str(exc)),
+            error="LLM usage 无效，拒绝释放 HOLD 和创建真实扣费",
+        )
+        raise LlmBillingStateError(f"LLM usage 无效：{exc}") from exc
+
+
+async def _ensure_token_usage_once(
+    db: AsyncSession,
+    ctx: LlmBillingContext,
+    usage: dict[str, Any],
+) -> TokenUsage:
+    supplied_id = str(usage.get("token_usage_id") or "").strip() or None
+    if supplied_id:
+        supplied_result = await db.execute(
+            select(TokenUsage).where(TokenUsage.id == supplied_id).limit(1)
+        )
+        supplied = supplied_result.scalar_one_or_none()
+        if supplied is not None:
+            if supplied.user_id not in (None, ctx.user_id):
+                raise LlmBillingStateError("TokenUsage 用户归属与当前账务上下文不一致")
+            if supplied.biz_key not in (None, ctx.charge_biz_key):
+                raise LlmBillingStateError("TokenUsage biz_key 与当前 charge 不一致")
+            supplied.user_id = supplied.user_id or ctx.user_id
+            supplied.owner_type = supplied.owner_type or ctx.owner_type
+            supplied.owner_id = supplied.owner_id or ctx.owner_id
+            supplied.biz_key = supplied.biz_key or ctx.charge_biz_key
+            supplied.source_module = supplied.source_module or ctx.source_module
+            supplied.source_step_code = supplied.source_step_code or ctx.source_step_code
+            usage["token_usage_id"] = supplied.id
+            ctx.token_usage_id = supplied.id
+            _log(
+                ctx,
+                LlmBillingEvent.TOKEN_USAGE_REUSED,
+                detail=_context_detail(ctx, token_usage_id=supplied.id, source="supplied_id"),
+            )
+            return supplied
+
+    result = await db.execute(
+        select(TokenUsage)
+        .where(
+            TokenUsage.user_id == ctx.user_id,
+            TokenUsage.biz_key == ctx.charge_biz_key,
+        )
+        .limit(1)
+    )
+    existing = result.scalar_one_or_none()
+    if existing is not None:
+        usage["token_usage_id"] = existing.id
+        ctx.token_usage_id = existing.id
+        _log(
+            ctx,
+            LlmBillingEvent.TOKEN_USAGE_REUSED,
+            detail=_context_detail(ctx, token_usage_id=existing.id, source="biz_key"),
+        )
+        return existing
+
+    token_usage = TokenUsage(
+        id=generate_id(),
+        model_config_id=usage.get("model_config_id"),
+        user_id=ctx.user_id,
+        input_tokens=int(usage["input_tokens"]),
+        output_tokens=int(usage["output_tokens"]),
+        total_tokens=int(usage["total_tokens"]),
+        owner_type=ctx.owner_type,
+        owner_id=ctx.owner_id,
+        biz_key=ctx.charge_biz_key,
+        source_module=ctx.source_module,
+        source_step_code=ctx.source_step_code,
+    )
+    try:
+        async with db.begin_nested():
+            db.add(token_usage)
+            await db.flush()
+    except IntegrityError:
+        if token_usage in db.sync_session:
+            db.sync_session.expunge(token_usage)
+        result = await db.execute(
+            select(TokenUsage)
+            .where(
+                TokenUsage.user_id == ctx.user_id,
+                TokenUsage.biz_key == ctx.charge_biz_key,
+            )
+            .limit(1)
+        )
+        token_usage = result.scalar_one_or_none()
+        if token_usage is None:
+            raise
+        event = LlmBillingEvent.TOKEN_USAGE_REUSED
+    else:
+        event = LlmBillingEvent.TOKEN_USAGE_CREATED
+
+    usage["token_usage_id"] = token_usage.id
+    ctx.token_usage_id = token_usage.id
+    _log(
+        ctx,
+        event,
+        detail=_context_detail(
+            ctx,
+            token_usage_id=token_usage.id,
+            input_tokens=token_usage.input_tokens,
+            output_tokens=token_usage.output_tokens,
+            total_tokens=token_usage.total_tokens,
+        ),
+    )
+    return token_usage
+
+
 async def _build_charge_meta(db: AsyncSession, ctx: LlmBillingContext, usage: Mapping[str, Any]) -> CreditRecordMeta:
     usage_snapshot = dict(usage or {})
     ctx.provider = str(usage_snapshot.get("provider") or usage_snapshot.get("model_provider") or "") or ctx.provider
@@ -680,26 +955,6 @@ async def _build_charge_meta(db: AsyncSession, ctx: LlmBillingContext, usage: Ma
             usage=usage_snapshot,
         )
     if ctx.charge_kind == CreditRecordChargeKind.VIDEO_ANALYSIS.value:
-        if not usage_snapshot.get("token_usage_id"):
-            input_tokens = _safe_int(usage_snapshot.get("input_tokens"))
-            output_tokens = _safe_int(usage_snapshot.get("output_tokens"))
-            token_usage = TokenUsage(
-                id=generate_id(),
-                model_config_id=usage_snapshot.get("model_config_id"),
-                user_id=ctx.user_id,
-                input_tokens=input_tokens,
-                output_tokens=output_tokens,
-                total_tokens=_safe_int(usage_snapshot.get("total_tokens"), input_tokens + output_tokens),
-                owner_type=ctx.owner_type,
-                owner_id=ctx.owner_id,
-                biz_key=ctx.charge_biz_key,
-                source_module=ctx.source_module or CreditRecordSourceModule.SHOT_REPLICATE.value,
-                source_step_code=ctx.source_step_code,
-            )
-            db.add(token_usage)
-            await db.flush()
-            usage_snapshot["token_usage_id"] = token_usage.id
-            ctx.token_usage_id = token_usage.id
         return await build_shot_video_analysis_meta(
             db,
             owner_type=ctx.owner_type,
@@ -791,12 +1046,27 @@ async def _settle_success_impl(
             f"当前attempt账务状态为{ledger.state.value}，不能执行成功结算"
         )
 
-    _log(ctx, LlmBillingEvent.SETTLE_START, status="started", detail=_context_detail(ctx, ledger_state=ledger.state.value, usage=dict(usage or {})))
-    release_item = await _release_active_hold(db, ctx, hold_record=ledger.hold, reason="success")
-    input_tokens = _safe_int((usage or {}).get("input_tokens"))
-    output_tokens = _safe_int((usage or {}).get("output_tokens"))
+    normalized_usage = _normalize_settlement_usage(ctx, usage)
+    await _ensure_token_usage_once(db, ctx, normalized_usage)
+    _log(
+        ctx,
+        LlmBillingEvent.SETTLE_START,
+        status="started",
+        detail=_context_detail(
+            ctx,
+            ledger_state=ledger.state.value,
+            input_tokens=normalized_usage["input_tokens"],
+            output_tokens=normalized_usage["output_tokens"],
+            total_tokens=normalized_usage["total_tokens"],
+            provider=normalized_usage.get("provider") or normalized_usage.get("model_provider"),
+            model_name=normalized_usage.get("model_name") or normalized_usage.get("model"),
+        ),
+    )
+    input_tokens = int(normalized_usage["input_tokens"])
+    output_tokens = int(normalized_usage["output_tokens"])
     amount = await calc_text_credits(db, input_tokens, output_tokens)
-    meta = await _build_charge_meta(db, ctx, usage)
+    meta = await _build_charge_meta(db, ctx, normalized_usage)
+    release_item = await _release_active_hold(db, ctx, hold_record=ledger.hold, reason="success")
     if meta.charge_action is None:
         meta.charge_action = CreditRecordAction.CHARGE.value
     meta.billing_scene = meta.billing_scene or ctx.billing_scene
@@ -831,7 +1101,7 @@ async def _settle_success_impl(
         step = result.scalar_one_or_none()
         if step:
             step.token_usage_id = meta.token_usage_id
-            step.model_config_id = (usage or {}).get("model_config_id")
+            step.model_config_id = normalized_usage.get("model_config_id")
             step.input_tokens = meta.input_tokens
             step.output_tokens = meta.output_tokens
             step.total_tokens = meta.total_tokens

video-gen-api/app/services/shot_replicate_recovery_service.py

@@ -327,7 +327,7 @@ async def recover_shot_analysis_tasks_once(db: AsyncSession) -> dict[str, Any]:
         try:
             if owner_type == "task_set":
                 analyze_original_video.apply_async(
-                    args=[owner_id],
+                    args=[owner_id, attempt],
                     queue=CeleryQueue.GEN_SHOT_ANALYSIS.value,
                     countdown=0,
                     priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
@@ -335,7 +335,7 @@ async def recover_shot_analysis_tasks_once(db: AsyncSession) -> dict[str, Any]:
                 )
             else:
                 analyze_custom_segment_video.apply_async(
-                    args=[owner_id],
+                    args=[owner_id, attempt],
                     queue=CeleryQueue.GEN_SHOT_ANALYSIS.value,
                     countdown=0,
                     priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER,

video-gen-api/app/services/shot_replicate_taskset_service.py

@@ -54,7 +54,12 @@ from app.schemas.shot_replicate import (
     ShotTaskSetOut,
 )
 from app.services.module_generation_log_service import log_module_event_file
-from app.services.llm_billing import LlmBillingContext, release_on_failure, start_hold
+from app.services.llm_billing import (
+    LlmBillingContext,
+    release_on_failure,
+    start_hold,
+    validate_retryable_previous_attempt,
+)
 from app.services.resource_accounting_service import SOURCE_MODEL_SHOT_SEGMENT, soft_delete_resources_by_source
 from app.enums.upload_resource import UploadResourceModuleEnum, UploadResourceSourceModelEnum
 from app.services.upload_resource import release_upload_resources_by_source
@@ -398,10 +403,15 @@ async def _next_segment_index(db: AsyncSession, task_set_id: str) -> int:
     return int(result.scalar() or 0) + 1
 
 
-async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[str] | list[str]) -> None:
+async def refresh_task_set_split_summaries(
+    db: AsyncSession,
+    task_set_ids: set[str] | list[str],
+    *,
+    log_changes: bool = True,
+) -> list[dict[str, Any]]:
     ids = sorted({str(item) for item in task_set_ids if item})
     if not ids:
-        return
+        return []
 
     task_set_result = await db.execute(
         select(ShotReplicateTaskSet)
@@ -411,7 +421,11 @@ async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[s
     )
     task_sets = list(task_set_result.scalars().all())
     if not task_sets:
-        return
+        return []
+
+    # 项目 AsyncSession 关闭了 autoflush。聚合查询前必须把当前事务中刚修改的
+    # segment.split_status/deleted_at 等字段落到数据库，否则最后一个片段会少统计一次。
+    await db.flush()
 
     count_result = await db.execute(
         select(
@@ -441,14 +455,18 @@ async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[s
         for row in count_result.all()
     }
 
+    changes: list[dict[str, Any]] = []
     for task_set in task_sets:
         total, completed, failed = count_map.get(str(task_set.id), (0, 0, 0))
+        old_status = task_set.status
+        old_split_status = task_set.split_status
+        old_total = int(task_set.segment_count or 0)
+        old_completed = int(task_set.completed_segment_count or 0)
+        old_failed = int(task_set.failed_segment_count or 0)
         task_set.segment_count = total
         task_set.completed_segment_count = completed
         task_set.failed_segment_count = failed
 
-        old_status = task_set.status
-        old_split_status = task_set.split_status
         if total <= 0:
             task_set.split_status = ShotSplitStatusEnum.NONE.value
             if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value:
@@ -466,24 +484,39 @@ async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[s
             task_set.split_status = ShotSplitStatusEnum.PROCESSING.value
             task_set.status = ShotTaskSetStatusEnum.SPLITTING.value
 
-        if old_status != task_set.status or old_split_status != task_set.split_status:
-            log_module_event_file(
-                module=MODULE,
-                event_type="SHOT_SPLIT_STATUS_CHANGED",
-                project_id=task_set.id,
-                user_id=task_set.user_id,
-                message="拆镜总任务集拆分状态变更",
-                detail={
-                    "task_set_id": task_set.id,
-                    "from_status": old_status,
-                    "to_status": task_set.status,
-                    "from_split_status": old_split_status,
-                    "to_split_status": task_set.split_status,
-                    "segment_count": total,
-                    "completed_segment_count": completed,
-                    "failed_segment_count": failed,
-                },
-            )
+        changed = (
+            old_status != task_set.status
+            or old_split_status != task_set.split_status
+            or old_total != total
+            or old_completed != completed
+            or old_failed != failed
+        )
+        if changed:
+            change = {
+                "task_set_id": str(task_set.id),
+                "user_id": str(task_set.user_id),
+                "from_status": old_status,
+                "to_status": task_set.status,
+                "from_split_status": old_split_status,
+                "to_split_status": task_set.split_status,
+                "from_segment_count": old_total,
+                "segment_count": total,
+                "from_completed_segment_count": old_completed,
+                "completed_segment_count": completed,
+                "from_failed_segment_count": old_failed,
+                "failed_segment_count": failed,
+            }
+            changes.append(change)
+            if log_changes:
+                log_module_event_file(
+                    module=MODULE,
+                    event_type=ShotReplicateLogEventEnum.SPLIT_STATUS_CHANGED.value,
+                    project_id=task_set.id,
+                    user_id=task_set.user_id,
+                    message="拆镜总任务集拆分状态变更",
+                    detail=change,
+                )
+    return changes
 
 
 async def refresh_task_set_split_summary(db: AsyncSession, task_set_id: str) -> None:
@@ -1100,39 +1133,60 @@ async def prepare_reanalyze_task_set(
     *,
     current_user: User,
     task_set_id: str,
-    force: bool = False,
     reason: str | None = None,
 ) -> ShotReanalyzeOut:
-    """重置原视频分析状态，供 API 重新投递 Celery。"""
+    """仅对已失败且旧账务已关闭的原视频分析创建新 attempt。"""
     task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True)
-    if task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value:
+    previous_attempt_no = max(1, int(task_set.analysis_attempt_no or 1))
+    if task_set.analysis_status != ShotAnalysisStatusEnum.FAILED.value:
         log_module_event_file(
             module=MODULE,
             event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value,
             project_id=task_set.id,
             user_id=task_set.user_id,
-            message="原视频分析正在处理中，拒绝再次分析",
-            detail={"task_set_id": task_set.id, "analysis_status": task_set.analysis_status, "reason": reason},
+            message="原视频分析不是失败终态，拒绝再次分析",
+            detail={
+                "task_set_id": task_set.id,
+                "analysis_status": task_set.analysis_status,
+                "analysis_attempt_no": previous_attempt_no,
+                "reason": reason,
+            },
             event_status="rejected",
         )
-        raise HTTPException(status_code=409, detail="原视频分析正在处理中，不能重复投递")
-    if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value and not force:
-        raise HTTPException(status_code=409, detail="原视频分析已完成，如确需重跑请传 force=true")
-    if force:
-        active_segments_result = await db.execute(
-            select(func.count())
-            .select_from(ShotReplicateSegment)
-            .where(
-                ShotReplicateSegment.task_set_id == task_set.id,
-                ShotReplicateSegment.deleted_at.is_(None),
-            )
+        raise HTTPException(
+            status_code=409,
+            detail=f"只有分析失败的原视频任务才能重新分析，当前状态：{task_set.analysis_status}",
+        )
+
+    previous_context = build_task_set_analysis_billing_context(task_set)
+    previous_validation = await validate_retryable_previous_attempt(db, previous_context)
+    if not previous_validation.can_execute:
+        log_module_event_file(
+            module=MODULE,
+            event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value,
+            project_id=task_set.id,
+            user_id=task_set.user_id,
+            message="原视频旧分析 attempt 账务未关闭，拒绝再次分析",
+            detail={
+                "task_set_id": task_set.id,
+                "analysis_attempt_no": previous_attempt_no,
+                "ledger_state": previous_validation.state.value,
+                "ledger_reason": previous_validation.reason,
+                "reason": reason,
+            },
+            event_status="rejected",
+        )
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                "上一次原视频分析的冻结积分尚未完成释放或账务状态异常，"
+                f"当前账务状态：{previous_validation.state.value}"
+            ),
         )
-        if int(active_segments_result.scalar() or 0) > 0:
-            raise HTTPException(status_code=409, detail="当前总任务集已存在拆镜片段，不能强制重跑原视频分析")
 
     task_set.status = ShotTaskSetStatusEnum.PENDING_ANALYSIS.value
     task_set.analysis_status = ShotAnalysisStatusEnum.PENDING.value
-    task_set.analysis_attempt_no = max(1, int(task_set.analysis_attempt_no or 1)) + 1
+    task_set.analysis_attempt_no = previous_attempt_no + 1
     task_set.analysis_claim_token = None
     task_set.analysis_started_at = None
     task_set.analysis_lease_until = None
@@ -1151,7 +1205,14 @@ async def prepare_reanalyze_task_set(
         project_id=task_set.id,
         user_id=task_set.user_id,
         message="原视频再次分析已重置状态",
-        detail={"task_set_id": task_set.id, "force": force, "reason": reason, "video_url": task_set.video_url},
+        detail={
+            "task_set_id": task_set.id,
+            "previous_analysis_attempt_no": previous_attempt_no,
+            "analysis_attempt_no": int(task_set.analysis_attempt_no),
+            "previous_ledger_state": previous_validation.state.value,
+            "reason": reason,
+            "video_url": task_set.video_url,
+        },
     )
     return ShotReanalyzeOut(
         message="原视频再次分析任务已准备投递",
@@ -1168,34 +1229,69 @@ async def prepare_reanalyze_segment(
     *,
     current_user: User,
     segment_id: str,
-    force: bool = False,
     reason: str | None = None,
 ) -> ShotReanalyzeOut:
-    """重置自定义切片视频分析状态，供 API 重新投递 Celery。"""
+    """仅对已失败且旧账务已关闭的自定义切片分析创建新 attempt。"""
     segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True)
     if segment.split_status != ShotSplitStatusEnum.COMPLETED.value:
         raise HTTPException(status_code=409, detail="当前片段还未切割完成，不能再次分析")
     if not segment.segment_video_url:
         raise HTTPException(status_code=409, detail="当前片段缺少 segment_video_url，不能再次分析")
-    if segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value:
+    if segment.source_mode != ShotSegmentSourceModeEnum.CUSTOM.value:
+        raise HTTPException(status_code=409, detail="只有自定义切片视频支持重新分析")
+
+    previous_attempt_no = max(1, int(segment.analysis_attempt_no or 1))
+    if segment.analysis_status != ShotSegmentAnalysisStatusEnum.FAILED.value:
         log_module_event_file(
             module=MODULE,
             event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value,
             project_id=segment.task_set_id,
             step_id=segment.id,
             user_id=segment.user_id,
-            message="切片视频分析正在处理中，拒绝再次分析",
-            detail={"segment_id": segment.id, "analysis_status": segment.analysis_status, "reason": reason},
+            message="切片视频分析不是失败终态，拒绝再次分析",
+            detail={
+                "segment_id": segment.id,
+                "analysis_status": segment.analysis_status,
+                "analysis_attempt_no": previous_attempt_no,
+                "reason": reason,
+            },
             event_status="rejected",
         )
-        raise HTTPException(status_code=409, detail="切片视频分析正在处理中，不能重复投递")
-    if segment.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value and not force:
-        raise HTTPException(status_code=409, detail="切片视频分析已完成，如确需重跑请传 force=true")
-    if segment.source_mode != ShotSegmentSourceModeEnum.CUSTOM.value and not force:
-        raise HTTPException(status_code=409, detail="AI 建议片段默认无需单独分析，如确需重跑请传 force=true")
+        raise HTTPException(
+            status_code=409,
+            detail=f"只有分析失败的切片视频才能重新分析，当前状态：{segment.analysis_status}",
+        )
+
+    previous_context = build_segment_analysis_billing_context(segment)
+    previous_validation = await validate_retryable_previous_attempt(db, previous_context)
+    if not previous_validation.can_execute:
+        log_module_event_file(
+            module=MODULE,
+            event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value,
+            project_id=segment.task_set_id,
+            step_id=segment.id,
+            user_id=segment.user_id,
+            message="切片视频旧分析 attempt 账务未关闭，拒绝再次分析",
+            detail={
+                "segment_id": segment.id,
+                "task_set_id": segment.task_set_id,
+                "analysis_attempt_no": previous_attempt_no,
+                "ledger_state": previous_validation.state.value,
+                "ledger_reason": previous_validation.reason,
+                "reason": reason,
+            },
+            event_status="rejected",
+        )
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                "上一次切片视频分析的冻结积分尚未完成释放或账务状态异常，"
+                f"当前账务状态：{previous_validation.state.value}"
+            ),
+        )
 
     segment.analysis_status = ShotSegmentAnalysisStatusEnum.PENDING.value
-    segment.analysis_attempt_no = max(1, int(segment.analysis_attempt_no or 1)) + 1
+    segment.analysis_attempt_no = previous_attempt_no + 1
     segment.analysis_claim_token = None
     segment.analysis_started_at = None
     segment.analysis_lease_until = None
@@ -1216,7 +1312,15 @@ async def prepare_reanalyze_segment(
         step_id=segment.id,
         user_id=segment.user_id,
         message="切片视频再次分析已重置状态",
-        detail={"segment_id": segment.id, "task_set_id": segment.task_set_id, "force": force, "reason": reason, "video_url": segment.segment_video_url},
+        detail={
+            "segment_id": segment.id,
+            "task_set_id": segment.task_set_id,
+            "previous_analysis_attempt_no": previous_attempt_no,
+            "analysis_attempt_no": int(segment.analysis_attempt_no),
+            "previous_ledger_state": previous_validation.state.value,
+            "reason": reason,
+            "video_url": segment.segment_video_url,
+        },
     )
     return ShotReanalyzeOut(
         message="切片视频再次分析任务已准备投递",
@@ -1233,9 +1337,12 @@ async def mark_task_set_analysis_dispatch_failed(
     *,
     current_user: User,
     task_set_id: str,
+    expected_attempt_no: int,
     error_message: str,
 ) -> bool:
     task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True)
+    if int(task_set.analysis_attempt_no or 1) != int(expected_attempt_no):
+        return False
     if (
         task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value
         and task_set.analysis_claim_token
@@ -1245,12 +1352,11 @@ async def mark_task_set_analysis_dispatch_failed(
         return False
     if task_set.analysis_status in (ShotAnalysisStatusEnum.COMPLETED.value, ShotAnalysisStatusEnum.FAILED.value):
         return False
-    if task_set.analysis_status not in (ShotAnalysisStatusEnum.COMPLETED.value, ShotAnalysisStatusEnum.FAILED.value):
-        task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value
-        task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value
-        task_set.analysis_claim_token = None
-        task_set.analysis_lease_until = None
-        task_set.analysis_error_message = error_message
+    task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value
+    task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value
+    task_set.analysis_claim_token = None
+    task_set.analysis_lease_until = None
+    task_set.analysis_error_message = error_message
     await release_on_failure(db, build_task_set_analysis_billing_context(task_set), error=error_message)
     log_module_event_file(
         module=MODULE,
@@ -1269,9 +1375,12 @@ async def mark_segment_analysis_dispatch_failed(
     *,
     current_user: User,
     segment_id: str,
+    expected_attempt_no: int,
     error_message: str,
 ) -> bool:
     segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True)
+    if int(segment.analysis_attempt_no or 1) != int(expected_attempt_no):
+        return False
     if (
         segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value
         and segment.analysis_claim_token
@@ -1281,11 +1390,10 @@ async def mark_segment_analysis_dispatch_failed(
         return False
     if segment.analysis_status in (ShotSegmentAnalysisStatusEnum.COMPLETED.value, ShotSegmentAnalysisStatusEnum.FAILED.value):
         return False
-    if segment.analysis_status not in (ShotSegmentAnalysisStatusEnum.COMPLETED.value, ShotSegmentAnalysisStatusEnum.FAILED.value):
-        segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value
-        segment.analysis_claim_token = None
-        segment.analysis_lease_until = None
-        segment.analysis_error_message = error_message
+    segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value
+    segment.analysis_claim_token = None
+    segment.analysis_lease_until = None
+    segment.analysis_error_message = error_message
     await release_on_failure(db, build_segment_analysis_billing_context(segment), error=error_message)
     log_module_event_file(
         module=MODULE,

video-gen-api/app/services/shot_video_analysis_service.py

@@ -756,7 +756,12 @@ async def analyze_video_for_shot_split(
     result = ensure_result_schema(result)
     result = filter_and_normalize_breakdown(result, mode=mode)
 
-    usage = raw.get("usage") or {}
+    raw_usage = raw.get("usage")
+    usage_reported = bool(
+        isinstance(raw_usage, dict)
+        and any(key in raw_usage for key in ("prompt_tokens", "completion_tokens", "input_tokens", "output_tokens", "total_tokens"))
+    )
+    usage = raw_usage if isinstance(raw_usage, dict) else {}
     token_usage = {
         "input_tokens": _int_usage(usage.get("prompt_tokens") or usage.get("input_tokens")),
         "output_tokens": _int_usage(usage.get("completion_tokens") or usage.get("output_tokens")),
@@ -771,6 +776,7 @@ async def analyze_video_for_shot_split(
         "split_max_seconds": _split_max_seconds(),
         "analysis_mode": mode,
         "trace_id": trace_id,
+        "usage_reported": usage_reported,
     }
     if not token_usage["total_tokens"]:
         token_usage["total_tokens"] = token_usage["input_tokens"] + token_usage["output_tokens"]