import logging import urllib.parse from string import digits from django.db.models import F, Q from phonenumbers import COUNTRY_CODE_TO_REGION_CODE from twilio.base.exceptions import TwilioRestException from twilio.rest import Client from twilio.twiml.voice_response import Gather, Say, VoiceResponse from apps.base.models import LiveSetting from apps.base.utils import live_settings from apps.phone_notifications.exceptions import ( FailedToFinishVerification, FailedToMakeCall, FailedToSendSMS, FailedToStartVerification, ) from apps.phone_notifications.phone_provider import PhoneProvider, ProviderFlags from apps.twilioapp.gather import get_alert_group_gather_instructions, get_gather_url from apps.twilioapp.models import ( TwilioCallStatuses, TwilioPhoneCall, TwilioPhoneCallSender, TwilioSMS, TwilioSmsSender, TwilioVerificationSender, ) from apps.twilioapp.status_callback import get_call_status_callback_url, get_sms_status_callback_url logger = logging.getLogger(__name__) class TwilioPhoneProvider(PhoneProvider): def make_notification_call(self, number: str, message: str) -> TwilioPhoneCall | None: message = self._escape_call_message(message) twiml = self._message_to_twiml_gather(message) response = None try_without_callback = False try: response = self._call_create(twiml, number, with_callback=True) except TwilioRestException as e: # If status callback is not valid and not accessible from public url then trying to send message without it # https://www.twilio.com/docs/api/errors/21609 if e.code == 21609: logger.info("TwilioPhoneProvider.make_notification_call: error 21609, calling without callback_url") try_without_callback = True else: logger.error(f"TwilioPhoneProvider.make_notification_call: failed {e}") raise FailedToMakeCall(graceful_msg=self._get_graceful_msg(e, number)) if try_without_callback: try: response = self._call_create(twiml, number, with_callback=False) except TwilioRestException as e: logger.error(f"TwilioPhoneProvider.make_notification_call: failed {e}") raise FailedToMakeCall(graceful_msg=self._get_graceful_msg(e, number)) if response and response.status and response.sid: return TwilioPhoneCall( status=TwilioCallStatuses.DETERMINANT.get(response.status, None), sid=response.sid, ) return None def send_notification_sms(self, number: str, message: str) -> TwilioSMS | None: try_without_callback = False response = None try: response = self._messages_create(number, message, with_callback=True) except TwilioRestException as e: # If status callback is not valid and not accessible from public url then trying to send message without it # https://www.twilio.com/docs/api/errors/21609 if e.code == 21609: logger.info("TwilioPhoneProvider.send_notification_sms: error 21609, sending without callback_url") try_without_callback = True else: logger.error(f"TwilioPhoneProvider.send_notification_sms: failed {e}") raise FailedToSendSMS(graceful_msg=self._get_graceful_msg(e, number)) if try_without_callback: try: response = self._messages_create(number, message, with_callback=False) except TwilioRestException as e: logger.error(f"TwilioPhoneProvider.send_notification_sms: failed {e}") raise FailedToSendSMS(graceful_msg=self._get_graceful_msg(e, number)) if response and response.status and response.sid: return TwilioSMS( status=TwilioCallStatuses.DETERMINANT.get(response.status, None), sid=response.sid, ) return None def send_verification_sms(self, number: str): self._send_verification_code(number, via="sms") def finish_verification(self, number: str, code: str): # I'm not sure if we need verification_and_parse via twilio pipeline here # Verification code anyway is sent to not verified phone number. # Just leaving it as it was before phone_provider refactoring. normalized_number, _ = self._normalize_phone_number(number) if normalized_number: try: client, verify_service_sid = self._verify_sender(number) verification_check = client.verify.services(verify_service_sid).verification_checks.create( to=normalized_number, code=code ) logger.info(f"TwilioPhoneProvider.finish_verification: verification_status {verification_check.status}") if verification_check.status == "approved": return normalized_number except TwilioRestException as e: logger.error(f"TwilioPhoneProvider.finish_verification: failed to verify number {number}: {e}") raise FailedToFinishVerification(graceful_msg=self._get_graceful_msg(e, number)) else: return None """ Errors we will raise without graceful messages: 20404 - We should not be requesting missing resources 30808 - Unknown error, likely on the carrier side 30007, 32017 - Blocked or filtered, Intermediary / Carrier Analytics blocked call due to poor reputation score on the telephone number: * We need to register our number or sender with the analytics provider or carrier for that jurisdiction """ def _get_graceful_msg(self, e, number): if e.code in (30003, 30005): return f"Destination handset {number} is unreachable" elif e.code == 30004: return f"Sending message to {number} is blocked" elif e.code == 30006: return f"Cannot send to {number} is landline or unreachable carrier" elif e.code == 30410: return f"Provider for {number} is experiencing timeouts" elif e.code == 60200: return f"{number} is incorrectly formatted" elif e.code in (21215, 60410, 60605): return f"Verification to {number} is blocked" elif e.code == 60203: return f"Max verification attempts for {number} reached" return None def make_call(self, number: str, message: str): twiml = self._message_to_twiml_say(message) try: self._call_create(twiml, number, with_callback=False) except TwilioRestException as e: logger.error(f"TwilioPhoneProvider.make_call: failed {e}") raise FailedToMakeCall(graceful_msg=self._get_graceful_msg(e, number)) def send_sms(self, number: str, message: str): try: self._messages_create(number, message, with_callback=False) except TwilioRestException as e: logger.error(f"TwilioPhoneProvider.send_sms: failed {e}") raise FailedToSendSMS(graceful_msg=self._get_graceful_msg(e, number)) def _message_to_twiml_say(self, message: str) -> VoiceResponse: response = VoiceResponse() say = Say(message) response.append(say) return response def _message_to_twiml_gather(self, message: str) -> VoiceResponse: response = VoiceResponse() gather = Gather(action=get_gather_url(), method="POST", num_digits=1) gather.say(message) gather.pause(length=1) gather.say(get_alert_group_gather_instructions()) response.append(gather) return response def _call_create(self, twiml: VoiceResponse, to: str, with_callback: bool): client, from_ = self._phone_sender(to) # encode twiml VoiceResponse to use in url twiml_query = urllib.parse.quote(str(twiml), safe="") url = "http://twimlets.com/echo?Twiml=" + twiml_query if with_callback: status_callback = get_call_status_callback_url() status_callback_events = ["initiated", "ringing", "answered", "completed"] return client.calls.create( url=url, to=to, from_=from_, method="GET", status_callback=status_callback, status_callback_event=status_callback_events, status_callback_method="POST", ) else: return client.calls.create( url=url, to=to, from_=from_, method="GET", ) def _messages_create(self, number: str, text: str, with_callback: bool): client, from_ = self._sms_sender(number) if with_callback: status_callback = get_sms_status_callback_url() return client.messages.create(body=text, to=number, from_=from_, status_callback=status_callback) else: return client.messages.create( body=text, to=number, from_=from_, ) def _send_verification_code(self, number: str, via: str): # https://www.twilio.com/docs/verify/api/verification?code-sample=code-start-a-verification-with-sms&code-language=Python&code-sdk-version=6.x try: client, verify_service_sid = self._verify_sender(number) verification = client.verify.services(verify_service_sid).verifications.create(to=number, channel=via) logger.info(f"TwilioPhoneProvider._send_verification_code: verification status {verification.status}") except TwilioRestException as e: logger.error(f"Twilio verification start error: {e} to number {number}") raise FailedToStartVerification(graceful_msg=self._get_graceful_msg(e, number)) def _normalize_phone_number(self, number: str): # TODO: phone_provider: is it best place to parse phone number? number = self._parse_phone_number(number) # Verify and parse phone number with Twilio API normalized_phone_number = None country_code = None if number != "" and number != "+": try: ok, normalized_phone_number, country_code = self._parse_number(number) if normalized_phone_number == "": normalized_phone_number = None country_code = None if not ok: normalized_phone_number = None country_code = None except TypeError: return None, None return normalized_phone_number, country_code # Use responsibly def _parse_number(self, number: str): try: response = self._default_twilio_api_client.lookups.phone_numbers(number).fetch() return True, response.phone_number, self._get_calling_code(response.country_code) except TwilioRestException as e: if e.code == 20404: # Not sure, why 20404 (NotFound according to TwilioDocs) handled gracefully, leaving it as it is. # https://www.twilio.com/docs/api/errors/20404" return False, None, None if e.code == 20003: raise e except KeyError as e: # Don't know why KeyError is gracefully handled here, probably exception raised by twilio_client. logger.info(f"twilio_client._parse_number: Gracefully handle KeyError: {e}") return False, None, None @property def _default_twilio_api_client(self): if live_settings.TWILIO_API_KEY_SID and live_settings.TWILIO_API_KEY_SECRET: return Client( live_settings.TWILIO_API_KEY_SID, live_settings.TWILIO_API_KEY_SECRET, live_settings.TWILIO_ACCOUNT_SID ) else: return Client(live_settings.TWILIO_ACCOUNT_SID, live_settings.TWILIO_AUTH_TOKEN) @property def _default_twilio_number(self): return live_settings.TWILIO_NUMBER def _twilio_sender(self, sender_model, to): _, _, country_code = self._parse_number(to) sender = ( sender_model.objects.filter(Q(country_code=country_code) | Q(country_code__isnull=True)) .order_by(F("country_code").desc(nulls_last=True)) .first() ) if sender: return sender.account.get_twilio_api_client(), sender return self._default_twilio_api_client, None def _sms_sender(self, to): client, sender = self._twilio_sender(TwilioSmsSender, to) return client, sender.sender if sender else self._default_twilio_number def _phone_sender(self, to): client, sender = self._twilio_sender(TwilioPhoneCallSender, to) return client, sender.number if sender else self._default_twilio_number def _verify_sender(self, to): client, sender = self._twilio_sender(TwilioVerificationSender, to) return client, sender.verify_service_sid if sender else live_settings.TWILIO_VERIFY_SERVICE_SID def _get_calling_code(self, iso): for code, isos in COUNTRY_CODE_TO_REGION_CODE.items(): if iso.upper() in isos: return code return None def _escape_call_message(self, message): # https://www.twilio.com/docs/api/errors/12100 message = message.replace("&", "&") message = message.replace(">", ">") message = message.replace("<", "<") return message def _parse_phone_number(self, raw_phone_number): return "+" + "".join(c for c in raw_phone_number if c in digits) @property def flags(self) -> ProviderFlags: return ProviderFlags( configured=not LiveSetting.objects.filter(name__startswith="TWILIO", error__isnull=False).exists(), test_sms=False, test_call=True, verification_call=False, verification_sms=True, )