from __future__ import annotations from dataclasses import dataclass from decimal import Decimal, ROUND_FLOOR from django.utils import timezone from .models import ExchangeRate, PricingRule, normalize_resolution class BillingConfigError(RuntimeError): code = "billing_config_error" class NoPricingRuleError(BillingConfigError): code = "no_pricing_rule" def __init__(self, operation_type: str, alias: str, resolution: str | None = None): normalized_resolution = normalize_resolution(resolution) or "*" super().__init__( "No pricing rule configured for " f"operation={operation_type}, alias={alias}, resolution={normalized_resolution}." ) class NoExchangeRateError(BillingConfigError): code = "no_exchange_rate" def __init__(self, currency: str): super().__init__(f"No active exchange rate configured for currency={currency}.") @dataclass(frozen=True) class RechargeQuote: amount: Decimal currency: str points_per_unit: Decimal points_granted: int exchange_rate: ExchangeRate def get_pricing_rule( operation_type: str, alias: str, resolution: str | None = None, ) -> PricingRule: normalized_alias = str(alias or "").strip() normalized_resolution = normalize_resolution(resolution) rules = PricingRule.objects.filter( operation_type=operation_type, alias=normalized_alias, is_active=True, ) if normalized_resolution: exact_rule = rules.filter(resolution=normalized_resolution).first() if exact_rule is not None: return exact_rule default_rule = rules.filter(resolution="").first() if default_rule is not None: return default_rule raise NoPricingRuleError(operation_type, normalized_alias, normalized_resolution) def calculate_points_cost( operation_type: str, alias: str, resolution: str | None = None, ) -> int: return get_pricing_rule(operation_type, alias, resolution).points_cost def get_current_exchange_rate( *, currency: str = "CNY", at=None, ) -> ExchangeRate: normalized_currency = str(currency or "").strip().upper() effective_at = at or timezone.now() exchange_rate = ( ExchangeRate.objects.filter( currency=normalized_currency, is_active=True, effective_from__lte=effective_at, ) .order_by("-effective_from", "-id") .first() ) if exchange_rate is None: raise NoExchangeRateError(normalized_currency) return exchange_rate def quote_recharge_points( amount, *, currency: str = "CNY", at=None, ) -> RechargeQuote: decimal_amount = Decimal(str(amount)) if decimal_amount <= 0: raise ValueError("amount must be greater than 0") exchange_rate = get_current_exchange_rate(currency=currency, at=at) points_decimal = decimal_amount * exchange_rate.points_per_unit points_granted = int(points_decimal.to_integral_value(rounding=ROUND_FLOOR)) return RechargeQuote( amount=decimal_amount, currency=exchange_rate.currency, points_per_unit=exchange_rate.points_per_unit, points_granted=points_granted, exchange_rate=exchange_rate, ) def calculate_points_granted( amount, *, currency: str = "CNY", at=None, ) -> int: return quote_recharge_points(amount, currency=currency, at=at).points_granted