SaaS 구독을 출시할 때 /payment-processing-billing-automation은 반복 청구, 결제 실패 회수, 비례 계산, 세금 처리를 구축합니다. — Claude Skill

청구 자동화

반복 청구, 송장 생성, 결제 실패 회수, 비례 계산, 세금 계산을 포함한 자동 청구 시스템을 다룹니다.

이 스킬을 사용할 때

• SaaS 구독 청구를 구현할 때
• 송장 생성과 발송을 자동화할 때
• 결제 실패 회수(dunning)를 관리할 때
• 요금제 변경에 따른 비례 요금을 계산할 때
• 판매세, VAT, GST를 처리할 때
• 사용량 기반 청구를 처리할 때
• 청구 주기와 갱신을 관리할 때

핵심 개념

1. 청구 주기

일반적인 간격:

• 월간(SaaS에서 가장 흔함)
• 연간(장기 할인)
• 분기별
• 주간
• 맞춤형(사용량 기반, 좌석당 과금)

2. 구독 상태

trial → active → past_due → canceled
              → paused → resumed

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):
        """결제 실패 후 구독을 past_due로 표시합니다."""
        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:
    """판매세, VAT, GST를 계산합니다."""

    def __init__(self):
        # 지역별 세율
        self.tax_rates = {
            'US_CA': 0.0725,  # 캘리포니아 판매세
            'US_NY': 0.04,    # 뉴욕 판매세
            'GB': 0.20,       # 영국 VAT
            'DE': 0.19,       # 독일 VAT
            'FR': 0.20,       # 프랑스 VAT
            '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']:
            # EU: VAT
            return customer.country
        elif customer.country == 'AU':
            # 호주: GST
            return 'AU'
        else:
            return None

    def get_tax_type(self, jurisdiction):
        """관할권의 세금 유형을 반환합니다."""
        if jurisdiction.startswith('US_'):
            return 'Sales Tax'
        elif jurisdiction in ['GB', 'DE', 'FR']:
            return 'VAT'
        elif jurisdiction == 'AU':
            return 'GST'
        return 'Tax'

    def validate_vat_number(self, vat_number, country):
        """EU VAT 번호를 검증합니다."""
        # 검증에는 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 #{invoice_number}</title></head>
        <body>
            <h1>Invoice #{invoice_number}</h1>
            <p>Date: {date}</p>
            <h2>Bill To:</h2>
            <p>{customer_name}<br>{customer_address}</p>
            <table>
                <tr><th>Description</th><th>Quantity</th><th>Amount</th></tr>
                {line_items}
            </table>
            <p>Subtotal: ${subtotal}</p>
            <p>Tax: ${tax}</p>
            <h3>Total: ${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