# Webhooks for external integrations. from typing import Dict, Any, List 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 check_send_webhook_message from zerver.models import UserProfile subject_types = { 'app': [ # Object type name ['name'], # Title ['html_url'], # Automatically put into title ['language'], # Other properties. ['framework'] ], 'base': [ ['title'], ['html_url'], ['#summary'], ['subject'] ], 'comment': [ [''], ['subject'] ], 'errorgroup': [ ['E#{}', 'number'], ['html_url'], ['last_occurrence:error'] ], 'error': [ [''], ['">**Most recent Occurrence**'], ['in {}', 'extra/pathname'], ['!message'] ] } # type: Dict[str, List[List[str]]] def get_value(_obj: Dict[str, Any], key: str) -> str: for _key in key.lstrip('!').split('/'): if _key in _obj.keys(): _obj = _obj[_key] else: return '' return str(_obj) def format_object( obj: Dict[str, Any], subject_type: str, message: str ) -> str: if subject_type not in subject_types.keys(): return message keys = subject_types[subject_type][1:] # type: List[List[str]] title = subject_types[subject_type][0] if title[0] != '': title_str = '' if len(title) > 1: title_str = title[0].format(get_value(obj, title[1])) else: title_str = obj[title[0]] if obj['html_url'] is not None: url = obj['html_url'] # type: str if 'opbeat.com' not in url: url = 'https://opbeat.com/' + url.lstrip('/') message += '\n**[{}]({})**'.format(title_str, url) else: message += '\n**{}**'.format(title_str) for key_list in keys: if len(key_list) > 1: value = key_list[0].format(get_value(obj, key_list[1])) message += '\n>{}'.format(value) else: key = key_list[0] key_raw = key.lstrip('!').lstrip('#').lstrip('"') if key_raw != 'html_url' and key_raw != 'subject' and ':' not in key_raw: value = get_value(obj, key_raw) if key.startswith('!'): message += '\n>{}'.format(value) elif key.startswith('#'): message += '\n{}'.format(value) elif key.startswith('"'): message += '\n{}'.format(key_raw) else: message += '\n>{}: {}'.format(key, value) if key == 'subject': message = format_object( obj['subject'], obj['subject_type'], message + '\n') if ':' in key: value, value_type = key.split(':') message = format_object(obj[value], value_type, message + '\n') return message @api_key_only_webhook_view("Opbeat") @has_request_variables def api_opbeat_webhook(request: HttpRequest, user_profile: UserProfile, payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse: """ This uses the subject name from opbeat to make the subject, and the summary from Opbeat as the message body, with details about the object mentioned. """ message_subject = payload['title'] message = format_object(payload, 'base', '') check_send_webhook_message(request, user_profile, message_subject, message) return json_success()