# By: Riasat Ullah # This file contains live call router related functions. from twilio.twiml.voice_response import Dial, Gather, VoiceResponse from utils.communication_vendors import Twilio from utils import constants, info, label_names as lnm, url_paths, var_names from translators import label_translator as _lt from validations import string_validator import configuration as configs import re class LiveCallRouting(object): def __init__(self, call_routing_id, organization_id, routing_name, iso_country_code, phone, phone_type, greeting_text, greeting_audio_filename, greeting_audio_url, ending_text, ending_audio_filename, ending_audio_url, default_forwarding_service, option_forwarding_services, text_language, is_male_voice, max_forwarding_users, forwarding_timeout, urgency_level, block_numbers, show_caller_id, resolve_answered_calls, resolve_unanswered_calls, to_alert, record_voicemail, record_call, call_routing_ref_id=None, routing_desc=None, caller=None, twilio_call_sid=None, policy_on_calls=None, on_call_phones=None, instance_id=None, call_status=None, forwarding_count=0, call_duration=None, machine_answered=False, prompt_call_acceptance=True, prompt_format=None, incident_title_format=None, caller_details=None): self.call_routing_id = call_routing_id self.organization_id = organization_id self.routing_name = routing_name self.iso_country_code = iso_country_code self.phone = phone self.phone_type = phone_type self.greeting_text = greeting_text self.greeting_audio_filename = greeting_audio_filename self.greeting_audio_url = greeting_audio_url self.ending_text = ending_text self.ending_audio_filename = ending_audio_filename self.ending_audio_url = ending_audio_url self.default_forwarding_service = default_forwarding_service self.option_forwarding_services = option_forwarding_services self.text_language = text_language self.is_male_voice = is_male_voice self.max_forwarding_users = max_forwarding_users self.forwarding_timeout = forwarding_timeout self.urgency_level = urgency_level self.block_numbers = block_numbers self.show_caller_id = show_caller_id self.resolve_answered_calls = resolve_answered_calls self.resolve_unanswered_calls = resolve_unanswered_calls self.to_alert = to_alert self.record_voicemail = record_voicemail self.record_call = record_call self.call_routing_ref_id = call_routing_ref_id self.routing_desc = routing_desc self.caller = caller self.twilio_call_sid = twilio_call_sid self.policy_on_calls = policy_on_calls self.on_call_phones = on_call_phones self.instance_id = instance_id self.call_status = call_status self.forwarding_count = forwarding_count if forwarding_count is not None else 0 self.call_duration = call_duration self.machine_answered = machine_answered self.prompt_call_acceptance = prompt_call_acceptance self.prompt_format = prompt_format self.incident_title_format = incident_title_format self.caller_details = caller_details if self.forwarding_timeout is None or self.forwarding_timeout > configs.call_routing_max_forwarding_timeout: self.forwarding_timeout = configs.call_routing_max_forwarding_timeout @staticmethod def create_live_call_routing(details): ''' Creates a LiveCallRouting object from a dict of live call routing details. :param details: (dict) of live call routing details :return: LiveCallRouting object ''' return LiveCallRouting( details[var_names.call_routing_id], details[var_names.organization_id], details[var_names.routing_name], details[var_names.iso_country_code], details[var_names.phone], details[var_names.phone_type] if var_names.phone_type in details else None, details[var_names.greeting_text], details[var_names.greeting_audio_filename], details[var_names.greeting_audio_url], details[var_names.ending_text], details[var_names.ending_audio_filename], details[var_names.ending_audio_url], details[var_names.default_forwarding_service], details[var_names.option_forwarding_services], details[var_names.text_language], details[var_names.is_male_voice], details[var_names.max_forwarding_users], details[var_names.forwarding_timeout], details[var_names.urgency_level], details[var_names.block_numbers], details[var_names.show_caller_id], details[var_names.resolve_answered_calls], details[var_names.resolve_unanswered_calls], details[var_names.to_alert], details[var_names.record_voicemail], details[var_names.record_call], details[var_names.call_routing_ref_id] if var_names.call_routing_ref_id in details else None, details[var_names.description] if var_names.description in details else None, details[var_names.from_number] if var_names.from_number in details else None, details[var_names.twilio_call_sid] if var_names.twilio_call_sid in details else None, details[var_names.on_call] if var_names.on_call in details else None, {int(key): details[var_names.on_call_phones][key] for key in details[var_names.on_call_phones]} if var_names.on_call_phones in details and details[var_names.on_call_phones] is not None else None, details[var_names.instance_id] if var_names.instance_id in details else None, details[var_names.call_status] if var_names.call_status in details else None, details[var_names.forwarding_count] if var_names.forwarding_count in details else None, details[var_names.call_duration] if var_names.call_duration in details else None, details[var_names.machine_answered] if var_names.machine_answered in details else None, details[var_names.prompt_call_acceptance] if var_names.prompt_call_acceptance in details else True, details[var_names.prompt_format] if var_names.prompt_format in details else None, details[var_names.incident_title_format] if var_names.incident_title_format in details else None, details[var_names.caller_details] if var_names.caller_details in details else None ) def get_voice_gender(self): ''' Get the Twilio standardized values for voice gender. :return: (str) man | woman ''' if self.is_male_voice: return Twilio.voice_man return Twilio.voice_woman def is_number_blocked(self, calling_number): ''' Checks if a number is blocked or not. :param calling_number: (str) the incoming caller phone number :return: (boolean) True if it is blocked; False otherwise ''' if self.block_numbers is not None and calling_number in self.block_numbers: return True return False def get_forwarding_index(self): if len(self.policy_on_calls) > 0: return self.forwarding_count % len(self.policy_on_calls) return 0 def has_next_on_call(self): ''' Checks if there is a next on-call available or not. :return: (boolean) True if there is; False otherwise ''' if len(self.policy_on_calls) > 0 and self.forwarding_count < self.max_forwarding_users: return True return False def next_on_call_user_id(self): ''' Gets the next on call user ID. :return: (int) user ID ''' # We can do this because the forwarding count is expected to be updated after a dial # and it starts from 1 unlike list index which starts at 0 return self.policy_on_calls[self.get_forwarding_index()][0] def next_on_call_phone_number(self): ''' Gets the next on call user phone number. :return: (str) phone number in the format +xxxxxxxx ''' on_call_phone_number = self.on_call_phones[self.next_on_call_user_id()] if on_call_phone_number[0] != '+': on_call_phone_number = '+' + on_call_phone_number return on_call_phone_number def increase_forwarding_count(self): ''' Increases the forwarding count by 1. ''' self.forwarding_count += 1 def last_on_call_user_id(self): ''' Gets the ID of the last user who was called. :return: (int) user ID ''' # We can do this because the forwarding count is expected to be updated after a dial # and it starts from 1 unlike list index which starts at 0 return self.policy_on_calls[self.get_forwarding_index() - 1][0] def has_greeting(self): ''' Check if there is a greeting note (text or audio). :return: (boolean) True if it is; False otherwise ''' if self.greeting_text is None and self.greeting_audio_url is None: return False return True @staticmethod def twilio_reject(): ''' Gets the Twiml for rejecting a call. The cost of the call will be 0. :return: Twiml ''' resp = VoiceResponse() resp.reject() return resp def twilio_greeting(self, timeout=5): ''' Get the Twiml for greeting the caller. :param timeout: (int) number of seconds the call should wait for the user to do something after being greeted :return: Twiml ''' resp = VoiceResponse() gather = Gather(action=url_paths.live_call_routing_twilio_forwarding_start, method='POST', timeout=timeout, action_on_empty_result=True) if self.greeting_text is not None: gather.say(self.greeting_text, voice=self.get_voice_gender(), language=self.text_language) elif self.greeting_audio_url is not None: gather.play(self.greeting_audio_url) else: gather = Gather(action=url_paths.live_call_routing_twilio_forwarding_start, method='POST', timeout=0, action_on_empty_result=True) gather.say('') resp.append(gather) return resp def twilio_dial(self, on_call_phone_number): ''' Get the Twiml for dialing a user. :param on_call_phone_number: the phone number to dial. :return: Twiml ''' resp = VoiceResponse() action_url = url_paths.live_call_routing_twilio_forwarding_outcome record_url = url_paths.live_call_routing_twilio_recording dial = Dial(caller_id=self.caller if self.show_caller_id else self.phone, timeout=self.forwarding_timeout, action=action_url, answer_on_bridge=True, record=Twilio.record_from_answer if self.record_call else Twilio.do_not_record, recording_status_callback=record_url if self.record_call else None) dial.number(on_call_phone_number, url=url_paths.live_call_routing_twilio_forwarding_accept_prompt, method='POST') resp.append(dial) return resp def twilio_connection_prompt(self, timeout=10): ''' Get the Twiml for greeting the caller. :param timeout: (int) number of seconds the call should wait for the user to do something after being greeted :return: Twiml ''' lang = constants.lang_es if self.text_language == constants.lang_es else constants.lang_en resp = VoiceResponse() gather = Gather(action=url_paths.live_call_routing_twilio_forwarding_accept_confirm, method='POST', num_digits=1, timeout=timeout, action_on_empty_result=True) gather.say(self.get_call_acceptance_prompt(), voice=self.get_voice_gender(), language=lang) resp.append(gather) return resp def twilio_outbound_dial(self, outbound_number): ''' Get the Twiml for making an outbound call. :param outbound_number: the phone number to dial. :return: Twiml ''' resp = VoiceResponse() action_url = url_paths.live_call_routing_twilio_outgoing_outcome record_url = url_paths.live_call_routing_twilio_recording dial = Dial(caller_id=self.phone, timeout=30, action=action_url, record=Twilio.record_from_answer if self.record_call else Twilio.do_not_record, recording_status_callback=record_url if self.record_call else None) dial.number(outbound_number, status_callback=action_url, status_callback_method='POST', status_callback_event=' '.join([Twilio.call_status_answered, Twilio.call_status_completed])) resp.append(dial) return resp def twilio_ending(self): ''' Get the Twiml for ending the call. If there is no ending note, then send directly to voicemail recording. When that happens, the check_twilio_ending_view will be skipped. :return: Twiml ''' resp = VoiceResponse() gather = Gather(action=url_paths.live_call_routing_twilio_ending_check, method='POST', timeout=2, action_on_empty_result=True) if self.ending_text is not None: gather.say(self.ending_text, voice=self.get_voice_gender(), language=self.text_language) elif self.ending_audio_url is not None: gather.play(self.ending_audio_url) else: gather.say('') resp.append(gather) return resp @staticmethod def twilio_voicemail(): ''' Get the Twiml for recording voicemail. :return: Twiml ''' resp = VoiceResponse() resp.record(action=url_paths.live_call_routing_twilio_voicemail_process, method='POST', timeout=configs.call_routing_max_voicemail_timeout) return resp @staticmethod def twilio_hangup(): ''' Gets the Twiml for hanging up a call. :return: Twiml ''' resp = VoiceResponse() resp.hangup() return resp @staticmethod def twilio_empty(): ''' Returns an empty response. :return: Twiml ''' resp = VoiceResponse() return resp def caller_name(self): ''' Get the name of the caller. :return: caller's name ''' if self.caller_details is not None and var_names.name in self.caller_details and \ self.caller_details[var_names.name] is not None: return self.caller_details[var_names.name] return None def caller_group(self): ''' Get the name of the group the caller belongs to. :return: caller's group name ''' if self.caller_details is not None and var_names.group_name in self.caller_details and \ self.caller_details[var_names.group_name] is not None: return self.caller_details[var_names.group_name] return None def replace_dynamic_data_fields(self, msg): ''' Replace param fields in a text (e.g. {{ caller_name }}) with internal field values for live call routing. :param msg: text message :return: (str) updated text message ''' try: matches = re.findall(r'({{\s*[a-zA-Z0-9_.]+\s*(?:\|\s*[^{]+)?}})', msg) if len(matches) == 0: return msg else: lang = constants.lang_es if self.text_language == constants.lang_es else constants.lang_en for item in matches: param_matches = re.findall(r'^{{\s*([a-zA-Z0-9_.]+)\s*(?:\|\s*([^{]+))?}}$', item) for prm_item in param_matches: param = prm_item[0] if string_validator.is_empty_string(prm_item[1]): dft_val = _lt.get_label(lnm.dsm_unknown, lang) else: dft_val = prm_item[1].lstrip().rstrip() if param == var_names.caller_name: clr_name = self.caller_name() msg = msg.replace(item, clr_name) if clr_name is not None else msg.replace(item, dft_val) elif param == var_names.caller_group: clr_group = self.caller_group() msg = msg.replace(item, clr_group) if clr_group is not None else msg.replace(item, dft_val) elif param == var_names.routing_name: msg = msg.replace(item, self.routing_name) elif param == var_names.from_number: msg = msg.replace(item, self.caller)\ if self.caller is not None else msg.replace(item, dft_val) elif param == var_names.to_number: msg = msg.replace(item, self.phone) return msg except Exception: return msg def get_incident_title(self): ''' Get the title to create the incident with. :return: (string) incident title ''' if self.incident_title_format is None or string_validator.is_empty_string(self.incident_title_format): lang = constants.lang_es if self.text_language == constants.lang_es else constants.lang_en msg = _lt.get_label(info.msg_live_call_route_default_incident_title, lang) else: msg = self.incident_title_format return self.replace_dynamic_data_fields(msg) def get_incident_description(self): ''' Get the payload that can be displayed as description for the incident. :return: (string) incident description ''' data = {var_names.routing_name: self.routing_name, var_names.from_number: self.caller, var_names.to_number: self.phone, var_names.caller_name: self.caller_name(), var_names.group: self.caller_group()} return str(data) def get_call_acceptance_prompt(self): ''' Get the text to play as call acceptance prompt. :return: (string) call acceptance prompt ''' if self.prompt_format is None or string_validator.is_empty_string(self.prompt_format): lang = constants.lang_es if self.text_language == constants.lang_es else constants.lang_en msg = _lt.get_label(info.msg_live_call_route_default_call_acceptance_prompt, lang) else: msg = self.prompt_format return self.replace_dynamic_data_fields(msg) @staticmethod def get_return_call_instructions(digits): ''' Get the return call instructions. :param digits: the digits that were sent as instructions :return: (tuple) -> phone number, organization instance ID ''' digits = '+' + digits if '*' in digits: digits_arr = digits.split('*') outbound_number = digits_arr[0] org_inst_id = int(digits_arr[1]) if string_validator.is_number(digits_arr[1]) else None return outbound_number, org_inst_id return digits, None def to_dict(self): data = { var_names.call_routing_id: self.call_routing_id, var_names.organization_id: self.organization_id, var_names.routing_name: self.routing_name, var_names.description: self.routing_desc, var_names.iso_country_code: self.iso_country_code, var_names.phone: self.phone, var_names.phone_type: self.phone_type, var_names.greeting_text: self.greeting_text, var_names.greeting_audio_filename: self.greeting_audio_filename, var_names.greeting_audio_url: self.greeting_audio_url, var_names.ending_text: self.ending_text, var_names.ending_audio_filename: self.ending_audio_filename, var_names.ending_audio_url: self.ending_audio_url, var_names.default_forwarding_service: self.default_forwarding_service, var_names.option_forwarding_services: self.option_forwarding_services, var_names.text_language: self.text_language, var_names.is_male_voice: self.is_male_voice, var_names.max_forwarding_users: self.max_forwarding_users, var_names.forwarding_timeout: self.forwarding_timeout, var_names.urgency_level: self.urgency_level, var_names.block_numbers: self.block_numbers, var_names.show_caller_id: self.show_caller_id, var_names.resolve_answered_calls: self.resolve_answered_calls, var_names.resolve_unanswered_calls: self.resolve_unanswered_calls, var_names.to_alert: self.to_alert, var_names.record_voicemail: self.record_voicemail, var_names.record_call: self.record_call, var_names.call_routing_ref_id: self.call_routing_ref_id, var_names.from_number: self.caller, var_names.twilio_call_sid: self.twilio_call_sid, var_names.on_call: self.policy_on_calls, var_names.on_call_phones: self.on_call_phones, var_names.instance_id: self.instance_id, var_names.call_status: self.call_status, var_names.forwarding_count: self.forwarding_count, var_names.call_duration: self.call_duration, var_names.machine_answered: self.machine_answered, var_names.prompt_call_acceptance: self.prompt_call_acceptance, var_names.prompt_format: self.prompt_format, var_names.incident_title_format: self.incident_title_format, var_names.caller_details: self.caller_details } return data