# Webhooks for external integrations. from typing import Any, Dict, 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: Dict[str, List[List[str]]] = { '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'], ], } 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: List[List[str]] = subject_types[subject_type][1:] 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: str = obj['html_url'] if 'opbeat.com' not in url: url = 'https://opbeat.com/' + url.lstrip('/') message += f'\n**[{title_str}]({url})**' else: message += f'\n**{title_str}**' for key_list in keys: if len(key_list) > 1: value = key_list[0].format(get_value(obj, key_list[1])) message += f'\n>{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 += f'\n>{value}' elif key.startswith('#'): message += f'\n{value}' elif key.startswith('"'): message += f'\n{key_raw}' else: message += f'\n>{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()