# By: Riasat Ullah # This file contains communication vendors through whom sms/voice calls can be made. # There is a super class and then each vendor is a subclass of it. from taskcallrest import settings from twilio.rest import Client as TwilioClient from urllib.parse import urlencode from utils import constants, file_storage, helpers, logging, s3, url_paths, var_names from validations import string_validator import configuration as configs class CommunicationVendor(object): def __init__(self, vendor_client): self.vendor_client = vendor_client class Twilio(CommunicationVendor): ''' This is a Twilio text message/voice call dispatcher class. ''' # Twilio call response variables var_AnsweredBy = 'AnsweredBy' var_CallDuration = 'CallDuration' var_CallSid = 'CallSid' var_CallStatus = 'CallStatus' var_DialCallStatus = 'DialCallStatus' var_Digits = 'Digits' var_From = 'From' var_FromCountry = 'FromCountry' var_MachineDetectionDuration = 'MachineDetectionDuration' var_ParentCallSid = 'ParentCallSid' var_RecordingUrl = 'RecordingUrl' var_To = 'To' # Twilio call status call_status_answered = 'answered' call_status_completed = 'completed' call_status_in_progress = 'in-progress' # Twilio voices voice_man = 'man' voice_woman = 'woman' # Twilio answered by options machine_start = 'machine_start' fax = 'fax' # Twilio record options do_not_record = 'do-not-record' record_from_answer = 'record-from-answer' def __init__(self, vendor_client): assert isinstance(vendor_client, TwilioClient) CommunicationVendor.__init__(self, vendor_client) @staticmethod def get_authentication(test=False): ''' Gets the credentials to connect to the Twilio account. :return: Twilio Client ''' str_live, str_test = 'live', 'test' str_sid = 'account_sid' str_token = 'auth_token' try: data = s3.read_json(file_storage.S3_BUCKET_TASKCALL_PROD_DATA, file_storage.S3_KEY_TWILIO_CREDENTIALS) if not test or not settings.TEST_MODE: creds = data[str_live] else: creds = data[str_test] client = TwilioClient(creds[str_sid], creds[str_token]) return client except (OSError, IOError) as e: err = 'Could not read Twilio credentials file' + '\n' + str(e) raise OSError(err) def send_sms(self, from_number: str, to_number: str, message: str): from_number = helpers.internationalize_phone_number(from_number) to_number = helpers.internationalize_phone_number(to_number) for i in range(0, 2): try: text_msg = self.vendor_client.messages.create(from_=from_number, to=to_number, body=message) return text_msg.sid except Exception as e: logging.exception(str(e)) continue def call(self, from_number: str, to_number: str, message: str, lang=constants.lang_en): from_number = helpers.internationalize_phone_number(from_number) to_number = helpers.internationalize_phone_number(to_number) message = helpers.add_space_between_numbers(message) for i in range(0, 2): try: params = {'language': lang, 'message': message} call_url = url_paths.communication_twilio_voice_create_content + '?' + urlencode(params) call = self.vendor_client.calls.create(from_=from_number, to=to_number, url=call_url, method='GET', time_limit=30) return call.sid except Exception as e: logging.exception(str(e)) continue def get_available_numbers(self, iso_code, phone_type, area_code=None, list_size=10): ''' Gets the list of available numbers on Twilio. Numbers are not available in all countries and only some countries have toll-free numbers. :param iso_code: 2-letter ISO code of the country the number is wanted of :param phone_type: (string) type of phone; (local, mobile or toll_free) :param area_code: (str) the area code of the desired number; this may not render a result :param list_size: (int) number of results to retrieve :return: (list) of phone numbers and their details ''' assert phone_type in configs.allowed_phone_number_types if area_code is not None: assert string_validator.is_number(area_code) available_numbers = [] try: if phone_type == constants.toll_free: numbers_list = self.vendor_client.available_phone_numbers(iso_code)\ .toll_free.list(area_code=area_code, limit=list_size) elif phone_type == constants.mobile: numbers_list = self.vendor_client.available_phone_numbers(iso_code)\ .mobile.list(area_code=area_code, limit=list_size) else: numbers_list = self.vendor_client.available_phone_numbers(iso_code)\ .local.list(area_code=area_code, limit=list_size) for item in numbers_list: available_numbers.append([item.friendly_name, item.phone_number]) except Exception as e: logging.exception(str(e)) finally: return available_numbers def create_address_resource(self, org_name, street, city, region, zip_code, iso_code): try: address_instance = self.vendor_client.addresses.create( friendly_name=org_name, customer_name=org_name, street=street, city=city, region=region, postal_code=zip_code, iso_country=iso_code ) return address_instance.sid except Exception: raise def provision_number(self, phone_number, address_sid, bundle_sid=None): ''' BE CAREFUL To provision or buy a number. When the number is provisioned, the account is charged. :param phone_number: (str) the phone number to provision in the format +xxxxxxx :param address_sid: SID of the address that should be associated with the phone number :param bundle_sid: approved regulatory bundle id (some countries require additional verification) :return: (str) -> Twilio phone SID ''' try: phone_instance = self.vendor_client.incoming_phone_numbers.create( phone_number=phone_number, voice_url=url_paths.live_call_routing_twilio_initiate, address_sid=address_sid, bundle_sid=bundle_sid ) return phone_instance.sid except Exception: raise def delete_number(self, phone_sid): ''' Deletes a provisioned phone number. :param phone_sid: phone SID :return: ''' try: response = self.vendor_client.incoming_phone_numbers(phone_sid).delete() return response except Exception: raise def force_call_end(self, call_sid): ''' Force a call to end by updating its status. :param call_sid: the call SID :return: ''' try: response = self.vendor_client.calls(call_sid).update(status=self.call_status_completed) return response except Exception: raise def get_parent_sid(self, child_sid): ''' Get the SID of the parent call. :param child_sid: SID of the child call :return: parent call SID ''' try: return self.vendor_client.calls(child_sid).fetch().parent_call_sid except Exception: raise def get_call_duration(self, call_sid): try: return int(self.vendor_client.calls(call_sid).fetch().duration) except Exception: raise def get_phone_iso_code(self, phone_number): ''' Get the ISO code of a phone number through Twilio lookup. :param phone_number: (str) the phone number to provision in the format +xxxxxxx :return: 2-letter ISO code (None if it is a valid number) ''' lookup_details = self.vendor_client.lookups.v2.phone_numbers(phone_number).fetch() return lookup_details.country_code