from __future__ import annotations from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP from typing import Any, Mapping from app.enums.model_pricing import ( ModelPricingBillingMode, ModelPricingCalculatorVersion, PricingBillBy, ) from app.services.model_pricing.usage_normalizer import safe_bool, safe_int MILLION = Decimal("1000000") MONEY_QUANT = Decimal("0.00000001") class PricingCalculationError(ValueError): pass @dataclass(slots=True) class PricingCalculationResult: amount: Decimal currency: str is_estimated: bool selected_rate: Decimal | None breakdown: dict[str, Any] usage_source: str def to_decimal(value: Any, default: str = "0") -> Decimal: try: if value in (None, ""): return Decimal(default) return Decimal(str(value)) except Exception: return Decimal(default) def money(value: Decimal) -> Decimal: return value.quantize(MONEY_QUANT, rounding=ROUND_HALF_UP) def _select_text_tier(rule_json: Mapping[str, Any], context_tokens: int) -> Mapping[str, Any]: for tier in rule_json.get("tiers") or []: maximum = tier.get("max_context_tokens") if maximum is None or context_tokens <= safe_int(maximum): return tier raise PricingCalculationError(f"没有匹配到文本 Token 档位: context_tokens={context_tokens}") def _calculate_text(rule_json: Mapping[str, Any], usage: Mapping[str, Any], currency: str) -> PricingCalculationResult: input_tokens = max(0, safe_int(usage.get("input_tokens"))) output_tokens = max(0, safe_int(usage.get("output_tokens"))) context_tokens = max(0, safe_int(usage.get("context_tokens"), input_tokens)) cached_input = max(0, min(input_tokens, safe_int(usage.get("cached_input_tokens")))) audio_input = max(0, min(input_tokens, safe_int(usage.get("audio_input_tokens")))) cached_audio = max(0, min(audio_input, safe_int(usage.get("cached_audio_input_tokens")))) tier = _select_text_tier(rule_json, context_tokens) cached_text_input = max(0, cached_input - cached_audio) normal_audio_input = max(0, audio_input - cached_audio) # cached_input_tokens may include cached audio tokens. Add cached_audio back once # so the four buckets always sum exactly to input_tokens. normal_text_input = max(0, input_tokens - audio_input - cached_text_input) input_rate = to_decimal(tier.get("input_rate")) output_rate = to_decimal(tier.get("output_rate")) cached_rate = to_decimal(tier.get("cached_input_rate"), str(input_rate)) audio_rate = to_decimal(tier.get("audio_input_rate"), str(input_rate)) cached_audio_rate = to_decimal(tier.get("cached_audio_input_rate"), str(cached_rate)) normal_input_cost = to_decimal(normal_text_input) * input_rate / MILLION cached_input_cost = to_decimal(cached_text_input) * cached_rate / MILLION audio_input_cost = to_decimal(normal_audio_input) * audio_rate / MILLION cached_audio_cost = to_decimal(cached_audio) * cached_audio_rate / MILLION output_cost = to_decimal(output_tokens) * output_rate / MILLION cache_storage_tokens = max(0, safe_int(usage.get("cache_storage_tokens"))) cache_storage_hours = max(Decimal("0"), to_decimal(usage.get("cache_storage_duration_hours"))) storage_rate = to_decimal(rule_json.get("cache_storage_rate_per_million_token_hour")) cache_storage_cost = to_decimal(cache_storage_tokens) * cache_storage_hours * storage_rate / MILLION total = money( normal_input_cost + cached_input_cost + audio_input_cost + cached_audio_cost + output_cost + cache_storage_cost ) return PricingCalculationResult( amount=total, currency=currency, is_estimated=False, selected_rate=None, usage_source=str(usage.get("usage_source") or "provider"), breakdown={ "formula": "token_items * corresponding_rate / 1e6", "context_tokens": context_tokens, "selected_tier": dict(tier), "normal_text_input_tokens": normal_text_input, "cached_text_input_tokens": cached_text_input, "normal_audio_input_tokens": normal_audio_input, "cached_audio_input_tokens": cached_audio, "output_tokens": output_tokens, "cache_storage_tokens": cache_storage_tokens, "cache_storage_duration_hours": str(cache_storage_hours), "normal_input_cost": str(money(normal_input_cost)), "cached_input_cost": str(money(cached_input_cost)), "audio_input_cost": str(money(audio_input_cost)), "cached_audio_input_cost": str(money(cached_audio_cost)), "output_cost": str(money(output_cost)), "cache_storage_cost": str(money(cache_storage_cost)), "total_cost": str(total), }, ) def _resolve_billable_output_count(rule_json: Mapping[str, Any], usage: Mapping[str, Any]) -> tuple[int, str, bool]: bill_by = str(rule_json.get("bill_by") or PricingBillBy.SUCCESSFUL_OUTPUT_COUNT.value) if bill_by == PricingBillBy.REQUESTED_OUTPUT_COUNT.value: return max(0, safe_int(usage.get("requested_output_count"))), bill_by, True if bill_by == PricingBillBy.PROVIDER_BILLED_COUNT.value: count = max(0, safe_int(usage.get("provider_billed_count"))) return count, bill_by, count <= 0 if bill_by != PricingBillBy.SUCCESSFUL_OUTPUT_COUNT.value: raise PricingCalculationError(f"不支持的图片计费数量来源: {bill_by}") return max(0, safe_int(usage.get("successful_output_count"), safe_int(usage.get("generated_image_count")))), bill_by, False def _calculate_image_per_output(rule_json: Mapping[str, Any], usage: Mapping[str, Any], currency: str) -> PricingCalculationResult: count, bill_by, count_estimated = _resolve_billable_output_count(rule_json, usage) if count <= 0: raise PricingCalculationError("图片计价缺少有效输出数量") rate = to_decimal(rule_json.get("output_rate")) total = money(to_decimal(count) * rate) return PricingCalculationResult( amount=total, currency=currency, is_estimated=count_estimated or safe_bool(usage.get("output_count_is_estimated")), selected_rate=rate, usage_source=str(usage.get("usage_source") or "provider_response"), breakdown={ "formula": "billable_output_count * output_rate", "bill_by": bill_by, "billable_output_count": count, "output_rate": str(rate), "total_cost": str(total), }, ) def _select_image_tier(output_tiers: list[Mapping[str, Any]], pixels: int) -> Mapping[str, Any]: for tier in output_tiers: maximum = tier.get("max_pixels") if maximum is None or pixels <= safe_int(maximum): return tier raise PricingCalculationError(f"没有匹配到图片输出像素档位: pixels={pixels}") def _calculate_image_tiered(rule_json: Mapping[str, Any], usage: Mapping[str, Any], currency: str) -> PricingCalculationResult: input_count = max(0, safe_int(usage.get("provider_input_image_count"), safe_int(usage.get("input_image_count")))) free_count = max(0, safe_int(rule_json.get("free_input_images"))) billable_input_count = max(0, input_count - free_count) input_rate = to_decimal(rule_json.get("input_image_rate")) input_cost = to_decimal(billable_input_count) * input_rate output_items = usage.get("output_items") or [] if not isinstance(output_items, list): output_items = [] count, bill_by, count_estimated = _resolve_billable_output_count(rule_json, usage) if count <= 0: raise PricingCalculationError("图片计价缺少有效输出数量") output_tiers = list(rule_json.get("output_tiers") or []) output_cost = Decimal("0") item_breakdown: list[dict[str, Any]] = [] pixels_estimated = False if output_items: priced_count = 0 for index, item in enumerate(output_items[:count]): if not isinstance(item, Mapping): continue pixels = max(0, safe_int(item.get("pixels"))) if pixels <= 0: width = max(0, safe_int(item.get("width"))) height = max(0, safe_int(item.get("height"))) pixels = width * height if pixels <= 0: raise PricingCalculationError(f"第 {index + 1} 张输出图片缺少明确像素") tier = _select_image_tier(output_tiers, pixels) rate = to_decimal(tier.get("rate")) output_cost += rate priced_count += 1 item_breakdown.append({"index": index, "pixels": pixels, "tier": dict(tier), "rate": str(rate)}) # provider_billed_count/requested_output_count may be greater than the returned # output item array. Only use an explicit fallback size; never silently under-bill. remaining = count - priced_count if remaining > 0: fallback_pixels = max(0, safe_int(usage.get("output_pixels"))) if fallback_pixels <= 0: fallback_width = max(0, safe_int(usage.get("output_width"))) fallback_height = max(0, safe_int(usage.get("output_height"))) fallback_pixels = fallback_width * fallback_height if fallback_pixels <= 0: raise PricingCalculationError(f"仍有 {remaining} 张计费输出缺少明确像素") tier = _select_image_tier(output_tiers, fallback_pixels) rate = to_decimal(tier.get("rate")) output_cost += to_decimal(remaining) * rate pixels_estimated = True item_breakdown.append( { "count": remaining, "pixels": fallback_pixels, "tier": dict(tier), "rate": str(rate), "size_source": "explicit_fallback", } ) else: pixels = max(0, safe_int(usage.get("output_pixels"))) if pixels <= 0: width = max(0, safe_int(usage.get("output_width"))) height = max(0, safe_int(usage.get("output_height"))) pixels = width * height if pixels <= 0: raise PricingCalculationError("图片输出缺少明确像素") tier = _select_image_tier(output_tiers, pixels) rate = to_decimal(tier.get("rate")) output_cost = to_decimal(count) * rate pixels_estimated = True item_breakdown = [{"count": count, "pixels": pixels, "tier": dict(tier), "rate": str(rate)}] total = money(input_cost + output_cost) return PricingCalculationResult( amount=total, currency=currency, is_estimated=count_estimated or pixels_estimated or safe_bool(usage.get("output_pixels_is_estimated")), selected_rate=None, usage_source=str(usage.get("usage_source") or "provider_response"), breakdown={ "formula": "billable_input_count*input_image_rate + sum(output_item_rate)", "bill_by": bill_by, "provider_input_image_count": input_count, "free_input_images": free_count, "billable_input_image_count": billable_input_count, "input_image_rate": str(input_rate), "input_cost": str(money(input_cost)), "billable_output_count": count, "output_items": item_breakdown, "output_cost": str(money(output_cost)), "total_cost": str(total), }, ) def _normalize_resolution(value: Any) -> str: resolution = str(value or "").lower().strip().replace(" ", "") return {"2160p": "4k", "uhd": "4k"}.get(resolution, resolution) def _normalize_ratio(value: Any) -> str: return str(value or "").strip().replace("：", ":") def _resolve_video_dimensions(rule_json: Mapping[str, Any], usage: Mapping[str, Any]) -> tuple[int, int, str]: width = max(0, safe_int(usage.get("output_width"))) height = max(0, safe_int(usage.get("output_height"))) if width > 0 and height > 0: return width, height, str(usage.get("dimension_source") or "provider_response") resolution = _normalize_resolution(usage.get("resolution")) ratio = _normalize_ratio(usage.get("aspect_ratio")) dimension_map = rule_json.get("dimension_map") or {} resolution_map = dimension_map.get(resolution) if isinstance(dimension_map, Mapping) else None value = resolution_map.get(ratio) if isinstance(resolution_map, Mapping) else None if isinstance(value, Mapping): width = max(0, safe_int(value.get("width"))) height = max(0, safe_int(value.get("height"))) elif isinstance(value, (list, tuple)) and len(value) >= 2: width, height = max(0, safe_int(value[0])), max(0, safe_int(value[1])) if width <= 0 or height <= 0: raise PricingCalculationError(f"视频缺少明确尺寸，且价格规则未配置 dimension_map: resolution={resolution}, ratio={ratio}") return width, height, "pricing_rule_map" def _rate_specificity(row: Mapping[str, Any]) -> tuple[int, int, int]: resolutions = {_normalize_resolution(v) for v in (row.get("resolutions") or [])} modes = {str(v).lower() for v in (row.get("inference_modes") or [])} constrained = int(bool(resolutions)) + int(row.get("has_input_video") is not None) + int( row.get("generate_audio") is not None ) + int(bool(modes)) # More constrained dimensions win; within a dimension, a smaller allowed set is # more specific. Empty sets represent wildcard and therefore score lowest. return constrained, -(len(resolutions) if resolutions else 10_000), -(len(modes) if modes else 10_000) def _match_video_rate(rule_json: Mapping[str, Any], usage: Mapping[str, Any]) -> Mapping[str, Any]: resolution = _normalize_resolution(usage.get("resolution")) has_input_video = safe_bool(usage.get("has_input_video")) generate_audio = safe_bool(usage.get("generate_audio")) inference_mode = str(usage.get("inference_mode") or "online").lower().strip() matched: list[Mapping[str, Any]] = [] for row in rule_json.get("rates") or []: resolutions = [_normalize_resolution(v) for v in (row.get("resolutions") or [])] if resolutions and resolution not in resolutions: continue if row.get("has_input_video") is not None and safe_bool(row.get("has_input_video")) != has_input_video: continue if row.get("generate_audio") is not None and safe_bool(row.get("generate_audio")) != generate_audio: continue modes = [str(v).lower() for v in (row.get("inference_modes") or [])] if modes and inference_mode not in modes: continue matched.append(row) if not matched: raise PricingCalculationError( "没有匹配到视频价格档位: " f"resolution={resolution}, has_input_video={has_input_video}, " f"generate_audio={generate_audio}, inference_mode={inference_mode}" ) matched.sort(key=_rate_specificity, reverse=True) if len(matched) > 1 and _rate_specificity(matched[0]) == _rate_specificity(matched[1]): raise PricingCalculationError("视频价格档位存在同等优先级重叠，请修正规则") return matched[0] def calculate_video_formula_tokens(rule_json: Mapping[str, Any], usage: Mapping[str, Any]) -> tuple[int, dict[str, Any]]: input_seconds = max(Decimal("0"), to_decimal(usage.get("input_video_duration_seconds"))) output_seconds = max(Decimal("0"), to_decimal(usage.get("output_video_duration_seconds"))) fps = max(Decimal("0"), to_decimal(usage.get("fps"))) if fps <= 0: fps = max(Decimal("0"), to_decimal(rule_json.get("default_fps"))) width, height, dimension_source = _resolve_video_dimensions(rule_json, usage) if output_seconds <= 0 or fps <= 0: raise PricingCalculationError("视频公式估算缺少输出时长或 FPS") if safe_bool(usage.get("has_input_video")) and input_seconds <= 0: raise PricingCalculationError("视频包含输入视频，但缺少输入视频时长，禁止估算") tokens = (input_seconds + output_seconds) * Decimal(width) * Decimal(height) * fps / Decimal("1024") rounded = max(0, int(tokens.quantize(Decimal("1"), rounding=ROUND_HALF_UP))) return rounded, { "input_video_duration_seconds": str(input_seconds), "output_video_duration_seconds": str(output_seconds), "output_width": width, "output_height": height, "fps": str(fps), "dimension_source": dimension_source, } def _calculate_video(rule_json: Mapping[str, Any], usage: Mapping[str, Any], currency: str) -> PricingCalculationResult: actual_tokens = max(0, safe_int(usage.get("total_tokens"))) formula_detail: dict[str, Any] = {} if actual_tokens > 0: total_tokens = actual_tokens use_formula = False else: total_tokens, formula_detail = calculate_video_formula_tokens(rule_json, usage) use_formula = True rate_row = _match_video_rate(rule_json, usage) rate = to_decimal(rate_row.get("rate")) total = money(to_decimal(total_tokens) * rate / MILLION) return PricingCalculationResult( amount=total, currency=currency, is_estimated=use_formula, selected_rate=rate, usage_source="request_formula" if use_formula else str(usage.get("usage_source") or "provider"), breakdown={ "formula": "billable_total_tokens * rate / 1e6", "token_source": "request_formula" if use_formula else "provider", "provider_total_tokens": actual_tokens, "billable_total_tokens": total_tokens, "formula_parameters": formula_detail or None, "selected_rate_rule": dict(rate_row), "rate": str(rate), "total_cost": str(total), }, ) def _expected_calculator(billing_mode: str) -> str: mapping = { ModelPricingBillingMode.TEXT_TOKEN_TIERED.value: ModelPricingCalculatorVersion.TEXT_TOKEN_TIERED_V1.value, ModelPricingBillingMode.IMAGE_PER_OUTPUT.value: ModelPricingCalculatorVersion.IMAGE_PER_OUTPUT_V1.value, ModelPricingBillingMode.IMAGE_INPUT_OUTPUT_TIERED.value: ModelPricingCalculatorVersion.IMAGE_INPUT_OUTPUT_TIERED_V1.value, ModelPricingBillingMode.VIDEO_TOKEN_RATE.value: ModelPricingCalculatorVersion.VIDEO_PIXEL_TOKEN_V1.value, } value = mapping.get(billing_mode) if not value: raise PricingCalculationError(f"不支持的计价模式: {billing_mode}") return value def _constraint_set(values: Any, *, normalize) -> set[str] | None: normalized = {normalize(value) for value in (values or []) if str(value or "").strip()} return normalized or None def _constraints_overlap(left: Mapping[str, Any], right: Mapping[str, Any]) -> bool: left_res = _constraint_set(left.get("resolutions"), normalize=_normalize_resolution) right_res = _constraint_set(right.get("resolutions"), normalize=_normalize_resolution) if left_res is not None and right_res is not None and left_res.isdisjoint(right_res): return False for key in ("has_input_video", "generate_audio"): lv, rv = left.get(key), right.get(key) if lv is not None and rv is not None and safe_bool(lv) != safe_bool(rv): return False left_modes = _constraint_set(left.get("inference_modes"), normalize=lambda value: str(value).lower()) right_modes = _constraint_set(right.get("inference_modes"), normalize=lambda value: str(value).lower()) if left_modes is not None and right_modes is not None and left_modes.isdisjoint(right_modes): return False return True def _constraint_subset(child: Mapping[str, Any], parent: Mapping[str, Any]) -> bool: child_res = _constraint_set(child.get("resolutions"), normalize=_normalize_resolution) parent_res = _constraint_set(parent.get("resolutions"), normalize=_normalize_resolution) if parent_res is not None and (child_res is None or not child_res.issubset(parent_res)): return False for key in ("has_input_video", "generate_audio"): child_value, parent_value = child.get(key), parent.get(key) if parent_value is not None and (child_value is None or safe_bool(child_value) != safe_bool(parent_value)): return False child_modes = _constraint_set(child.get("inference_modes"), normalize=lambda value: str(value).lower()) parent_modes = _constraint_set(parent.get("inference_modes"), normalize=lambda value: str(value).lower()) if parent_modes is not None and (child_modes is None or not child_modes.issubset(parent_modes)): return False return True def _validate_video_rate_overlaps(rates: list[Mapping[str, Any]]) -> None: for left_index, left in enumerate(rates): for right_index in range(left_index + 1, len(rates)): right = rates[right_index] if not _constraints_overlap(left, right): continue left_subset_right = _constraint_subset(left, right) right_subset_left = _constraint_subset(right, left) if left_subset_right and right_subset_left: raise PricingCalculationError("视频价格档位存在重复条件") if not left_subset_right and not right_subset_left: raise PricingCalculationError("视频价格档位存在交叉重叠，无法确定唯一价格") # The more specific row must be before its fallback row, matching the UI # and keeping exported rule JSON human-readable and deterministic. if right_subset_left: raise PricingCalculationError("视频价格档位顺序错误：具体条件必须放在通用兜底条件之前") def validate_pricing_rule(*, billing_mode: str, calculator_version: str, rule_json: Mapping[str, Any]) -> None: if calculator_version != _expected_calculator(billing_mode): raise PricingCalculationError(f"计价模式与计算器版本不匹配: {billing_mode}/{calculator_version}") if billing_mode == ModelPricingBillingMode.TEXT_TOKEN_TIERED.value: tiers = list(rule_json.get("tiers") or []) if not tiers: raise PricingCalculationError("文本计价至少需要一个 Token 档位") previous_max = 0 for index, tier in enumerate(tiers, start=1): maximum = tier.get("max_context_tokens") if maximum is None and index != len(tiers): raise PricingCalculationError("无上限 Token 档位只能放在最后") if maximum is not None: maximum_int = safe_int(maximum) if maximum_int <= previous_max: raise PricingCalculationError("Token 档位上限必须严格递增") previous_max = maximum_int for key in ("input_rate", "output_rate"): if to_decimal(tier.get(key), "-1") < 0: raise PricingCalculationError(f"{key} 不能为空且不能小于 0") return if billing_mode == ModelPricingBillingMode.IMAGE_PER_OUTPUT.value: if to_decimal(rule_json.get("output_rate"), "-1") < 0: raise PricingCalculationError("图片输出单价不能为空且不能小于 0") if str(rule_json.get("bill_by") or PricingBillBy.SUCCESSFUL_OUTPUT_COUNT.value) not in {v.value for v in PricingBillBy}: raise PricingCalculationError("bill_by 不受支持") return if billing_mode == ModelPricingBillingMode.IMAGE_INPUT_OUTPUT_TIERED.value: if safe_int(rule_json.get("free_input_images")) < 0: raise PricingCalculationError("免费输入图片数不能小于 0") if to_decimal(rule_json.get("input_image_rate"), "-1") < 0: raise PricingCalculationError("输入图片单价不能为空且不能小于 0") tiers = list(rule_json.get("output_tiers") or []) if not tiers: raise PricingCalculationError("图片输出至少需要一个像素档位") previous_max = 0 for index, tier in enumerate(tiers, start=1): maximum = tier.get("max_pixels") if maximum is None and index != len(tiers): raise PricingCalculationError("无上限像素档位只能放在最后") if maximum is not None: maximum_int = safe_int(maximum) if maximum_int <= previous_max: raise PricingCalculationError("图片像素档位上限必须严格递增") previous_max = maximum_int if to_decimal(tier.get("rate"), "-1") < 0: raise PricingCalculationError("图片输出单价不能为空且不能小于 0") return if billing_mode == ModelPricingBillingMode.VIDEO_TOKEN_RATE.value: rates = list(rule_json.get("rates") or []) if not rates: raise PricingCalculationError("视频计价至少需要一个价格档位") signatures: set[tuple[Any, ...]] = set() for row in rates: if to_decimal(row.get("rate"), "-1") < 0: raise PricingCalculationError("视频 Token 单价不能为空且不能小于 0") signature = ( tuple(sorted(_normalize_resolution(v) for v in (row.get("resolutions") or []))), row.get("has_input_video"), row.get("generate_audio"), tuple(sorted(str(v).lower() for v in (row.get("inference_modes") or []))), ) if signature in signatures: raise PricingCalculationError("视频价格档位存在重复条件") signatures.add(signature) _validate_video_rate_overlaps(rates) dimension_map = rule_json.get("dimension_map") or {} if dimension_map and not isinstance(dimension_map, Mapping): raise PricingCalculationError("dimension_map 必须是对象") return raise PricingCalculationError(f"不支持的计价模式: {billing_mode}") def calculate_pricing( *, billing_mode: str, calculator_version: str, rule_json: Mapping[str, Any], usage: Mapping[str, Any], currency: str = "CNY", ) -> PricingCalculationResult: validate_pricing_rule( billing_mode=billing_mode, calculator_version=calculator_version, rule_json=rule_json, ) if calculator_version == ModelPricingCalculatorVersion.TEXT_TOKEN_TIERED_V1.value: return _calculate_text(rule_json, usage, currency) if calculator_version == ModelPricingCalculatorVersion.IMAGE_PER_OUTPUT_V1.value: return _calculate_image_per_output(rule_json, usage, currency) if calculator_version == ModelPricingCalculatorVersion.IMAGE_INPUT_OUTPUT_TIERED_V1.value: return _calculate_image_tiered(rule_json, usage, currency) if calculator_version == ModelPricingCalculatorVersion.VIDEO_PIXEL_TOKEN_V1.value: return _calculate_video(rule_json, usage, currency) raise PricingCalculationError(f"不支持的计算器版本: {calculator_version}")