# By: Riasat Ullah # This class allocates notices to the correct type of dispatches and does the dispatches # for sms and voice alerts. It helps you decide how to send the instance alert notices. # This should not be used for determining whether to dispatch or escalate. Use the EventAllocator # class to do that. from notices import business_impact_notices, incident_notices, status_page_notices from objects.alert_log import AlertLog from threading import Thread from translators import label_translator as _lt from utils import constants, info, logging, times, var_names class NoticeAllocator(object): def __init__(self, user_info=None, taskcall_numbers=None, vendor_api_clients=None, policies=None, services=None): self.user_info = user_info self.taskcall_numbers = taskcall_numbers self.vendor_api_clients = vendor_api_clients self.policies = policies self.services = services self.email_messages = [] self.push_notices = [] self.alert_logs = [] def handle_bulk_email_dispatch(self, subject, body, policy_ids, instance_id=None, organization_id=None): ''' Adds in the logs of the email dispatch event in the list of logs to be inserted in the database. :param subject: the subject of the email :param body: the body of the email :param policy_ids: (list of int) policy id of the assignee who the email is to be sent to :param instance_id: the instance ID :param organization_id: (optional) organization ID of the user who the email is being sent to ''' to_addr = [self.user_info[x][var_names.email] for x in policy_ids] user_ids = [self.user_info[x][var_names.user_id] for x in policy_ids] if len(to_addr) > 0: self.handle_email_dispatch(subject, body, to_addr, user_ids, instance_id=instance_id, organization_id=organization_id) def handle_email_dispatch(self, subject, body, to_addr, user_ids, instance_id=None, organization_id=None): ''' Adds in the logs of the email dispatch event in the list of logs to be inserted in the database. :param subject: the subject of the email :param body: the body of the email :param to_addr: (str or list of str) email addresses to send the email to :param user_ids: (list of int) ID of the user who the email is being sent to :param instance_id: the instance ID :param organization_id: (optional) organization ID of the user who the email is being sent to ''' self.email_messages.append({ var_names.email_subject: subject, var_names.email_body: body, var_names.email_to: to_addr }) for u_id in user_ids: log = AlertLog(times.get_current_timestamp(), side=constants.outgoing_side, event_method=constants.email, event_type=constants.dispatch_event, organization_id=organization_id, user_id=u_id, instance_id=instance_id, message=subject) self.alert_logs.append(log) def handle_app_dispatch(self, push_title, push_message, policy_id=None, tokens=None, instance_id=None, user_id=None, organization_id=None, is_critical=False): ''' Puts the push notifications in a bucket to be sent after all the instances are traversed. :param push_title: the title of the push notification :param push_message: the message body of the push notification :param policy_id: policy id of the assignee who the dispatch is being made to :param tokens: (list) of push tokens to send the message to :param instance_id: (optional) ID of the instance this push notification is for :param user_id: (int) ID of the user who the push notification is being sent to :param organization_id: (optional) organization ID of the user who the push message is being sent to :param is_critical: (boolean) True if the push notification should be sent as a critical alert ''' if user_id is None: user_id = self.user_info[policy_id][var_names.user_id] if tokens is None: tokens = self.user_info[policy_id][var_names.push_token] for token_ in tokens: if token_ is not None and token_ != 'null': self.push_notices.append({ var_names.push_token: token_, var_names.notice_title: push_title, var_names.message: push_message, var_names.is_critical: is_critical, var_names.extra_data: { var_names.title: push_title, var_names.body: push_message } }) log = AlertLog(times.get_current_timestamp(), side=constants.outgoing_side, event_method=constants.app, event_type=constants.dispatch_event, organization_id=organization_id, user_id=user_id, instance_id=instance_id, message=push_message) self.alert_logs.append(log) def handle_text_dispatch(self, policy_id: int, phone_message, instance_id=None, organization_id=None): ''' Sends a text message alert. :param policy_id: user policy ID :param phone_message: the text message to be sent out :param instance_id: (optional) ID of the instance this text alert is related to :param organization_id: (optional) organization ID of the user who the text message is being sent to ''' policy_details = self.user_info[policy_id] user_phone = policy_details[var_names.phone] user_iso = policy_details[var_names.iso_country_code] if (user_iso, var_names.text_allowed) in self.taskcall_numbers: vendor_details = self.taskcall_numbers[(user_iso, var_names.text_allowed)] vendor_name = vendor_details[var_names.vendor_name] vendor_client = self.vendor_api_clients[vendor_name] vendor_phone = vendor_details[var_names.phone] logging.info('Sending text alert to policy id - ' + str(policy_id)) Thread(target=vendor_client.send_sms, args=(vendor_phone, user_phone, phone_message,)).start() log = AlertLog(times.get_current_timestamp(), side=constants.outgoing_side, event_method=constants.text, event_type=constants.dispatch_event, organization_id=organization_id, user_id=policy_details[var_names.user_id], sender_iso_code=user_iso, sender_phone_number=vendor_phone, recipient_iso_code=user_iso, recipient_phone_number=user_phone, instance_id=instance_id, message=phone_message, vendor=vendor_name) self.alert_logs.append(log) def handle_voice_dispatch(self, policy_id: int, phone_message, instance_id=None, language=constants.lang_en, organization_id=None): ''' Sends a voice message alert. :param policy_id: user policy ID :param phone_message: the text message to be sent out :param instance_id: (optional) ID of the instance this voice alert is related to :param language: (optional) the language to send the call content in :param organization_id: (optional) organization ID of the user who is being called ''' policy_details = self.user_info[policy_id] user_phone = policy_details[var_names.phone] user_iso = policy_details[var_names.iso_country_code] if (user_iso, var_names.call_allowed) in self.taskcall_numbers: vendor_details = self.taskcall_numbers[(user_iso, var_names.call_allowed)] vendor_name = vendor_details[var_names.vendor_name] vendor_client = self.vendor_api_clients[vendor_name] vendor_phone = vendor_details[var_names.phone] logging.info('Sending voice call alert to policy id - ' + str(policy_id)) Thread(target=vendor_client.call, args=(vendor_phone, user_phone, phone_message, language,)).start() log = AlertLog(times.get_current_timestamp(), side=constants.outgoing_side, event_method=constants.call, event_type=constants.dispatch_event, organization_id=organization_id, user_id=policy_details[var_names.user_id], sender_iso_code=user_iso, sender_phone_number=vendor_phone, recipient_iso_code=user_iso, recipient_phone_number=user_phone, instance_id=instance_id, message=phone_message, vendor=vendor_name) self.alert_logs.append(log) def handle_impacted_business_services_dispatch( self, content_language, business_service_name, event_type, impact_start, impact_end, incidents, timezone, email_addr=None, tokens=None, user_id=None, organization_id=None, update_after_time=None ): ''' Sends out the impacted business service notice by email and puts the push notification messages in the push_notices list of the class. If policy_ids are provided the function assumes that user_info has also been provided to the class. If policy_ids are not provided then the function tries to use email_addr and tokens to do the dispatches. :param content_language: the language to send the textual content in :param business_service_name: name of the impacted business service this particular message is being sent to :param event_type: adhoc way to identify the status of the business service (not related to instance event) :param impact_start: time when the impact started :param impact_end: time when the impact ended :param timezone: timezone the timestamps should be converted to :param incidents: (list) of incidents :param email_addr: (str or list of str) email addresses to email to :param tokens: (list) of push tokens :param user_id: ID of the user :param organization_id: (optional) organization ID of the user who the notifications are being sent to :param update_after_time: (timestamp) if provided only the status updates after the given time will be sent ''' subject, message = business_impact_notices.business_impact_update_email_content( content_language, business_service_name, event_type, impact_start, impact_end, timezone, incidents ) push_title, push_message = business_impact_notices.business_impact_update_app_content( content_language, business_service_name, event_type, incidents, update_after_time=update_after_time) self.handle_email_dispatch(subject, message, email_addr, user_ids=[user_id], organization_id=organization_id) if tokens is not None: self.handle_app_dispatch(push_title, push_message, user_id=user_id, tokens=tokens, organization_id=organization_id) def handle_status_update_dispatch(self, instance_id, org_instance_id, instance_time, instance_title, new_status, subscribers, organization_id=None, is_resolution_update=False): ''' Handle the dispatching of a status update event. :param instance_id: ID of the instance :param org_instance_id: organization instance ID :param instance_time: time when the instance was created on :param instance_title: title of the instance :param new_status: the new status update :param subscribers: (list of dict) -> [{user_id: , email: , subscribers: ,}, ...] :param organization_id: (optional) organization ID of the user who the update is being sent to :param is_resolution_update: True if this update is for incident resolution ''' sub_usr_ids = [] sub_emails = [] for item in subscribers: user_id, email, user_lang, ptk = item[var_names.user_id], item[var_names.email], \ item[var_names.language], item[var_names.push_token] if new_status is None: if is_resolution_update: new_status = _lt.get_label(info.msg_incident_resolved, user_lang) else: continue sub_usr_ids.append(user_id) sub_emails.append(email) subject, message = incident_notices.instance_status_update_email_content( user_lang, org_instance_id, instance_time, instance_title, new_status) push_title, push_message = incident_notices.instance_status_update_app_content( user_lang, org_instance_id, instance_title, new_status) if ptk is not None and len(ptk) > 0: self.handle_app_dispatch(push_title, push_message, user_id=user_id, tokens=ptk, instance_id=instance_id, organization_id=organization_id) self.handle_email_dispatch(subject, message, sub_emails, user_ids=sub_usr_ids, instance_id=instance_id, organization_id=organization_id) def handle_conference_bridge_dispatch(self, instance_id, org_instance_id, instance_title, conference_bridge, assignees, organization_id=None): ''' Handle the dispatching of conference bridge notifications. :param instance_id: ID of the instance :param org_instance_id: organization instance ID :param instance_title: title of the instance :param conference_bridge: (dict) -> {conference phone: ..., conference url: ...} :param assignees: (list of dict) -> [{user_id: , email: , subscribers: ,}, ...] :param organization_id: (optional) organization ID of the user who the update is being sent to ''' for item in assignees: user_id, email, user_lang, ptk = item[var_names.user_id], item[var_names.email], \ item[var_names.language], item[var_names.push_token] subject, message = incident_notices.conference_bridge_added_email_content( user_lang, org_instance_id, instance_title, conference_bridge) push_title, push_message = incident_notices.conference_bridge_added_app_content( user_lang, org_instance_id, instance_title, conference_bridge) self.handle_email_dispatch(subject, message, email, user_ids=[user_id], instance_id=instance_id, organization_id=organization_id) if ptk is not None and len(ptk) > 0: self.handle_app_dispatch(push_title, push_message, user_id=user_id, tokens=ptk, instance_id=instance_id, organization_id=organization_id) def handle_status_page_pending_post_dispatch(self, page_ref_id, page_name, post_title, post_message, event_type, event_ref_id, recipients): ''' Inform users who have the right to approve pending posts on a status page about a pending post. :param page_ref_id: (concealed) reference ID of the page :param page_name: name of the status page :param post_title: title of the post :param post_message: message of the post :param event_type: type of event (MAINTENANCE, NEW_INCIDENT, STATUS_UPDATE) :param event_ref_id: (concealed) reference ID of the pending event :param recipients: (list of dict) details of the recipients who the notifications will be sent to ''' for item in recipients: user_id, email, user_lang, ptk = item[var_names.user_id], item[var_names.email], \ item[var_names.language], item[var_names.push_token] subject, message = status_page_notices.status_page_pending_post_approval_email_content( user_lang, page_ref_id, page_name, post_title, post_message, event_type, event_ref_id) push_title, push_message = status_page_notices.status_page_pending_post_approval_notice( user_lang, page_name, post_title, post_message) self.handle_email_dispatch(subject, message, email, user_ids=[user_id]) if ptk is not None and len(ptk) > 0: self.handle_app_dispatch(push_title, push_message, user_id=user_id, tokens=ptk)