from __future__ import annotations from dataclasses import dataclass from datetime import datetime from app.enums.credit_record import CreditRecordChargeKind from app.enums.llm_billing import LlmBillingLedgerState from app.services.generation.billing_service import build_credit_biz_key class LlmBillingConfigurationError(RuntimeError): """LLM 场景固定预扣配置无效。""" class LlmBillingStateError(RuntimeError): """业务 attempt 的固定预扣状态不允许继续执行。""" class LlmProviderPostprocessError(RuntimeError): """供应商请求已成功并返回用量,但响应内容或业务结构处理失败。""" def __init__(self, message: str, *, usage: dict | None = None) -> None: super().__init__(message) self.usage = dict(usage or {}) @dataclass(slots=True, frozen=True) class LlmBillingPolicy: pre_deduct_credits: float valid: bool = True error: str | None = None policy_id: str | None = None version: int | None = None scene_name: str | None = None @dataclass(slots=True) class LlmBillingContext: user_id: str owner_type: str owner_id: str attempt_no: int charge_kind: str = CreditRecordChargeKind.TEXT_PROMPT.value billing_scene: str | None = None source_module: str | None = None source_project_id: str | None = None source_step_id: str | None = None source_step_code: str | None = None related_id: str | None = None description_prefix: str = "LLM" trace_id: str | None = None request_id: str | None = None celery_task_id: str | None = None provider: str | None = None model_name: str | None = None model_config_id: str | None = None model_parameters_snapshot: dict | None = None token_usage_id: str | None = None # 供应商响应成功事实先于审计落库设置。即使审计事务暂时失败, # 后续异常处理也不能把真实供应商成功覆盖成 provider failed。 provider_call_succeeded: bool = False provider_usage_snapshot: dict | None = None request_time: datetime | None = None billing_execution_id: str | None = None current_call_attempt_id: str | None = None @property def pre_deduct_biz_key(self) -> str: return build_credit_biz_key( owner_type=self.owner_type, owner_id=self.owner_id, attempt_no=self.attempt_no, charge_kind=self.charge_kind, action="pre_deduct", ) @property def refund_biz_key(self) -> str: return build_credit_biz_key( owner_type=self.owner_type, owner_id=self.owner_id, attempt_no=self.attempt_no, charge_kind=self.charge_kind, action="refund", ) @property def billing_biz_key(self) -> str: return self.pre_deduct_biz_key @dataclass(slots=True, frozen=True) class LlmPreDeductResult: amount: float state: LlmBillingLedgerState created: bool = False record_id: str | None = None reason: str | None = None execution_id: str | None = None @dataclass(slots=True, frozen=True) class LlmPreDeductValidation: can_execute: bool amount: float state: LlmBillingLedgerState reason: str | None = None pre_deduct_record_id: str | None = None execution_id: str | None = None