feat: add billing pricing rules

apps/billing/pricing.py

@@ -0,0 +1,122 @@
+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