From 1e4b4355ce00b178cebbfd5e5300fb0cb6888457 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Mon, 22 Jul 2024 18:23:39 -0400 Subject: [PATCH] Twilio WhatsApp support added (#1173) --- apprise/plugins/twilio.py | 90 +++++++++++++++++++++++++++++++------- apprise/url.py | 4 +- apprise/utils.py | 12 +++-- test/test_plugin_twilio.py | 28 +++++++++++- 4 files changed, 111 insertions(+), 23 deletions(-) diff --git a/apprise/plugins/twilio.py b/apprise/plugins/twilio.py index 4205e37f..82569a2d 100644 --- a/apprise/plugins/twilio.py +++ b/apprise/plugins/twilio.py @@ -43,6 +43,7 @@ # or consider purchasing a short-code from here: # https://www.twilio.com/docs/glossary/what-is-a-short-code # +import re import requests from json import loads @@ -55,6 +56,22 @@ from ..utils import validate_regex from ..locale import gettext_lazy as _ +# Twilio Mode Detection +MODE_DETECT_RE = re.compile( + r'\s*((?P[^:]+)\s*:\s*)?(?P.+)$', re.I) + + +class TwilioMessageMode: + """ + Twilio Message Mode + """ + # SMS/MMS + TEXT = 'T' + + # via WhatsApp + WHATSAPP = 'W' + + class NotifyTwilio(NotifyBase): """ A wrapper for Twilio Notifications @@ -117,14 +134,14 @@ class NotifyTwilio(NotifyBase): 'name': _('From Phone No'), 'type': 'string', 'required': True, - 'regex': (r'^\+?[0-9\s)(+-]+$', 'i'), + 'regex': (r'^([a-z]+:)?\+?[0-9\s)(+-]+$', 'i'), 'map_to': 'source', }, 'target_phone': { 'name': _('Target Phone No'), 'type': 'string', 'prefix': '+', - 'regex': (r'^[0-9\s)(+-]+$', 'i'), + 'regex': (r'^([a-z]+:)?[0-9\s)(+-]+$', 'i'), 'map_to': 'targets', }, 'short_code': { @@ -190,7 +207,22 @@ class NotifyTwilio(NotifyBase): self.apikey = validate_regex( apikey, *self.template_args['apikey']['regex']) - result = is_phone_no(source, min_len=5) + # Detect mode + result = MODE_DETECT_RE.match(source) + if not result: + msg = 'The Account (From) Phone # or Short-code specified ' \ + '({}) is invalid.'.format(source) + self.logger.warning(msg) + raise TypeError(msg) + + # prepare our default mode to use for all numbers that follow in + # target definitions + self.default_mode = TwilioMessageMode.WHATSAPP \ + if result.group('mode') and \ + result.group('mode')[0].lower() == 'w' \ + else TwilioMessageMode.TEXT + + result = is_phone_no(result.group('phoneno'), min_len=5) if not result: msg = 'The Account (From) Phone # or Short-code specified ' \ '({}) is invalid.'.format(source) @@ -220,18 +252,35 @@ class NotifyTwilio(NotifyBase): # Parse our targets self.targets = list() - for target in parse_phone_no(targets): + for entry in parse_phone_no(targets, prefix=True): + # Detect mode + # w: (or whatsapp:) will trigger whatsapp message otherwise + # sms/mms as normal + result = MODE_DETECT_RE.match(entry) + mode = TwilioMessageMode.WHATSAPP if result.group('mode') and \ + result.group('mode')[0].lower() == 'w' else self.default_mode + # Validate targets and drop bad ones: - result = is_phone_no(target) + result = is_phone_no(result.group('phoneno')) if not result: self.logger.warning( 'Dropped invalid phone # ' - '({}) specified.'.format(target), + '({}) specified.'.format(entry), + ) + continue + + # We can't send twilio messages using short-codes as our source + if len(self.source) in (5, 6) and mode is \ + TwilioMessageMode.WHATSAPP: + self.logger.warning( + 'Dropped WhatsApp phone # ' + '({}) because source provided was a short-code.'.format( + entry), ) continue # store valid phone number - self.targets.append('+{}'.format(result['full'])) + self.targets.append((mode, '+{}'.format(result['full']))) return @@ -260,9 +309,8 @@ class NotifyTwilio(NotifyBase): # Prepare our payload payload = { 'Body': body, - 'From': self.source, - - # The To gets populated in the loop below + # The From and To gets populated in the loop below + 'From': None, 'To': None, } @@ -277,14 +325,20 @@ class NotifyTwilio(NotifyBase): if len(targets) == 0: # No sources specified, use our own phone no - targets.append(self.source) + targets.append((self.default_mode, self.source)) while len(targets): # Get our target to notify - target = targets.pop(0) + (mode, target) = targets.pop(0) # Prepare our user - payload['To'] = target + if mode is TwilioMessageMode.TEXT: + payload['From'] = self.source + payload['To'] = target + + else: # WhatsApp support (via Twilio) + payload['From'] = f'whatsapp:{self.source}' + payload['To'] = f'whatsapp:{target}' # Some Debug Logging self.logger.debug('Twilio POST URL: {} (cert_verify={})'.format( @@ -376,9 +430,13 @@ class NotifyTwilio(NotifyBase): sid=self.pprint( self.account_sid, privacy, mode=PrivacyMode.Tail, safe=''), token=self.pprint(self.auth_token, privacy, safe=''), - source=NotifyTwilio.quote(self.source, safe=''), + source=NotifyTwilio.quote( + self.source if self.default_mode is TwilioMessageMode.TEXT + else 'w:{}'.format(self.source), safe=''), targets='/'.join( - [NotifyTwilio.quote(x, safe='') for x in self.targets]), + [NotifyTwilio.quote( + x[1] if x[0] is TwilioMessageMode.TEXT + else 'w:{}'.format(x[1]), safe='') for x in self.targets]), params=NotifyTwilio.urlencode(params)) def __len__(self): @@ -442,6 +500,6 @@ class NotifyTwilio(NotifyBase): # The 'to' makes it easier to use yaml configuration if 'to' in results['qsd'] and len(results['qsd']['to']): results['targets'] += \ - NotifyTwilio.parse_phone_no(results['qsd']['to']) + NotifyTwilio.parse_phone_no(results['qsd']['to'], prefix=True) return results diff --git a/apprise/url.py b/apprise/url.py index 39daec86..76d623be 100644 --- a/apprise/url.py +++ b/apprise/url.py @@ -577,7 +577,7 @@ class URLBase: return content @staticmethod - def parse_phone_no(content, unquote=True): + def parse_phone_no(content, unquote=True, prefix=False): """A wrapper to utils.parse_phone_no() with unquoting support Parses a specified set of data and breaks it into a list. @@ -600,7 +600,7 @@ class URLBase: # Nothing further to do return [] - content = parse_phone_no(content) + content = parse_phone_no(content, prefix=prefix) return content diff --git a/apprise/utils.py b/apprise/utils.py index fd7f743a..5cd8256f 100644 --- a/apprise/utils.py +++ b/apprise/utils.py @@ -174,6 +174,11 @@ IS_PHONE_NO = re.compile(r'^\+?(?P[0-9\s)(+-]+)\s*$') PHONE_NO_DETECTION_RE = re.compile( r'\s*([+(\s]*[0-9][0-9()\s-]+[0-9])(?=$|[\s,+(]+[0-9])', re.I) +# Support for prefix: (string followed by colon) infront of phone no +PHONE_NO_WPREFIX_DETECTION_RE = re.compile( + r'\s*((?:[a-z]+:)?[+(\s]*[0-9][0-9()\s-]+[0-9])' + r'(?=$|(?:[a-z]+:)?[\s,+(]+[0-9])', re.I) + # A simple verification check to make sure the content specified # rougly conforms to a ham radio call sign before we parse it further IS_CALL_SIGN = re.compile( @@ -939,7 +944,7 @@ def parse_bool(arg, default=False): return bool(arg) -def parse_phone_no(*args, store_unparseable=True, **kwargs): +def parse_phone_no(*args, store_unparseable=True, prefix=False, **kwargs): """ Takes a string containing phone numbers separated by comma's and/or spaces and returns a list. @@ -948,7 +953,8 @@ def parse_phone_no(*args, store_unparseable=True, **kwargs): result = [] for arg in args: if isinstance(arg, str) and arg: - _result = PHONE_NO_DETECTION_RE.findall(arg) + _result = (PHONE_NO_DETECTION_RE if not prefix + else PHONE_NO_WPREFIX_DETECTION_RE).findall(arg) if _result: result += _result @@ -966,7 +972,7 @@ def parse_phone_no(*args, store_unparseable=True, **kwargs): elif isinstance(arg, (set, list, tuple)): # Use recursion to handle the list of phone numbers result += parse_phone_no( - *arg, store_unparseable=store_unparseable) + *arg, store_unparseable=store_unparseable, prefix=prefix) return result diff --git a/test/test_plugin_twilio.py b/test/test_plugin_twilio.py index 0e4f653f..29672e15 100644 --- a/test/test_plugin_twilio.py +++ b/test/test_plugin_twilio.py @@ -69,8 +69,8 @@ apprise_url_tests = ( # sid and token provided and from but invalid from no 'instance': TypeError, }), - ('twilio://AC{}:{}@{}/123/{}/abcd/'.format( - 'a' * 32, 'b' * 32, '3' * 11, '9' * 15), { + ('twilio://AC{}:{}@{}/123/{}/abcd/w:{}'.format( + 'a' * 32, 'b' * 32, '3' * 11, '9' * 15, 8 * 11), { # valid everything but target numbers 'instance': NotifyTwilio, }), @@ -81,6 +81,20 @@ apprise_url_tests = ( # Our expected url(privacy=True) startswith() response: 'privacy_url': 'twilio://...aaaa:b...b@12345', }), + ('twilio://AC{}:{}@98765/{}/w:{}/'.format( + 'a' * 32, 'b' * 32, '4' * 11, '5' * 11), { + # using short-code (5 characters) and 1 twillio address ignored + # because source phone number can not be a short code + 'instance': NotifyTwilio, + + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'twilio://...aaaa:b...b@98765', + }), + ('twilio://AC{}:{}@w:12345/{}/{}'.format( + 'a' * 32, 'b' * 32, '4' * 11, '5' * 11), { + # Invalid short-code + 'instance': TypeError, + }), ('twilio://AC{}:{}@123456/{}'.format('a' * 32, 'b' * 32, '4' * 11), { # using short-code (6 characters) 'instance': NotifyTwilio, @@ -95,6 +109,11 @@ apprise_url_tests = ( # use get args to acomplish the same thing 'instance': NotifyTwilio, }), + ('twilio://_?sid=AC{}&token={}&from={}&to=w:{}'.format( + 'a' * 32, 'b' * 32, '5' * 11, '6' * 11), { + # Support whatsapp (w: before number) + 'instance': NotifyTwilio, + }), ('twilio://_?sid=AC{}&token={}&source={}'.format( 'a' * 32, 'b' * 32, '5' * 11), { # use get args to acomplish the same thing (use source instead of from) @@ -228,6 +247,11 @@ def test_plugin_twilio_edge_cases(mock_post): NotifyTwilio( account_sid=account_sid, auth_token=None, source=source) + # Source is bad + with pytest.raises(TypeError): + NotifyTwilio( + account_sid=account_sid, auth_token=auth_token, source='') + # a error response response.status_code = 400 response.content = dumps({