""" Billing Calculator Service Handles proration, usage-based billing, and invoice calculations """ import logging from datetime import datetime, timedelta from decimal import Decimal, ROUND_HALF_UP from typing import Optional, Dict, Any, List, Tuple from enum import Enum import json from sqlalchemy import select, and_, func from sqlalchemy.orm import Session from sqlalchemy.ext.asyncio import AsyncSession from app.models.billing import Invoice, Payment, UsageLog, UsageSummary, InvoiceStatus, PaymentStatus, PaymentMethod, UsageType from app.models.subscription import Subscription, Plan, SubscriptionStatus, BillingCycle from app.models.client import Client from app.config import settings logger = logging.getLogger(__name__) # Model-based pricing (cost per 1000 tokens in INR) MODEL_PRICING = { "gpt-4o": {"input": 0.375, "output": 1.125}, "gpt-4o-mini": {"input": 0.0113, "output": 0.045}, "gpt-4-turbo": {"input": 0.75, "output": 2.25}, "gpt-3.5-turbo": {"input": 0.0375, "output": 0.1125}, "claude-3-opus-20240229": {"input": 1.125, "output": 5.625}, "claude-3-sonnet-20240229": {"input": 0.225, "output": 1.125}, "claude-3-haiku-20240307": {"input": 0.01875, "output": 0.09375}, } # Overage pricing per unit (INR) OVERAGE_PRICING = { "message": Decimal("0.10"), # ₹0.10 per extra message "token": Decimal("0.0001"), # ₹0.0001 per extra token "vision_request": Decimal("0.50"), # ₹0.50 per extra vision request "whatsapp_message": Decimal("0.50"), # ₹0.50 per WhatsApp message } class ProrationMethod(str, Enum): """Proration calculation methods""" DAILY = "daily" MONTHLY = "monthly" class BillingCalculator: """Service for billing calculations - works with sync sessions (Celery workers)""" def __init__(self, session: Session): self.session = session # =========================================== # PRORATION FOR PLAN CHANGES # =========================================== def calculate_proration( self, subscription: Subscription, new_plan: Plan, new_billing_cycle: BillingCycle, change_date: Optional[datetime] = None, method: ProrationMethod = ProrationMethod.DAILY ) -> Dict[str, Any]: """ Calculate proration when changing plans Args: subscription: Current subscription new_plan: Plan to change to new_billing_cycle: New billing cycle change_date: When the change takes effect (default: now) method: Proration calculation method Returns: Dict with proration details including credits and charges """ change_date = change_date or datetime.utcnow() current_plan = subscription.plan # Get billing period dates period_start = subscription.start_date period_end = subscription.next_billing_date or self._calculate_period_end( period_start, subscription.billing_cycle ) # Calculate days in period and days remaining total_days = (period_end - period_start).days days_used = (change_date - period_start).days days_remaining = max(0, total_days - days_used) if method == ProrationMethod.DAILY: # Daily proration daily_rate_old = Decimal(str(subscription.amount)) / Decimal(str(max(total_days, 1))) credit_amount = daily_rate_old * Decimal(str(days_remaining)) else: # Monthly proration (based on month fractions) remaining_fraction = Decimal(str(days_remaining)) / Decimal(str(max(total_days, 1))) credit_amount = Decimal(str(subscription.amount)) * remaining_fraction # Round to 2 decimal places credit_amount = credit_amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) # Calculate new plan charge new_plan_price = self._get_plan_price(new_plan, new_billing_cycle) # If upgrading mid-cycle, charge prorated amount for remaining days if days_remaining > 0: new_period_end = self._calculate_period_end(change_date, new_billing_cycle) new_total_days = (new_period_end - change_date).days daily_rate_new = new_plan_price / Decimal(str(max(new_total_days, 1))) charge_amount = daily_rate_new * Decimal(str(days_remaining)) else: charge_amount = new_plan_price charge_amount = charge_amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) # Net amount (positive = charge, negative = credit) net_amount = charge_amount - credit_amount return { "change_date": change_date.isoformat(), "current_plan": { "name": current_plan.name, "amount": float(subscription.amount), "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), }, "new_plan": { "name": new_plan.name, "price": float(new_plan_price), "billing_cycle": new_billing_cycle.value, }, "proration": { "total_days": total_days, "days_used": days_used, "days_remaining": days_remaining, "credit_amount": float(credit_amount), "charge_amount": float(charge_amount), "net_amount": float(net_amount), "method": method.value, }, "is_upgrade": new_plan_price > Decimal(str(subscription.amount)), "immediate_charge": float(max(Decimal("0"), net_amount)), "credit_balance": float(max(Decimal("0"), -net_amount)), } def apply_plan_change( self, subscription: Subscription, new_plan: Plan, new_billing_cycle: BillingCycle, proration: Dict[str, Any] ) -> Tuple[Subscription, Optional[Invoice]]: """ Apply plan change with proration Args: subscription: Current subscription new_plan: New plan new_billing_cycle: New billing cycle proration: Proration calculation result Returns: Updated subscription and optional invoice for immediate charge """ change_date = datetime.fromisoformat(proration["change_date"]) net_amount = Decimal(str(proration["proration"]["net_amount"])) # Update subscription subscription.plan_id = new_plan.id subscription.billing_cycle = new_billing_cycle subscription.amount = Decimal(str(proration["new_plan"]["price"])) subscription.next_billing_date = self._calculate_period_end(change_date, new_billing_cycle) # Reset usage for new period subscription.messages_used = 0 subscription.vision_requests_used = 0 invoice = None if net_amount > 0: # Create invoice for immediate charge invoice = self._create_proration_invoice( subscription, proration, net_amount ) self.session.commit() return subscription, invoice # =========================================== # USAGE-BASED BILLING CALCULATION # =========================================== def calculate_usage_charges( self, client_id: int, billing_period_start: datetime, billing_period_end: datetime ) -> Dict[str, Any]: """ Calculate usage-based charges for a billing period Args: client_id: Client ID billing_period_start: Start of billing period billing_period_end: End of billing period Returns: Dict with detailed usage charges """ # Get client subscription and plan client = self.session.execute( select(Client).where(Client.id == client_id) ).scalar_one_or_none() if not client: return {"error": "Client not found"} # Get active subscription subscription = self.session.execute( select(Subscription).where( and_( Subscription.client_id == client_id, Subscription.status == SubscriptionStatus.ACTIVE ) ) ).scalar_one_or_none() if not subscription: return {"error": "No active subscription"} plan = self.session.execute( select(Plan).where(Plan.id == subscription.plan_id) ).scalar_one_or_none() if not plan: return {"error": "Plan not found"} # Get usage summary for the period billing_period = billing_period_start.strftime("%Y-%m") usage_summary = self.session.execute( select(UsageSummary).where( and_( UsageSummary.client_id == client_id, UsageSummary.period_type == "monthly", UsageSummary.period_date == billing_period ) ) ).scalar_one_or_none() charges = { "billing_period": billing_period, "base_plan": { "name": plan.name, "amount": float(subscription.amount), }, "included_limits": { "messages": plan.monthly_messages, "vision_requests": plan.vision_requests_monthly, }, "usage": { "messages": 0, "tokens": 0, "vision_requests": 0, "whatsapp_messages": 0, }, "overage": { "messages": {"quantity": 0, "rate": float(OVERAGE_PRICING["message"]), "amount": 0}, "vision_requests": {"quantity": 0, "rate": float(OVERAGE_PRICING["vision_request"]), "amount": 0}, "whatsapp_messages": {"quantity": 0, "rate": float(OVERAGE_PRICING["whatsapp_message"]), "amount": 0}, }, "ai_costs": [], "total_overage": Decimal("0"), "total_ai_cost": Decimal("0"), "subtotal": Decimal("0"), } if usage_summary: charges["usage"]["messages"] = usage_summary.total_messages charges["usage"]["tokens"] = usage_summary.total_tokens charges["usage"]["vision_requests"] = usage_summary.total_vision_requests # Get detailed usage logs for AI cost calculation usage_logs = self.session.execute( select(UsageLog).where( and_( UsageLog.client_id == client_id, UsageLog.billing_period == billing_period ) ) ).scalars().all() # Calculate AI costs by model model_usage = {} whatsapp_count = 0 for log in usage_logs: if log.usage_type == UsageType.WHATSAPP_MESSAGE: whatsapp_count += log.quantity if log.model and log.total_tokens > 0: if log.model not in model_usage: model_usage[log.model] = { "input_tokens": 0, "output_tokens": 0, "cost": Decimal("0"), } model_usage[log.model]["input_tokens"] += log.input_tokens model_usage[log.model]["output_tokens"] += log.output_tokens model_usage[log.model]["cost"] += Decimal(str(log.total_cost or 0)) charges["usage"]["whatsapp_messages"] = whatsapp_count # Calculate overage charges # Messages overage excess_messages = max(0, charges["usage"]["messages"] - plan.monthly_messages) if excess_messages > 0: overage_amount = Decimal(str(excess_messages)) * OVERAGE_PRICING["message"] charges["overage"]["messages"] = { "quantity": excess_messages, "rate": float(OVERAGE_PRICING["message"]), "amount": float(overage_amount), } charges["total_overage"] += overage_amount # Vision requests overage excess_vision = max(0, charges["usage"]["vision_requests"] - plan.vision_requests_monthly) if excess_vision > 0: overage_amount = Decimal(str(excess_vision)) * OVERAGE_PRICING["vision_request"] charges["overage"]["vision_requests"] = { "quantity": excess_vision, "rate": float(OVERAGE_PRICING["vision_request"]), "amount": float(overage_amount), } charges["total_overage"] += overage_amount # WhatsApp messages (all are billable extra) if whatsapp_count > 0: whatsapp_amount = Decimal(str(whatsapp_count)) * OVERAGE_PRICING["whatsapp_message"] charges["overage"]["whatsapp_messages"] = { "quantity": whatsapp_count, "rate": float(OVERAGE_PRICING["whatsapp_message"]), "amount": float(whatsapp_amount), } charges["total_overage"] += whatsapp_amount # AI costs breakdown for model, usage in model_usage.items(): charges["ai_costs"].append({ "model": model, "input_tokens": usage["input_tokens"], "output_tokens": usage["output_tokens"], "cost": float(usage["cost"]), }) charges["total_ai_cost"] += usage["cost"] # Calculate subtotal charges["subtotal"] = ( Decimal(str(subscription.amount)) + charges["total_overage"] + charges["total_ai_cost"] ) # Convert Decimals to float for JSON serialization charges["total_overage"] = float(charges["total_overage"]) charges["total_ai_cost"] = float(charges["total_ai_cost"]) charges["subtotal"] = float(charges["subtotal"]) return charges # =========================================== # AUTO-INVOICE GENERATION # =========================================== def generate_invoice( self, client_id: int, billing_period_start: datetime, billing_period_end: datetime, additional_items: Optional[List[Dict[str, Any]]] = None ) -> Invoice: """ Generate invoice for a client Args: client_id: Client ID billing_period_start: Billing period start billing_period_end: Billing period end additional_items: Additional line items Returns: Generated Invoice """ # Get usage charges usage_charges = self.calculate_usage_charges( client_id, billing_period_start, billing_period_end ) if "error" in usage_charges: raise ValueError(usage_charges["error"]) # Build line items line_items = [] # Base plan charge line_items.append({ "description": f"{usage_charges['base_plan']['name']} - Monthly Subscription", "hsn_sac": "998314", # SAC code for IT services "quantity": 1, "unit_price": usage_charges["base_plan"]["amount"], "amount": usage_charges["base_plan"]["amount"], "type": "subscription", }) # Overage charges for overage_type, overage in usage_charges["overage"].items(): if overage["quantity"] > 0: line_items.append({ "description": f"Additional {overage_type.replace('_', ' ').title()} ({overage['quantity']} units)", "hsn_sac": "998314", "quantity": overage["quantity"], "unit_price": overage["rate"], "amount": overage["amount"], "type": "overage", }) # AI costs (if applicable and not included in plan) if usage_charges["total_ai_cost"] > 0: for ai_cost in usage_charges["ai_costs"]: if ai_cost["cost"] > 0: line_items.append({ "description": f"AI Usage - {ai_cost['model']} ({ai_cost['input_tokens'] + ai_cost['output_tokens']:,} tokens)", "hsn_sac": "998314", "quantity": 1, "unit_price": ai_cost["cost"], "amount": ai_cost["cost"], "type": "ai_usage", }) # Additional items if additional_items: line_items.extend(additional_items) # Calculate totals subtotal = sum(Decimal(str(item["amount"])) for item in line_items) tax_percentage = Decimal(str(settings.TAX_PERCENTAGE)) # GST breakdown (CGST + SGST for same state, IGST for different state) cgst_rate = tax_percentage / 2 sgst_rate = tax_percentage / 2 cgst_amount = (subtotal * cgst_rate / 100).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) sgst_amount = (subtotal * sgst_rate / 100).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) tax_amount = cgst_amount + sgst_amount total_amount = subtotal + tax_amount # Get client client = self.session.execute( select(Client).where(Client.id == client_id) ).scalar_one() # Get subscription subscription = self.session.execute( select(Subscription).where( and_( Subscription.client_id == client_id, Subscription.status == SubscriptionStatus.ACTIVE ) ) ).scalar_one_or_none() # Create invoice invoice = Invoice( client_id=client_id, subscription_id=subscription.id if subscription else None, invoice_number=Invoice.generate_invoice_number(), status=InvoiceStatus.PENDING, billing_period_start=billing_period_start, billing_period_end=billing_period_end, subtotal=subtotal, tax_percentage=tax_percentage, tax_amount=tax_amount, discount_amount=Decimal("0"), total_amount=total_amount, currency="INR", line_items=line_items, issue_date=datetime.utcnow(), due_date=datetime.utcnow() + timedelta(days=settings.INVOICE_DUE_DAYS), notes=json.dumps({ "cgst_rate": float(cgst_rate), "cgst_amount": float(cgst_amount), "sgst_rate": float(sgst_rate), "sgst_amount": float(sgst_amount), }), ) self.session.add(invoice) self.session.commit() self.session.refresh(invoice) logger.info(f"Generated invoice {invoice.invoice_number} for client {client_id}") return invoice # =========================================== # HELPER METHODS # =========================================== def _get_plan_price(self, plan: Plan, billing_cycle: BillingCycle) -> Decimal: """Get plan price based on billing cycle""" if billing_cycle == BillingCycle.MONTHLY: return Decimal(str(plan.price_monthly)) elif billing_cycle == BillingCycle.QUARTERLY: return Decimal(str(plan.price_quarterly)) elif billing_cycle == BillingCycle.YEARLY: return Decimal(str(plan.price_yearly)) return Decimal("0") def _calculate_period_end( self, start_date: datetime, billing_cycle: BillingCycle ) -> datetime: """Calculate billing period end date""" if billing_cycle == BillingCycle.MONTHLY: month = start_date.month + 1 year = start_date.year if month > 12: month = 1 year += 1 day = min(start_date.day, 28) return datetime(year, month, day) elif billing_cycle == BillingCycle.QUARTERLY: month = start_date.month + 3 year = start_date.year while month > 12: month -= 12 year += 1 day = min(start_date.day, 28) return datetime(year, month, day) elif billing_cycle == BillingCycle.YEARLY: return datetime(start_date.year + 1, start_date.month, start_date.day) return start_date def _create_proration_invoice( self, subscription: Subscription, proration: Dict[str, Any], amount: Decimal ) -> Invoice: """Create proration invoice for plan change""" line_items = [ { "description": f"Plan Change: {proration['current_plan']['name']} → {proration['new_plan']['name']}", "hsn_sac": "998314", "quantity": 1, "unit_price": float(proration["proration"]["charge_amount"]), "amount": float(proration["proration"]["charge_amount"]), "type": "proration_charge", }, { "description": f"Credit for unused {proration['current_plan']['name']} ({proration['proration']['days_remaining']} days)", "hsn_sac": "998314", "quantity": 1, "unit_price": -float(proration["proration"]["credit_amount"]), "amount": -float(proration["proration"]["credit_amount"]), "type": "proration_credit", }, ] tax_percentage = Decimal(str(settings.TAX_PERCENTAGE)) tax_amount = (amount * tax_percentage / 100).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) total_amount = amount + tax_amount invoice = Invoice( client_id=subscription.client_id, subscription_id=subscription.id, invoice_number=Invoice.generate_invoice_number(), status=InvoiceStatus.PENDING, billing_period_start=datetime.fromisoformat(proration["change_date"]), billing_period_end=datetime.fromisoformat(proration["change_date"]), subtotal=amount, tax_percentage=tax_percentage, tax_amount=tax_amount, discount_amount=Decimal("0"), total_amount=total_amount, currency="INR", line_items=line_items, issue_date=datetime.utcnow(), due_date=datetime.utcnow(), # Immediate payment notes="Plan change proration invoice", ) self.session.add(invoice) return invoice class AsyncBillingCalculator: """Async version for FastAPI routes""" def __init__(self, session: AsyncSession): self.session = session async def calculate_proration( self, subscription_id: int, new_plan_id: int, new_billing_cycle: BillingCycle, change_date: Optional[datetime] = None ) -> Dict[str, Any]: """Calculate proration for plan change (async version)""" from sqlalchemy.orm import selectinload result = await self.session.execute( select(Subscription) .options(selectinload(Subscription.plan)) .where(Subscription.id == subscription_id) ) subscription = result.scalar_one_or_none() if not subscription: return {"error": "Subscription not found"} result = await self.session.execute( select(Plan).where(Plan.id == new_plan_id) ) new_plan = result.scalar_one_or_none() if not new_plan: return {"error": "Plan not found"} change_date = change_date or datetime.utcnow() current_plan = subscription.plan # Same calculation logic as sync version period_start = subscription.start_date period_end = subscription.next_billing_date or self._calculate_period_end( period_start, subscription.billing_cycle ) total_days = (period_end - period_start).days days_used = (change_date - period_start).days days_remaining = max(0, total_days - days_used) daily_rate_old = Decimal(str(subscription.amount)) / Decimal(str(max(total_days, 1))) credit_amount = daily_rate_old * Decimal(str(days_remaining)) credit_amount = credit_amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) new_plan_price = self._get_plan_price(new_plan, new_billing_cycle) if days_remaining > 0: new_period_end = self._calculate_period_end(change_date, new_billing_cycle) new_total_days = (new_period_end - change_date).days daily_rate_new = new_plan_price / Decimal(str(max(new_total_days, 1))) charge_amount = daily_rate_new * Decimal(str(days_remaining)) else: charge_amount = new_plan_price charge_amount = charge_amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) net_amount = charge_amount - credit_amount return { "change_date": change_date.isoformat(), "current_plan": { "id": current_plan.id, "name": current_plan.name, "amount": float(subscription.amount), "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), }, "new_plan": { "id": new_plan.id, "name": new_plan.name, "price": float(new_plan_price), "billing_cycle": new_billing_cycle.value, }, "proration": { "total_days": total_days, "days_used": days_used, "days_remaining": days_remaining, "credit_amount": float(credit_amount), "charge_amount": float(charge_amount), "net_amount": float(net_amount), }, "is_upgrade": new_plan_price > Decimal(str(subscription.amount)), "immediate_charge": float(max(Decimal("0"), net_amount)), "credit_balance": float(max(Decimal("0"), -net_amount)), } def _get_plan_price(self, plan: Plan, billing_cycle: BillingCycle) -> Decimal: """Get plan price based on billing cycle""" if billing_cycle == BillingCycle.MONTHLY: return Decimal(str(plan.price_monthly)) elif billing_cycle == BillingCycle.QUARTERLY: return Decimal(str(plan.price_quarterly)) elif billing_cycle == BillingCycle.YEARLY: return Decimal(str(plan.price_yearly)) return Decimal("0") def _calculate_period_end( self, start_date: datetime, billing_cycle: BillingCycle ) -> datetime: """Calculate billing period end date""" if billing_cycle == BillingCycle.MONTHLY: month = start_date.month + 1 year = start_date.year if month > 12: month = 1 year += 1 day = min(start_date.day, 28) return datetime(year, month, day) elif billing_cycle == BillingCycle.QUARTERLY: month = start_date.month + 3 year = start_date.year while month > 12: month -= 12 year += 1 day = min(start_date.day, 28) return datetime(year, month, day) elif billing_cycle == BillingCycle.YEARLY: return datetime(start_date.year + 1, start_date.month, start_date.day) return start_date