mirror of https://github.com/zulip/zulip.git
296 lines
10 KiB
Python
296 lines
10 KiB
Python
from functools import partial
|
|
from html.parser import HTMLParser
|
|
from typing import Any, Callable, Dict, List, 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.webhooks.common import UnexpectedWebhookEventType, check_send_webhook_message
|
|
from zerver.models import UserProfile
|
|
|
|
COMPANY_CREATED = """
|
|
New company **{name}** created:
|
|
* **User count**: {user_count}
|
|
* **Monthly spending**: {monthly_spend}
|
|
""".strip()
|
|
|
|
CONTACT_EMAIL_ADDED = "New email {email} added to contact."
|
|
|
|
CONTACT_CREATED = """
|
|
New contact created:
|
|
* **Name (or pseudonym)**: {name}
|
|
* **Email**: {email}
|
|
* **Location**: {location_info}
|
|
""".strip()
|
|
|
|
CONTACT_SIGNED_UP = """
|
|
Contact signed up:
|
|
* **Email**: {email}
|
|
* **Location**: {location_info}
|
|
""".strip()
|
|
|
|
CONTACT_TAG_CREATED = "Contact tagged with the `{name}` tag."
|
|
|
|
CONTACT_TAG_DELETED = "The tag `{name}` was removed from the contact."
|
|
|
|
CONVERSATION_ADMIN_ASSIGNED = "{name} assigned to conversation."
|
|
|
|
CONVERSATION_ADMIN_TEMPLATE = "{admin_name} {action} the conversation."
|
|
|
|
CONVERSATION_ADMIN_REPLY_TEMPLATE = """
|
|
{admin_name} {action} the conversation:
|
|
|
|
``` quote
|
|
{content}
|
|
```
|
|
""".strip()
|
|
|
|
CONVERSATION_ADMIN_INITIATED_CONVERSATION = """
|
|
{admin_name} initiated a conversation:
|
|
|
|
``` quote
|
|
{content}
|
|
```
|
|
""".strip()
|
|
|
|
EVENT_CREATED = "New event **{event_name}** created."
|
|
|
|
USER_CREATED = """
|
|
New user created:
|
|
* **Name**: {name}
|
|
* **Email**: {email}
|
|
""".strip()
|
|
|
|
class MLStripper(HTMLParser):
|
|
def __init__(self) -> None:
|
|
self.reset()
|
|
self.strict = False
|
|
self.convert_charrefs = True
|
|
self.fed: List[str] = []
|
|
|
|
def handle_data(self, d: str) -> None:
|
|
self.fed.append(d)
|
|
|
|
def get_data(self) -> str:
|
|
return ''.join(self.fed)
|
|
|
|
def strip_tags(html: str) -> str:
|
|
s = MLStripper()
|
|
s.feed(html)
|
|
return s.get_data()
|
|
|
|
def get_topic_for_contacts(user: Dict[str, Any]) -> str:
|
|
topic = "{type}: {name}".format(
|
|
type=user['type'].capitalize(),
|
|
name=user.get('name') or user.get('pseudonym') or user.get('email'),
|
|
)
|
|
|
|
return topic
|
|
|
|
def get_company_created_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
body = COMPANY_CREATED.format(**payload['data']['item'])
|
|
return ('Companies', body)
|
|
|
|
def get_contact_added_email_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']
|
|
body = CONTACT_EMAIL_ADDED.format(email=user['email'])
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_contact_created_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
contact = payload['data']['item']
|
|
body = CONTACT_CREATED.format(
|
|
name=contact.get('name') or contact.get('pseudonym'),
|
|
email=contact['email'],
|
|
location_info="{city_name}, {region_name}, {country_name}".format(
|
|
**contact['location_data'],
|
|
),
|
|
)
|
|
topic = get_topic_for_contacts(contact)
|
|
return (topic, body)
|
|
|
|
def get_contact_signed_up_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
contact = payload['data']['item']
|
|
body = CONTACT_SIGNED_UP.format(
|
|
email=contact['email'],
|
|
location_info="{city_name}, {region_name}, {country_name}".format(
|
|
**contact['location_data'],
|
|
),
|
|
)
|
|
topic = get_topic_for_contacts(contact)
|
|
return (topic, body)
|
|
|
|
def get_contact_tag_created_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
body = CONTACT_TAG_CREATED.format(**payload['data']['item']['tag'])
|
|
contact = payload['data']['item']['contact']
|
|
topic = get_topic_for_contacts(contact)
|
|
return (topic, body)
|
|
|
|
def get_contact_tag_deleted_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
body = CONTACT_TAG_DELETED.format(**payload['data']['item']['tag'])
|
|
contact = payload['data']['item']['contact']
|
|
topic = get_topic_for_contacts(contact)
|
|
return (topic, body)
|
|
|
|
def get_conversation_admin_assigned_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
body = CONVERSATION_ADMIN_ASSIGNED.format(**payload['data']['item']['assignee'])
|
|
user = payload['data']['item']['user']
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_conversation_admin_message(
|
|
payload: Dict[str, Any],
|
|
action: str,
|
|
) -> Tuple[str, str]:
|
|
assignee = payload['data']['item']['assignee']
|
|
user = payload['data']['item']['user']
|
|
body = CONVERSATION_ADMIN_TEMPLATE.format(
|
|
admin_name=assignee.get('name'),
|
|
action=action,
|
|
)
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_conversation_admin_reply_message(
|
|
payload: Dict[str, Any],
|
|
action: str,
|
|
) -> Tuple[str, str]:
|
|
assignee = payload['data']['item']['assignee']
|
|
user = payload['data']['item']['user']
|
|
note = payload['data']['item']['conversation_parts']['conversation_parts'][0]
|
|
content = strip_tags(note['body'])
|
|
body = CONVERSATION_ADMIN_REPLY_TEMPLATE.format(
|
|
admin_name=assignee.get('name'),
|
|
action=action,
|
|
content=content,
|
|
)
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_conversation_admin_single_created_message(
|
|
payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
assignee = payload['data']['item']['assignee']
|
|
user = payload['data']['item']['user']
|
|
conversation_body = payload['data']['item']['conversation_message']['body']
|
|
content = strip_tags(conversation_body)
|
|
body = CONVERSATION_ADMIN_INITIATED_CONVERSATION.format(
|
|
admin_name=assignee.get('name'),
|
|
content=content,
|
|
)
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_conversation_user_created_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']['user']
|
|
conversation_body = payload['data']['item']['conversation_message']['body']
|
|
content = strip_tags(conversation_body)
|
|
body = CONVERSATION_ADMIN_INITIATED_CONVERSATION.format(
|
|
admin_name=user.get('name'),
|
|
content=content,
|
|
)
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_conversation_user_replied_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']['user']
|
|
note = payload['data']['item']['conversation_parts']['conversation_parts'][0]
|
|
content = strip_tags(note['body'])
|
|
body = CONVERSATION_ADMIN_REPLY_TEMPLATE.format(
|
|
admin_name=user.get('name'),
|
|
action='replied to',
|
|
content=content,
|
|
)
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_event_created_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
event = payload['data']['item']
|
|
body = EVENT_CREATED.format(**event)
|
|
return ('Events', body)
|
|
|
|
def get_user_created_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']
|
|
body = USER_CREATED.format(**user)
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_user_deleted_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, 'User deleted.')
|
|
|
|
def get_user_email_updated_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']
|
|
body = 'User\'s email was updated to {}.'.format(user['email'])
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
def get_user_tagged_message(
|
|
payload: Dict[str, Any],
|
|
action: str,
|
|
) -> Tuple[str, str]:
|
|
user = payload['data']['item']['user']
|
|
tag = payload['data']['item']['tag']
|
|
topic = get_topic_for_contacts(user)
|
|
body = 'The tag `{tag_name}` was {action} the user.'.format(
|
|
tag_name=tag['name'],
|
|
action=action,
|
|
)
|
|
return (topic, body)
|
|
|
|
def get_user_unsubscribed_message(payload: Dict[str, Any]) -> Tuple[str, str]:
|
|
user = payload['data']['item']
|
|
body = 'User unsubscribed from emails.'
|
|
topic = get_topic_for_contacts(user)
|
|
return (topic, body)
|
|
|
|
EVENT_TO_FUNCTION_MAPPER = {
|
|
'company.created': get_company_created_message,
|
|
'contact.added_email': get_contact_added_email_message,
|
|
'contact.created': get_contact_created_message,
|
|
'contact.signed_up': get_contact_signed_up_message,
|
|
'contact.tag.created': get_contact_tag_created_message,
|
|
'contact.tag.deleted': get_contact_tag_deleted_message,
|
|
'conversation.admin.assigned': get_conversation_admin_assigned_message,
|
|
'conversation.admin.closed': partial(get_conversation_admin_message, action='closed'),
|
|
'conversation.admin.opened': partial(get_conversation_admin_message, action='opened'),
|
|
'conversation.admin.snoozed': partial(get_conversation_admin_message, action='snoozed'),
|
|
'conversation.admin.unsnoozed': partial(get_conversation_admin_message, action='unsnoozed'),
|
|
'conversation.admin.replied': partial(get_conversation_admin_reply_message, action='replied to'),
|
|
'conversation.admin.noted': partial(get_conversation_admin_reply_message, action='added a note to'),
|
|
'conversation.admin.single.created': get_conversation_admin_single_created_message,
|
|
'conversation.user.created': get_conversation_user_created_message,
|
|
'conversation.user.replied': get_conversation_user_replied_message,
|
|
'event.created': get_event_created_message,
|
|
'user.created': get_user_created_message,
|
|
'user.deleted': get_user_deleted_message,
|
|
'user.email.updated': get_user_email_updated_message,
|
|
'user.tag.created': partial(get_user_tagged_message, action='added to'),
|
|
'user.tag.deleted': partial(get_user_tagged_message, action='removed from'),
|
|
'user.unsubscribed': get_user_unsubscribed_message,
|
|
# Note that we do not have a payload for visitor.signed_up
|
|
# but it should be identical to contact.signed_up
|
|
'visitor.signed_up': get_contact_signed_up_message,
|
|
}
|
|
|
|
def get_event_handler(event_type: str) -> Callable[..., Tuple[str, str]]:
|
|
handler: Any = EVENT_TO_FUNCTION_MAPPER.get(event_type)
|
|
if handler is None:
|
|
raise UnexpectedWebhookEventType("Intercom", event_type)
|
|
return handler
|
|
|
|
@api_key_only_webhook_view('Intercom')
|
|
@has_request_variables
|
|
def api_intercom_webhook(request: HttpRequest, user_profile: UserProfile,
|
|
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
|
|
event_type = payload['topic']
|
|
if event_type == 'ping':
|
|
return json_success()
|
|
|
|
topic, body = get_event_handler(event_type)(payload)
|
|
|
|
check_send_webhook_message(request, user_profile, topic, body)
|
|
return json_success()
|