1from billing import BillingEngine, Subscription
 2
 3# Initialize billing engine
 4billing = BillingEngine()
 5
 6# Create subscription
 7subscription = billing.create_subscription(
 8    customer_id="cus_123",
 9    plan_id="plan_pro_monthly",
10    billing_cycle_anchor=datetime.now(),
11    trial_days=14
12)
13
14# Process billing cycle
15billing.process_billing_cycle(subscription.id)

 1from datetime import datetime, timedelta
 2from enum import Enum
 3
 4class SubscriptionStatus(Enum):
 5    TRIAL = "trial"
 6    ACTIVE = "active"
 7    PAST_DUE = "past_due"
 8    CANCELED = "canceled"
 9    PAUSED = "paused"
10
11class Subscription:
12    def __init__(self, customer_id, plan, billing_cycle_day=None):
13        self.id = generate_id()
14        self.customer_id = customer_id
15        self.plan = plan
16        self.status = SubscriptionStatus.TRIAL
17        self.current_period_start = datetime.now()
18        self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30)
19        self.billing_cycle_day = billing_cycle_day or self.current_period_start.day
20        self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None
21
22    def start_trial(self, trial_days):
23        """Start trial period."""
24        self.status = SubscriptionStatus.TRIAL
25        self.trial_end = datetime.now() + timedelta(days=trial_days)
26        self.current_period_end = self.trial_end
27
28    def activate(self):
29        """Activate subscription after trial or immediately."""
30        self.status = SubscriptionStatus.ACTIVE
31        self.current_period_start = datetime.now()
32        self.current_period_end = self.calculate_next_billing_date()
33
34    def mark_past_due(self):
35        """Mark subscription as past due after failed payment."""
36        self.status = SubscriptionStatus.PAST_DUE
37        # Trigger dunning workflow
38
39    def cancel(self, at_period_end=True):
40        """Cancel subscription."""
41        if at_period_end:
42            self.cancel_at_period_end = True
43            # Will cancel when current period ends
44        else:
45            self.status = SubscriptionStatus.CANCELED
46            self.canceled_at = datetime.now()
47
48    def calculate_next_billing_date(self):
49        """Calculate next billing date based on interval."""
50        if self.plan.interval == 'month':
51            return self.current_period_start + timedelta(days=30)
52        elif self.plan.interval == 'year':
53            return self.current_period_start + timedelta(days=365)
54        elif self.plan.interval == 'week':
55            return self.current_period_start + timedelta(days=7)

 1class BillingEngine:
 2    def process_billing_cycle(self, subscription_id):
 3        """Process billing for a subscription."""
 4        subscription = self.get_subscription(subscription_id)
 5
 6        # Check if billing is due
 7        if datetime.now() < subscription.current_period_end:
 8            return
 9
10        # Generate invoice
11        invoice = self.generate_invoice(subscription)
12
13        # Attempt payment
14        payment_result = self.charge_customer(
15            subscription.customer_id,
16            invoice.total
17        )
18
19        if payment_result.success:
20            # Payment successful
21            invoice.mark_paid()
22            subscription.advance_billing_period()
23            self.send_invoice(invoice)
24        else:
25            # Payment failed
26            subscription.mark_past_due()
27            self.start_dunning_process(subscription, invoice)
28
29    def generate_invoice(self, subscription):
30        """Generate invoice for billing period."""
31        invoice = Invoice(
32            customer_id=subscription.customer_id,
33            subscription_id=subscription.id,
34            period_start=subscription.current_period_start,
35            period_end=subscription.current_period_end
36        )
37
38        # Add subscription line item
39        invoice.add_line_item(
40            description=subscription.plan.name,
41            amount=subscription.plan.amount,
42            quantity=subscription.quantity or 1
43        )
44
45        # Add usage-based charges if applicable
46        if subscription.has_usage_billing:
47            usage_charges = self.calculate_usage_charges(subscription)
48            invoice.add_line_item(
49                description="Usage charges",
50                amount=usage_charges
51            )
52
53        # Calculate tax
54        tax = self.calculate_tax(invoice.subtotal, subscription.customer)
55        invoice.tax = tax
56
57        invoice.finalize()
58        return invoice
59
60    def charge_customer(self, customer_id, amount):
61        """Charge customer using saved payment method."""
62        customer = self.get_customer(customer_id)
63
64        try:
65            # Charge using payment processor
66            charge = stripe.Charge.create(
67                customer=customer.stripe_id,
68                amount=int(amount * 100),  # Convert to cents
69                currency='usd'
70            )
71
72            return PaymentResult(success=True, transaction_id=charge.id)
73        except stripe.error.CardError as e:
74            return PaymentResult(success=False, error=str(e))

 1class DunningManager:
 2    """Manage failed payment recovery."""
 3
 4    def __init__(self):
 5        self.retry_schedule = [
 6            {'days': 3, 'email_template': 'payment_failed_first'},
 7            {'days': 7, 'email_template': 'payment_failed_reminder'},
 8            {'days': 14, 'email_template': 'payment_failed_final'}
 9        ]
10
11    def start_dunning_process(self, subscription, invoice):
12        """Start dunning process for failed payment."""
13        dunning_attempt = DunningAttempt(
14            subscription_id=subscription.id,
15            invoice_id=invoice.id,
16            attempt_number=1,
17            next_retry=datetime.now() + timedelta(days=3)
18        )
19
20        # Send initial failure notification
21        self.send_dunning_email(subscription, 'payment_failed_first')
22
23        # Schedule retries
24        self.schedule_retries(dunning_attempt)
25
26    def retry_payment(self, dunning_attempt):
27        """Retry failed payment."""
28        subscription = self.get_subscription(dunning_attempt.subscription_id)
29        invoice = self.get_invoice(dunning_attempt.invoice_id)
30
31        # Attempt payment again
32        result = self.charge_customer(subscription.customer_id, invoice.total)
33
34        if result.success:
35            # Payment succeeded
36            invoice.mark_paid()
37            subscription.status = SubscriptionStatus.ACTIVE
38            self.send_dunning_email(subscription, 'payment_recovered')
39            dunning_attempt.mark_resolved()
40        else:
41            # Still failing
42            dunning_attempt.attempt_number += 1
43
44            if dunning_attempt.attempt_number < len(self.retry_schedule):
45                # Schedule next retry
46                next_retry_config = self.retry_schedule[dunning_attempt.attempt_number]
47                dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days'])
48                self.send_dunning_email(subscription, next_retry_config['email_template'])
49            else:
50                # Exhausted retries, cancel subscription
51                subscription.cancel(at_period_end=False)
52                self.send_dunning_email(subscription, 'subscription_canceled')
53
54    def send_dunning_email(self, subscription, template):
55        """Send dunning notification to customer."""
56        customer = self.get_customer(subscription.customer_id)
57
58        email_content = self.render_template(template, {
59            'customer_name': customer.name,
60            'amount_due': subscription.plan.amount,
61            'update_payment_url': f"https://app.example.com/billing"
62        })
63
64        send_email(
65            to=customer.email,
66            subject=email_content['subject'],
67            body=email_content['body']
68        )

 1class ProrationCalculator:
 2    """Calculate prorated charges for plan changes."""
 3
 4    @staticmethod
 5    def calculate_proration(old_plan, new_plan, period_start, period_end, change_date):
 6        """Calculate proration for plan change."""
 7        # Days in current period
 8        total_days = (period_end - period_start).days
 9
10        # Days used on old plan
11        days_used = (change_date - period_start).days
12
13        # Days remaining on new plan
14        days_remaining = (period_end - change_date).days
15
16        # Calculate prorated amounts
17        unused_amount = (old_plan.amount / total_days) * days_remaining
18        new_plan_amount = (new_plan.amount / total_days) * days_remaining
19
20        # Net charge/credit
21        proration = new_plan_amount - unused_amount
22
23        return {
24            'old_plan_credit': -unused_amount,
25            'new_plan_charge': new_plan_amount,
26            'net_proration': proration,
27            'days_used': days_used,
28            'days_remaining': days_remaining
29        }
30
31    @staticmethod
32    def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date):
33        """Calculate proration for seat changes."""
34        total_days = (period_end - period_start).days
35        days_remaining = (period_end - change_date).days
36
37        # Additional seats charge
38        additional_seats = new_seats - current_seats
39        prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining
40
41        return {
42            'additional_seats': additional_seats,
43            'prorated_charge': max(0, prorated_amount),  # No refund for removing seats mid-cycle
44            'effective_date': change_date
45        }

 1class TaxCalculator:
 2    """Calculate sales tax, VAT, GST."""
 3
 4    def __init__(self):
 5        # Tax rates by region
 6        self.tax_rates = {
 7            'US_CA': 0.0725,  # California sales tax
 8            'US_NY': 0.04,    # New York sales tax
 9            'GB': 0.20,       # UK VAT
10            'DE': 0.19,       # Germany VAT
11            'FR': 0.20,       # France VAT
12            'AU': 0.10,       # Australia GST
13        }
14
15    def calculate_tax(self, amount, customer):
16        """Calculate applicable tax."""
17        # Determine tax jurisdiction
18        jurisdiction = self.get_tax_jurisdiction(customer)
19
20        if not jurisdiction:
21            return 0
22
23        # Get tax rate
24        tax_rate = self.tax_rates.get(jurisdiction, 0)
25
26        # Calculate tax
27        tax = amount * tax_rate
28
29        return {
30            'tax_amount': tax,
31            'tax_rate': tax_rate,
32            'jurisdiction': jurisdiction,
33            'tax_type': self.get_tax_type(jurisdiction)
34        }
35
36    def get_tax_jurisdiction(self, customer):
37        """Determine tax jurisdiction based on customer location."""
38        if customer.country == 'US':
39            # US: Tax based on customer state
40            return f"US_{customer.state}"
41        elif customer.country in ['GB', 'DE', 'FR']:
42            # EU: VAT
43            return customer.country
44        elif customer.country == 'AU':
45            # Australia: GST
46            return 'AU'
47        else:
48            return None
49
50    def get_tax_type(self, jurisdiction):
51        """Get type of tax for jurisdiction."""
52        if jurisdiction.startswith('US_'):
53            return 'Sales Tax'
54        elif jurisdiction in ['GB', 'DE', 'FR']:
55            return 'VAT'
56        elif jurisdiction == 'AU':
57            return 'GST'
58        return 'Tax'
59
60    def validate_vat_number(self, vat_number, country):
61        """Validate EU VAT number."""
62        # Use VIES API for validation
63        # Returns True if valid, False otherwise
64        pass

 1class Invoice:
 2    def __init__(self, customer_id, subscription_id=None):
 3        self.id = generate_invoice_number()
 4        self.customer_id = customer_id
 5        self.subscription_id = subscription_id
 6        self.status = 'draft'
 7        self.line_items = []
 8        self.subtotal = 0
 9        self.tax = 0
10        self.total = 0
11        self.created_at = datetime.now()
12
13    def add_line_item(self, description, amount, quantity=1):
14        """Add line item to invoice."""
15        line_item = {
16            'description': description,
17            'unit_amount': amount,
18            'quantity': quantity,
19            'total': amount * quantity
20        }
21        self.line_items.append(line_item)
22        self.subtotal += line_item['total']
23
24    def finalize(self):
25        """Finalize invoice and calculate total."""
26        self.total = self.subtotal + self.tax
27        self.status = 'open'
28        self.finalized_at = datetime.now()
29
30    def mark_paid(self):
31        """Mark invoice as paid."""
32        self.status = 'paid'
33        self.paid_at = datetime.now()
34
35    def to_pdf(self):
36        """Generate PDF invoice."""
37        from reportlab.pdfgen import canvas
38
39        # Generate PDF
40        # Include: company info, customer info, line items, tax, total
41        pass
42
43    def to_html(self):
44        """Generate HTML invoice."""
45        template = """
46        <!DOCTYPE html>
47        <html>
48        <head><title>Invoice #{invoice_number}</title></head>
49        <body>
50            <h1>Invoice #{invoice_number}</h1>
51            <p>Date: {date}</p>
52            <h2>Bill To:</h2>
53            <p>{customer_name}<br>{customer_address}</p>
54            <table>
55                <tr><th>Description</th><th>Quantity</th><th>Amount</th></tr>
56                {line_items}
57            </table>
58            <p>Subtotal: ${subtotal}</p>
59            <p>Tax: ${tax}</p>
60            <h3>Total: ${total}</h3>
61        </body>
62        </html>
63        """
64
65        return template.format(
66            invoice_number=self.id,
67            date=self.created_at.strftime('%Y-%m-%d'),
68            customer_name=self.customer.name,
69            customer_address=self.customer.address,
70            line_items=self.render_line_items(),
71            subtotal=self.subtotal,
72            tax=self.tax,
73            total=self.total
74        )

 1class UsageBillingEngine:
 2    """Track and bill for usage."""
 3
 4    def track_usage(self, customer_id, metric, quantity):
 5        """Track usage event."""
 6        UsageRecord.create(
 7            customer_id=customer_id,
 8            metric=metric,
 9            quantity=quantity,
10            timestamp=datetime.now()
11        )
12
13    def calculate_usage_charges(self, subscription, period_start, period_end):
14        """Calculate charges for usage in billing period."""
15        usage_records = UsageRecord.get_for_period(
16            subscription.customer_id,
17            period_start,
18            period_end
19        )
20
21        total_usage = sum(record.quantity for record in usage_records)
22
23        # Tiered pricing
24        if subscription.plan.pricing_model == 'tiered':
25            charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers)
26        # Per-unit pricing
27        elif subscription.plan.pricing_model == 'per_unit':
28            charge = total_usage * subscription.plan.unit_price
29        # Volume pricing
30        elif subscription.plan.pricing_model == 'volume':
31            charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers)
32
33        return charge
34
35    def calculate_tiered_pricing(self, total_usage, tiers):
36        """Calculate cost using tiered pricing."""
37        charge = 0
38        remaining = total_usage
39
40        for tier in sorted(tiers, key=lambda x: x['up_to']):
41            tier_usage = min(remaining, tier['up_to'] - tier['from'])
42            charge += tier_usage * tier['unit_price']
43            remaining -= tier_usage
44
45            if remaining <= 0:
46                break
47
48        return charge