tools/burst/revenue.py (1920 bytes)
1 """The net server revenue of the last 30 days: the base of the burst spend cap. 2 3 Packs count for 30 days after the purchase. A subscription counts while it is 4 active and paid for the current period. Free credits count for nothing. Stripe 5 takes 2.9% + 30 cents of each payment, and 0.7% more of a subscription payment 6 (Stripe Billing). 7 """ 8 9 from __future__ import annotations 10 11 import math 12 from datetime import datetime, timedelta 13 14 from sqlalchemy import select 15 from sqlalchemy.orm import Session 16 17 from backend import pricing 18 from backend.db import Account, Purchase 19 20 21 def net_pack_cents(amount_cents: int) -> int: 22 return amount_cents - math.ceil(0.029 * amount_cents) - 30 23 24 25 def net_subscription_cents(amount_cents: int) -> int: 26 return amount_cents - math.ceil(0.036 * amount_cents) - 30 27 28 29 def _aware(when: datetime | None) -> datetime | None: 30 """SQLite gives back naive times. The app writes UTC.""" 31 if when is None or when.tzinfo is not None: 32 return when 33 from datetime import timezone 34 return when.replace(tzinfo=timezone.utc) 35 36 37 def revenue_30d_usd(session: Session, now: datetime) -> float: 38 since = now - timedelta(days=30) 39 total = 0 40 for plan_id, amount, created in session.execute( 41 select(Purchase.plan_id, Purchase.amount_cents, Purchase.created_at) 42 ): 43 plan = pricing.get(plan_id, retired=True) 44 if plan is None or plan.recurring or _aware(created) < since: 45 continue 46 total += net_pack_cents(amount) 47 for plan_id, ends in session.execute( 48 select(Account.subscription_plan_id, Account.subscription_period_end) 49 .where(Account.subscription_status == "active") 50 ): 51 plan = pricing.get(plan_id or "", retired=True) 52 if plan is None or not plan.recurring or ends is None or _aware(ends) <= now: 53 continue 54 total += net_subscription_cents(plan.price_cents) 55 return total / 100