名称: 计费自动化描述: 构建自动计费系统，用于循环付款、发票生成、订阅生命周期和催款管理。适用于实现订阅计费、自动化开票或管理循环付款系统。

计费自动化

掌握自动计费系统，包括循环计费、发票生成、催款管理、按比例计算和税务计算。

何时使用此技能

实现SaaS订阅计费
自动化发票生成和交付
管理失败付款恢复（催款）
计算计划更改的按比例费用
处理销售税、增值税和GST
处理基于用量的计费
管理计费周期和续订

核心概念

1. 计费周期

常见间隔：

月度（SaaS最常见）
年度（折扣长期）
季度
周度
自定义（基于用量、按席位）

2. 订阅状态

试用 → 活跃 → 逾期 → 取消
              → 暂停 → 恢复

3. 催款管理

自动化过程，通过以下方式恢复失败付款：

重试计划
客户通知
宽限期
账户限制

4. 按比例计算

调整费用当：

在周期中升级/降级
添加/移除席位
更改计费频率

快速开始

from billing import BillingEngine, Subscription

# 初始化计费引擎
billing = BillingEngine()

# 创建订阅
subscription = billing.create_subscription(
    customer_id="cus_123",
    plan_id="plan_pro_monthly",
    billing_cycle_anchor=datetime.now(),
    trial_days=14
)

# 处理计费周期
billing.process_billing_cycle(subscription.id)

订阅生命周期管理

from datetime import datetime, timedelta
from enum import Enum

class SubscriptionStatus(Enum):
    TRIAL = "trial"
    ACTIVE = "active"
    PAST_DUE = "past_due"
    CANCELED = "canceled"
    PAUSED = "paused"

class Subscription:
    def __init__(self, customer_id, plan, billing_cycle_day=None):
        self.id = generate_id()
        self.customer_id = customer_id
        self.plan = plan
        self.status = SubscriptionStatus.TRIAL
        self.current_period_start = datetime.now()
        self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30)
        self.billing_cycle_day = billing_cycle_day or self.current_period_start.day
        self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None

    def start_trial(self, trial_days):
        """开始试用期。"""
        self.status = SubscriptionStatus.TRIAL
        self.trial_end = datetime.now() + timedelta(days=trial_days)
        self.current_period_end = self.trial_end

    def activate(self):
        """在试用后或立即激活订阅。"""
        self.status = SubscriptionStatus.ACTIVE
        self.current_period_start = datetime.now()
        self.current_period_end = self.calculate_next_billing_date()

    def mark_past_due(self):
        """在付款失败后标记订阅为逾期。"""
        self.status = SubscriptionStatus.PAST_DUE
        # 触发催款工作流

    def cancel(self, at_period_end=True):
        """取消订阅。"""
        if at_period_end:
            self.cancel_at_period_end = True
            # 将在当前周期结束时取消
        else:
            self.status = SubscriptionStatus.CANCELED
            self.canceled_at = datetime.now()

    def calculate_next_billing_date(self):
        """基于间隔计算下一个计费日期。"""
        if self.plan.interval == 'month':
            return self.current_period_start + timedelta(days=30)
        elif self.plan.interval == 'year':
            return self.current_period_start + timedelta(days=365)
        elif self.plan.interval == 'week':
            return self.current_period_start + timedelta(days=7)

计费周期处理

class BillingEngine:
    def process_billing_cycle(self, subscription_id):
        """为订阅处理计费。"""
        subscription = self.get_subscription(subscription_id)

        # 检查是否到期
        if datetime.now() < subscription.current_period_end:
            return

        # 生成发票
        invoice = self.generate_invoice(subscription)

        # 尝试付款
        payment_result = self.charge_customer(
            subscription.customer_id,
            invoice.total
        )

        if payment_result.success:
            # 付款成功
            invoice.mark_paid()
            subscription.advance_billing_period()
            self.send_invoice(invoice)
        else:
            # 付款失败
            subscription.mark_past_due()
            self.start_dunning_process(subscription, invoice)

    def generate_invoice(self, subscription):
        """为计费周期生成发票。"""
        invoice = Invoice(
            customer_id=subscription.customer_id,
            subscription_id=subscription.id,
            period_start=subscription.current_period_start,
            period_end=subscription.current_period_end
        )

        # 添加订阅行项目
        invoice.add_line_item(
            description=subscription.plan.name,
            amount=subscription.plan.amount,
            quantity=subscription.quantity or 1
        )

        # 如果适用，添加基于用量的费用
        if subscription.has_usage_billing:
            usage_charges = self.calculate_usage_charges(subscription)
            invoice.add_line_item(
                description="用量费用",
                amount=usage_charges
            )

        # 计算税
        tax = self.calculate_tax(invoice.subtotal, subscription.customer)
        invoice.tax = tax

        invoice.finalize()
        return invoice

    def charge_customer(self, customer_id, amount):
        """使用保存的付款方式向客户收费。"""
        customer = self.get_customer(customer_id)

        try:
            # 使用支付处理器收费
            charge = stripe.Charge.create(
                customer=customer.stripe_id,
                amount=int(amount * 100),  # 转换为分
                currency='usd'
            )

            return PaymentResult(success=True, transaction_id=charge.id)
        except stripe.error.CardError as e:
            return PaymentResult(success=False, error=str(e))

催款管理

class DunningManager:
    """管理失败付款恢复。"""

    def __init__(self):
        self.retry_schedule = [
            {'days': 3, 'email_template': 'payment_failed_first'},
            {'days': 7, 'email_template': 'payment_failed_reminder'},
            {'days': 14, 'email_template': 'payment_failed_final'}
        ]

    def start_dunning_process(self, subscription, invoice):
        """为失败付款启动催款过程。"""
        dunning_attempt = DunningAttempt(
            subscription_id=subscription.id,
            invoice_id=invoice.id,
            attempt_number=1,
            next_retry=datetime.now() + timedelta(days=3)
        )

        # 发送初始失败通知
        self.send_dunning_email(subscription, 'payment_failed_first')

        # 计划重试
        self.schedule_retries(dunning_attempt)

    def retry_payment(self, dunning_attempt):
        """重试失败付款。"""
        subscription = self.get_subscription(dunning_attempt.subscription_id)
        invoice = self.get_invoice(dunning_attempt.invoice_id)

        # 再次尝试付款
        result = self.charge_customer(subscription.customer_id, invoice.total)

        if result.success:
            # 付款成功
            invoice.mark_paid()
            subscription.status = SubscriptionStatus.ACTIVE
            self.send_dunning_email(subscription, 'payment_recovered')
            dunning_attempt.mark_resolved()
        else:
            # 仍然失败
            dunning_attempt.attempt_number += 1

            if dunning_attempt.attempt_number < len(self.retry_schedule):
                # 计划下次重试
                next_retry_config = self.retry_schedule[dunning_attempt.attempt_number]
                dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days'])
                self.send_dunning_email(subscription, next_retry_config['email_template'])
            else:
                # 重试用尽，取消订阅
                subscription.cancel(at_period_end=False)
                self.send_dunning_email(subscription, 'subscription_canceled')

    def send_dunning_email(self, subscription, template):
        """向客户发送催款通知。"""
        customer = self.get_customer(subscription.customer_id)

        email_content = self.render_template(template, {
            'customer_name': customer.name,
            'amount_due': subscription.plan.amount,
            'update_payment_url': f"https://app.example.com/billing"
        })

        send_email(
            to=customer.email,
            subject=email_content['subject'],
            body=email_content['body']
        )

按比例计算

class ProrationCalculator:
    """计算计划更改的按比例费用。"""

    @staticmethod
    def calculate_proration(old_plan, new_plan, period_start, period_end, change_date):
        """计算计划更改的按比例。"""
        # 当前周期的天数
        total_days = (period_end - period_start).days

        # 使用旧计划的天数
        days_used = (change_date - period_start).days

        # 剩余新计划的天数
        days_remaining = (period_end - change_date).days

        # 计算按比例金额
        unused_amount = (old_plan.amount / total_days) * days_remaining
        new_plan_amount = (new_plan.amount / total_days) * days_remaining

        # 净费用/信用
        proration = new_plan_amount - unused_amount

        return {
            'old_plan_credit': -unused_amount,
            'new_plan_charge': new_plan_amount,
            'net_proration': proration,
            'days_used': days_used,
            'days_remaining': days_remaining
        }

    @staticmethod
    def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date):
        """计算席位更改的按比例。"""
        total_days = (period_end - period_start).days
        days_remaining = (period_end - change_date).days

        # 额外席位费用
        additional_seats = new_seats - current_seats
        prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining

        return {
            'additional_seats': additional_seats,
            'prorated_charge': max(0, prorated_amount),  # 移除席位不退款
            'effective_date': change_date
        }

税务计算

class TaxCalculator:
    """计算销售税、增值税、GST。"""

    def __init__(self):
        # 按地区的税率
        self.tax_rates = {
            'US_CA': 0.0725,  # 加利福尼亚销售税
            'US_NY': 0.04,    # 纽约销售税
            'GB': 0.20,       # 英国增值税
            'DE': 0.19,       # 德国增值税
            'FR': 0.20,       # 法国增值税
            'AU': 0.10,       # 澳大利亚GST
        }

    def calculate_tax(self, amount, customer):
        """计算适用税。"""
        # 确定税收管辖区
        jurisdiction = self.get_tax_jurisdiction(customer)

        if not jurisdiction:
            return 0

        # 获取税率
        tax_rate = self.tax_rates.get(jurisdiction, 0)

        # 计算税
        tax = amount * tax_rate

        return {
            'tax_amount': tax,
            'tax_rate': tax_rate,
            'jurisdiction': jurisdiction,
            'tax_type': self.get_tax_type(jurisdiction)
        }

    def get_tax_jurisdiction(self, customer):
        """基于客户位置确定税收管辖区。"""
        if customer.country == 'US':
            # 美国：基于客户州的税
            return f"US_{customer.state}"
        elif customer.country in ['GB', 'DE', 'FR']:
            # 欧盟：增值税
            return customer.country
        elif customer.country == 'AU':
            # 澳大利亚：GST
            return 'AU'
        else:
            return None

    def get_tax_type(self, jurisdiction):
        """获取管辖区的税类型。"""
        if jurisdiction.startswith('US_'):
            return '销售税'
        elif jurisdiction in ['GB', 'DE', 'FR']:
            return '增值税'
        elif jurisdiction == 'AU':
            return 'GST'
        return '税'

    def validate_vat_number(self, vat_number, country):
        """验证欧盟增值税号码。"""
        # 使用VIES API进行验证
        # 如果有效返回True，否则False
        pass

发票生成

class Invoice:
    def __init__(self, customer_id, subscription_id=None):
        self.id = generate_invoice_number()
        self.customer_id = customer_id
        self.subscription_id = subscription_id
        self.status = 'draft'
        self.line_items = []
        self.subtotal = 0
        self.tax = 0
        self.total = 0
        self.created_at = datetime.now()

    def add_line_item(self, description, amount, quantity=1):
        """向发票添加行项目。"""
        line_item = {
            'description': description,
            'unit_amount': amount,
            'quantity': quantity,
            'total': amount * quantity
        }
        self.line_items.append(line_item)
        self.subtotal += line_item['total']

    def finalize(self):
        """最终化发票并计算总金额。"""
        self.total = self.subtotal + self.tax
        self.status = 'open'
        self.finalized_at = datetime.now()

    def mark_paid(self):
        """标记发票为已支付。"""
        self.status = 'paid'
        self.paid_at = datetime.now()

    def to_pdf(self):
        """生成PDF发票。"""
        from reportlab.pdfgen import canvas

        # 生成PDF
        # 包括：公司信息、客户信息、行项目、税、总计
        pass

    def to_html(self):
        """生成HTML发票。"""
        template = """
        <!DOCTYPE html>
        <html>
        <head><title>发票 #{invoice_number}</title></head>
        <body>
            <h1>发票 #{invoice_number}</h1>
            <p>日期: {date}</p>
            <h2>账单接收人:</h2>
            <p>{customer_name}<br>{customer_address}</p>
            <table>
                <tr><th>描述</th><th>数量</th><th>金额</th></tr>
                {line_items}
            </table>
            <p>小计: ${subtotal}</p>
            <p>税: ${tax}</p>
            <h3>总计: ${total}</h3>
        </body>
        </html>
        """

        return template.format(
            invoice_number=self.id,
            date=self.created_at.strftime('%Y-%m-%d'),
            customer_name=self.customer.name,
            customer_address=self.customer.address,
            line_items=self.render_line_items(),
            subtotal=self.subtotal,
            tax=self.tax,
            total=self.total
        )

基于用量的计费

class UsageBillingEngine:
    """跟踪和计费用量。"""

    def track_usage(self, customer_id, metric, quantity):
        """跟踪用量事件。"""
        UsageRecord.create(
            customer_id=customer_id,
            metric=metric,
            quantity=quantity,
            timestamp=datetime.now()
        )

    def calculate_usage_charges(self, subscription, period_start, period_end):
        """计算计费周期的用量费用。"""
        usage_records = UsageRecord.get_for_period(
            subscription.customer_id,
            period_start,
            period_end
        )

        total_usage = sum(record.quantity for record in usage_records)

        # 分级定价
        if subscription.plan.pricing_model == 'tiered':
            charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers)
        # 每单位定价
        elif subscription.plan.pricing_model == 'per_unit':
            charge = total_usage * subscription.plan.unit_price
        # 批量定价
        elif subscription.plan.pricing_model == 'volume':
            charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers)

        return charge

    def calculate_tiered_pricing(self, total_usage, tiers):
        """使用分级定价计算成本。"""
        charge = 0
        remaining = total_usage

        for tier in sorted(tiers, key=lambda x: x['up_to']):
            tier_usage = min(remaining, tier['up_to'] - tier['from'])
            charge += tier_usage * tier['unit_price']
            remaining -= tier_usage

            if remaining <= 0:
                break

        return charge

资源

references/billing-cycles.md: 计费周期管理
references/dunning-management.md: 失败付款恢复
references/proration.md: 按比例费用计算
references/tax-calculation.md: 税务/VAT/GST处理
references/invoice-lifecycle.md: 发票状态管理
assets/billing-state-machine.yaml: 计费工作流
assets/invoice-template.html: 发票模板
assets/dunning-policy.yaml: 催款配置

最佳实践

自动化一切: 最小化手动干预
清晰沟通: 通知客户计费事件
灵活重试逻辑: 平衡恢复与客户体验
准确按比例计算: 公平计算计划更改
税务合规: 为管辖区计算正确税
审计跟踪: 记录所有计费事件
优雅降级: 处理边缘情况而不中断

常见陷阱

不正确的按比例计算: 未考虑部分周期
缺失税务: 忘记在发票中添加税
激进的催款: 过快取消
无通知: 未通知客户失败
硬编码周期: 不支持自定义计费日期

名称: 计费自动化 描述: 构建自动计费系统，用于循环付款、发票生成、订阅生命周期和催款管理。适用于实现订阅计费、自动化开票或管理循环付款系统。

计费自动化

何时使用此技能

核心概念

1. 计费周期

2. 订阅状态

3. 催款管理

4. 按比例计算

快速开始

订阅生命周期管理

计费周期处理

催款管理

按比例计算

税务计算

发票生成

基于用量的计费

资源

最佳实践

常见陷阱

名称: 计费自动化描述: 构建自动计费系统，用于循环付款、发票生成、订阅生命周期和催款管理。适用于实现订阅计费、自动化开票或管理循环付款系统。