from datetime import datetime, date, timedelta, timezone from decimal import Decimal from typing import Optional from calendar import monthrange from .models import Invoice, InvoiceLineItem, PricingConfig, UserSubscription from .usage_tracker import UsageTracker from .stripe_client import StripeClient from ..models import User class InvoiceGenerator: @staticmethod async def generate_monthly_invoice( user: User, year: int, month: int ) -> Optional[Invoice]: period_start = date(year, month, 1) days_in_month = monthrange(year, month)[1] period_end = date(year, month, days_in_month) usage = await UsageTracker.get_monthly_usage(user, year, month) pricing = await PricingConfig.all() pricing_dict = {p.config_key: p.config_value for p in pricing} storage_price_per_gb = pricing_dict.get( "storage_per_gb_month", Decimal("0.0045") ) bandwidth_price_per_gb = pricing_dict.get( "bandwidth_egress_per_gb", Decimal("0.009") ) free_storage_gb = pricing_dict.get("free_tier_storage_gb", Decimal("15")) free_bandwidth_gb = pricing_dict.get("free_tier_bandwidth_gb", Decimal("15")) tax_rate = pricing_dict.get("tax_rate_default", Decimal("0")) storage_gb = Decimal(str(usage["storage_gb_avg"])) bandwidth_gb = Decimal(str(usage["bandwidth_down_gb"])) billable_storage = max(Decimal("0"), storage_gb - free_storage_gb) billable_bandwidth = max(Decimal("0"), bandwidth_gb - free_bandwidth_gb) import math billable_storage_rounded = Decimal(math.ceil(float(billable_storage))) billable_bandwidth_rounded = Decimal(math.ceil(float(billable_bandwidth))) storage_cost = billable_storage_rounded * storage_price_per_gb bandwidth_cost = billable_bandwidth_rounded * bandwidth_price_per_gb subtotal = storage_cost + bandwidth_cost if subtotal <= 0: return None tax_amount = subtotal * tax_rate total = subtotal + tax_amount invoice_number = f"INV-{user.id:06d}-{year}{month:02d}" subscription = await UserSubscription.get_or_none(user=user) invoice = await Invoice.create( user=user, invoice_number=invoice_number, period_start=period_start, period_end=period_end, subtotal=subtotal, tax=tax_amount, total=total, currency="USD", status="draft", due_date=period_end + timedelta(days=7), metadata={ "usage": usage, "pricing": { "storage_per_gb": float(storage_price_per_gb), "bandwidth_per_gb": float(bandwidth_price_per_gb), }, }, ) if billable_storage_rounded > 0: await InvoiceLineItem.create( invoice=invoice, description=f"Storage usage for {period_start.strftime('%B %Y')} (Average: {storage_gb:.2f} GB, Billable: {billable_storage_rounded} GB)", quantity=billable_storage_rounded, unit_price=storage_price_per_gb, amount=storage_cost, item_type="storage", metadata={ "avg_gb": float(storage_gb), "free_gb": float(free_storage_gb), }, ) if billable_bandwidth_rounded > 0: await InvoiceLineItem.create( invoice=invoice, description=f"Bandwidth usage for {period_start.strftime('%B %Y')} (Total: {bandwidth_gb:.2f} GB, Billable: {billable_bandwidth_rounded} GB)", quantity=billable_bandwidth_rounded, unit_price=bandwidth_price_per_gb, amount=bandwidth_cost, item_type="bandwidth", metadata={ "total_gb": float(bandwidth_gb), "free_gb": float(free_bandwidth_gb), }, ) if subscription and subscription.stripe_customer_id: try: line_items = await invoice.line_items.all() stripe_line_items = [ { "amount": item.amount, "currency": "usd", "description": item.description, "metadata": item.metadata or {}, } for item in line_items ] stripe_invoice = await StripeClient.create_invoice( customer_id=subscription.stripe_customer_id, description=f"MyWebdav Usage Invoice for {period_start.strftime('%B %Y')}", line_items=stripe_line_items, metadata={"mywebdav_invoice_id": str(invoice.id)}, ) invoice.stripe_invoice_id = stripe_invoice.id await invoice.save() except Exception as e: print(f"Failed to create Stripe invoice: {e}") return invoice @staticmethod async def finalize_invoice(invoice: Invoice) -> Invoice: if invoice.status != "draft": raise ValueError("Only draft invoices can be finalized") invoice.status = "open" await invoice.save() if invoice.stripe_invoice_id: try: await StripeClient.finalize_invoice(invoice.stripe_invoice_id) except Exception as e: print(f"Failed to finalize Stripe invoice: {e}") # Send invoice email from ..mail import queue_email line_items = await invoice.line_items.all() items_text = "\n".join( [f"- {item.description}: ${item.amount}" for item in line_items] ) body = f"""Dear {invoice.user.username}, Your invoice {invoice.invoice_number} for the period {invoice.period_start} to {invoice.period_end} is now available. Invoice Details: {items_text} Subtotal: ${invoice.subtotal} Tax: ${invoice.tax} Total: ${invoice.total} Due Date: {invoice.due_date} You can view and pay your invoice at: {invoice.user.email} # Placeholder, should be a link to invoice page Best regards, The MyWebdav Team """ html = f"""
Dear {invoice.user.username},
Your invoice for the period {invoice.period_start} to {invoice.period_end} is now available.
| Description | Amount |
|---|---|
| {item.description} | ${item.amount} |
| Subtotal | ${invoice.subtotal} |
| Tax | ${invoice.tax} |
| Total | ${invoice.total} |
Due Date: {invoice.due_date}
You can view and pay your invoice at: Invoice Link
Best regards,
The MyWebdav Team