# Webhooks for external integrations. import time from typing import Any, Dict, Optional, Sequence, Tuple from django.http import HttpRequest, HttpResponse from zerver.decorator import api_key_only_webhook_view from zerver.lib.request import REQ, has_request_variables from zerver.lib.response import json_success from zerver.lib.timestamp import timestamp_to_datetime from zerver.lib.webhooks.common import UnexpectedWebhookEventType, check_send_webhook_message from zerver.models import UserProfile class SuppressedEvent(Exception): pass class NotImplementedEventType(SuppressedEvent): pass @api_key_only_webhook_view('Stripe') @has_request_variables def api_stripe_webhook(request: HttpRequest, user_profile: UserProfile, payload: Dict[str, Any]=REQ(argument_type='body'), stream: str=REQ(default='test')) -> HttpResponse: try: topic, body = topic_and_body(payload) except SuppressedEvent: # nocoverage return json_success() check_send_webhook_message(request, user_profile, topic, body) return json_success() def topic_and_body(payload: Dict[str, Any]) -> Tuple[str, str]: event_type = payload["type"] # invoice.created, customer.subscription.created, etc if len(event_type.split('.')) == 3: category, resource, event = event_type.split('.') else: resource, event = event_type.split('.') category = resource object_ = payload["data"]["object"] # The full, updated Stripe object # Set the topic to the customer_id when we can topic = '' customer_id = object_.get("customer", None) if customer_id is not None: # Running into the 60 character topic limit. # topic = '[{}](https://dashboard.stripe.com/customers/{})' % (customer_id, customer_id) topic = customer_id body = None def update_string(blacklist: Sequence[str] = []) -> str: assert('previous_attributes' in payload['data']) previous_attributes = payload['data']['previous_attributes'] for attribute in blacklist: previous_attributes.pop(attribute, None) if not previous_attributes: # nocoverage raise SuppressedEvent() return ''.join('\n* ' + attribute.replace('_', ' ').capitalize() + ' is now ' + stringify(object_[attribute]) for attribute in sorted(previous_attributes.keys())) def default_body(update_blacklist: Sequence[str] = []) -> str: body = '{resource} {verbed}'.format( resource=linkified_id(object_['id']), verbed=event.replace('_', ' ')) if event == 'updated': return body + update_string(blacklist=update_blacklist) return body if category == 'account': # nocoverage if resource == 'account': if event == 'updated': if 'previous_attributes' not in payload['data']: raise SuppressedEvent() topic = "account updates" body = update_string() else: # Part of Stripe Connect raise NotImplementedEventType() if category == 'application_fee': # nocoverage # Part of Stripe Connect raise NotImplementedEventType() if category == 'balance': # nocoverage # Not that interesting to most businesses, I think raise NotImplementedEventType() if category == 'charge': if resource == 'charge': if not topic: # only in legacy fixtures topic = 'charges' body = "{resource} for {amount} {verbed}".format( resource=linkified_id(object_['id']), amount=amount_string(object_['amount'], object_['currency']), verbed=event) if object_['failure_code']: # nocoverage body += '. Failure code: {}'.format(object_['failure_code']) if resource == 'dispute': topic = 'disputes' body = default_body() + '. Current status: {status}.'.format( status=object_['status'].replace('_', ' ')) if resource == 'refund': topic = 'refunds' body = 'A {resource} for a {charge} of {amount} {currency} was updated.'.format( resource=linkified_id(object_['id'], lower=True), charge=linkified_id(object_['charge'], lower=True), amount=object_['amount'], currency=object_['currency'].upper(), ) if category == 'checkout_beta': # nocoverage # Not sure what this is raise NotImplementedEventType() if category == 'coupon': # nocoverage # Not something that likely happens programmatically raise NotImplementedEventType() if category == 'customer': if resource == 'customer': # Running into the 60 character topic limit. # topic = '[{}](https://dashboard.stripe.com/customers/{})' % (object_['id'], object_['id']) topic = object_['id'] body = default_body(update_blacklist=['delinquent', 'currency', 'default_source']) if event == 'created': if object_['email']: body += '\nEmail: {}'.format(object_['email']) if object_['metadata']: # nocoverage for key, value in object_['metadata'].items(): body += f'\n{key}: {value}' if resource == 'discount': body = 'Discount {verbed} ([{coupon_name}]({coupon_url})).'.format( verbed=event.replace('_', ' '), coupon_name=object_['coupon']['name'], coupon_url='https://dashboard.stripe.com/{}/{}'.format('coupons', object_['coupon']['id']), ) if resource == 'source': # nocoverage body = default_body() if resource == 'subscription': body = default_body() if event == 'trial_will_end': DAY = 60 * 60 * 24 # seconds in a day # Basically always three: https://stripe.com/docs/api/python#event_types body += ' in {days} days'.format( days=int((object_["trial_end"] - time.time() + DAY//2) // DAY)) if event == 'created': if object_['plan']: body += '\nPlan: [{plan_nickname}](https://dashboard.stripe.com/plans/{plan_id})'.format( plan_nickname=object_['plan']['nickname'], plan_id=object_['plan']['id']) if object_['quantity']: body += '\nQuantity: {}'.format(object_['quantity']) if 'billing' in object_: # nocoverage body += '\nBilling method: {}'.format(object_['billing'].replace('_', ' ')) if category == 'file': # nocoverage topic = 'files' body = default_body() + ' ({purpose}). \nTitle: {title}'.format( purpose=object_['purpose'].replace('_', ' '), title=object_['title']) if category == 'invoice': if event == 'upcoming': # nocoverage body = 'Upcoming invoice created' elif (event == 'updated' and payload['data']['previous_attributes'].get('paid', None) is False and object_['paid'] is True and object_["amount_paid"] != 0 and object_["amount_remaining"] == 0): # We are taking advantage of logical AND short circuiting here since we need the else # statement below. object_id = object_['id'] invoice_link = f'https://dashboard.stripe.com/invoices/{object_id}' body = f'[Invoice]({invoice_link}) is now paid' else: body = default_body(update_blacklist=['lines', 'description', 'number', 'finalized_at', 'status_transitions', 'payment_intent']) if event == 'created': # Could potentially add link to invoice PDF here body += ' ({reason})\nTotal: {total}\nAmount due: {due}'.format( reason=object_['billing_reason'].replace('_', ' '), total=amount_string(object_['total'], object_['currency']), due=amount_string(object_['amount_due'], object_['currency'])) if category == 'invoiceitem': body = default_body(update_blacklist=['description', 'invoice']) if event == 'created': body += ' for {amount}'.format(amount=amount_string(object_['amount'], object_['currency'])) if category.startswith('issuing'): # nocoverage # Not implemented raise NotImplementedEventType() if category.startswith('order'): # nocoverage # Not implemented raise NotImplementedEventType() if category in ['payment_intent', 'payout', 'plan', 'product', 'recipient', 'reporting', 'review', 'sigma', 'sku', 'source', 'subscription_schedule', 'topup', 'transfer']: # nocoverage # Not implemented. In theory doing something like # body = default_body() # may not be hard for some of these raise NotImplementedEventType() if body is None: raise UnexpectedWebhookEventType('Stripe', event_type) return (topic, body) def amount_string(amount: int, currency: str) -> str: zero_decimal_currencies = ["bif", "djf", "jpy", "krw", "pyg", "vnd", "xaf", "xpf", "clp", "gnf", "kmf", "mga", "rwf", "vuv", "xof"] if currency in zero_decimal_currencies: decimal_amount = str(amount) # nocoverage else: decimal_amount = f'{float(amount) * 0.01:.02f}' if currency == 'usd': # nocoverage return '$' + decimal_amount return decimal_amount + f' {currency.upper()}' def linkified_id(object_id: str, lower: bool=False) -> str: names_and_urls: Dict[str, Tuple[str, Optional[str]]] = { # Core resources 'ch': ('Charge', 'charges'), 'cus': ('Customer', 'customers'), 'dp': ('Dispute', 'disputes'), 'file': ('File', 'files'), 'link': ('File link', 'file_links'), 'pi': ('Payment intent', 'payment_intents'), 'po': ('Payout', 'payouts'), 'prod': ('Product', 'products'), 're': ('Refund', 'refunds'), 'tok': ('Token', 'tokens'), # Payment methods # payment methods have URL prefixes like /customers/cus_id/sources 'ba': ('Bank account', None), 'card': ('Card', None), 'src': ('Source', None), # Billing # coupons have a configurable id, but the URL prefix is /coupons # discounts don't have a URL, I think 'in': ('Invoice', 'invoices'), 'ii': ('Invoice item', 'invoiceitems'), # products are covered in core resources # plans have a configurable id, though by default they are created with this pattern # 'plan': ('Plan', 'plans'), 'sub': ('Subscription', 'subscriptions'), 'si': ('Subscription item', 'subscription_items'), # I think usage records have URL prefixes like /subscription_items/si_id/usage_record_summaries 'mbur': ('Usage record', None), # Undocumented :| 'py': ('Payment', 'payments'), 'pyr': ('Refund', 'refunds'), # Psuedo refunds. Not fully tested. # Connect, Fraud, Orders, etc not implemented } name, url_prefix = names_and_urls[object_id.split('_')[0]] if lower: # nocoverage name = name.lower() if url_prefix is None: # nocoverage return name return f'[{name}](https://dashboard.stripe.com/{url_prefix}/{object_id})' def stringify(value: Any) -> str: if isinstance(value, int) and value > 1500000000 and value < 2000000000: return timestamp_to_datetime(value).strftime('%b %d, %Y, %H:%M:%S %Z') return str(value)