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"""
<h2>Invoice {invoice.invoice_number}</h2>
<p>Dear {invoice.user.username},</p>
<p>Your invoice for the period {invoice.period_start} to {invoice.period_end} is now available.</p>
<table border="1">
<tr><th>Description</th><th>Amount</th></tr>
{"".join([f"<tr><td>{item.description}</td><td>${item.amount}</td></tr>" for item in line_items])}
<tr><td><strong>Subtotal</strong></td><td><strong>${invoice.subtotal}</strong></td></tr>
<tr><td><strong>Tax</strong></td><td><strong>${invoice.tax}</strong></td></tr>
<tr><td><strong>Total</strong></td><td><strong>${invoice.total}</strong></td></tr>
</table>
<p>Due Date: {invoice.due_date}</p>
<p>You can view and pay your invoice at: <a href="#">Invoice Link</a></p>
<p>Best regards,<br>The MyWebdav Team</p>
"""
queue_email(
to_email=invoice.user.email,
subject=f"Your MyWebdav Invoice {invoice.invoice_number}",
body=body,
html=html,
)
return invoice
@staticmethod
async def mark_invoice_paid(invoice: Invoice) -> Invoice:
invoice.status = "paid"
invoice.paid_at = datetime.now(timezone.utc)
await invoice.save()
return invoice
Reference in New Issue View Git Blame Copy Permalink