123 lines
3.4 KiB
Python
123 lines
3.4 KiB
Python
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
|