mirror of https://github.com/zulip/zulip.git
258 lines
12 KiB
Python
258 lines
12 KiB
Python
# 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)
|