diff --git a/apprise/plugins/clickatell.py b/apprise/plugins/clickatell.py new file mode 100644 index 00000000..053eb06b --- /dev/null +++ b/apprise/plugins/clickatell.py @@ -0,0 +1,206 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +# To use this service you will need a Clickatell account to which you can get your +# API_TOKEN at: +# https://www.clickatell.com/ +import requests + +from .base import NotifyBase +from ..common import NotifyType +from ..locale import gettext_lazy as _ +from ..url import PrivacyMode +from ..utils.parse import validate_regex, parse_phone_no + + +class NotifyClickatell(NotifyBase): + """ + A wrapper for Clickatell Notifications + """ + + service_name = _('Clickatell') + service_url = 'https://www.clickatell.com/' + secure_protocol = 'clickatell' + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_clickatell' + notify_url = 'https://platform.clickatell.com/messages/http/send?apiKey={}' + + templates = ( + '{schema}://{api_token}/{targets}', + '{schema}://{api_token}@{from_phone}/{targets}', + ) + + template_tokens = dict(NotifyBase.template_tokens, **{ + 'api_token': { + 'name': _('API Token'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'from_phone': { + 'name': _('From Phone No'), + 'type': 'string', + 'regex': (r'^[0-9\s)(+-]+$', 'i'), + }, + 'target_phone': { + 'name': _('Target Phone No'), + 'type': 'string', + 'prefix': '+', + 'regex': (r'^[0-9\s)(+-]+$', 'i'), + 'map_to': 'targets', + }, + 'targets': { + 'name': _('Targets'), + 'type': 'list:string', + 'required': True, + }, + }) + + template_args = dict(NotifyBase.template_args, **{ + 'token': { + 'alias_of': 'api_token' + }, + 'to': { + 'alias_of': 'targets', + }, + 'from': { + 'alias_of': 'from_phone', + }, + }) + + def __init__(self, api_token, from_phone, targets=None, **kwargs): + """ + Initialize Clickatell Object + """ + + super().__init__(**kwargs) + + self.api_token = validate_regex(api_token) + if not self.api_token: + msg = 'An invalid Clickatell API Token ' \ + '({}) was specified.'.format(api_token) + self.logger.warning(msg) + raise TypeError(msg) + + self.from_phone = validate_regex(from_phone) + self.targets = parse_phone_no(targets, prefix=True) + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + return '{schema}://{apikey}/?{params}'.format( + schema=self.secure_protocol, + apikey=self.quote(self.api_token, safe='/'), + params=self.urlencode(params), + ) + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform Clickatell Notification + """ + + if not self.targets: + self.logger.warning('There are no valid targets to notify.') + return False + + headers = { + 'User-Agent': self.app_id, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + } + + url = self.notify_url.format(self.api_token) + if self.from_phone: + url += '&from={}'.format(self.from_phone) + url += '&to={}'.format(','.join(self.targets)) + url += '&content={}'.format(' '.join([title, body])) + + self.logger.debug('Clickatell GET URL: %s', url) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.get( + url, + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code != requests.codes.ok \ + and r.status_code != requests.codes.accepted: + # We had a problem + status_str = self.http_response_code_lookup(r.status_code) + + self.logger.warning( + 'Failed to send Clickatell notification: ' + '{}{}error={}.'.format( + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug('Response Details:\r\n{}'.format(r.content)) + return False + else: + self.logger.info('Sent Clickatell notification.') + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending Clickatell ' + 'notification to %s.' % self.host) + self.logger.debug('Socket Exception: %s' % str(e)) + return False + return True + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to re-instantiate this object. + """ + results = NotifyBase.parse_url(url, verify_host=False) + if not results: + # We're done early as we couldn't parse the URL + return results + + results['targets'] = NotifyClickatell.split_path(results['fullpath']) + + if not results['targets']: + return results + + if results['user']: + results['api_token'] = NotifyClickatell.unquote(results['user']) + results['from_phone'] = NotifyClickatell.unquote(results['host']) + else: + results['api_token'] = NotifyClickatell.unquote(results['host']) + results['from_phone'] = '' + + return results diff --git a/test/test_plugin_clickatell.py b/test/test_plugin_clickatell.py new file mode 100644 index 00000000..d9cfbfcb --- /dev/null +++ b/test/test_plugin_clickatell.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +import logging +from json import dumps +from unittest import mock + +import pytest +import requests + +from apprise.plugins.clickatell import NotifyClickatell +from helpers import AppriseURLTester + +logging.disable(logging.CRITICAL) + +# Our Testing URLs +apprise_url_tests = ( + ('clickatell://', { + # only schema provided + 'instance': TypeError, + }), + ('clickatell:///', { + # invalid api_token + 'instance': TypeError, + }), + ('clickatell://@/', { + # invalid api_token + 'instance': TypeError, + }), + ('clickatell://{}/'.format('a' * 32), { + # no targets provided + 'instance': TypeError, + }), + ('clickatell://{}@/'.format('a' * 32), { + # no targets provided + 'instance': TypeError, + }), + ('clickatell://{}@{}/'.format('a' * 32, '1' * 9), { + # no targets provided + 'instance': TypeError, + }), + ('clickatell://{}@{}'.format('a' * 32, 'b' * 32, '3' * 9), { + # no targets provided + 'instance': TypeError, + }), + ('clickatell://{}@{}/123/{}/abcd'.format( + 'a' * 32, '1' * 6, '3' * 11), { + # valid everything but target numbers + 'instance': NotifyClickatell, + }), + ('clickatell://{}/{}'.format('a' * 32, '1' * 9), { + # everything valid + 'instance': NotifyClickatell, + }), + ('clickatell://{}@{}/{}'.format('a' * 32, '1' * 9, '1' * 9), { + # everything valid + 'instance': NotifyClickatell, + }), + ('clickatell://_?token={}&from={}&to={},{}'.format( + 'a' * 32, '1' * 9, '1' * 9, '1' * 9), { + # use get args to accomplish the same thing + 'instance': NotifyClickatell, + }), + ('clickatell://_?token={}'.format('a' * 32), { + # use get args + 'instance': NotifyClickatell, + 'notify_response': False, + }), + ('clickatell://_?token={}&from={}'.format('a' * 32, '1' * 9), { + # use get args + 'instance': NotifyClickatell, + 'notify_response': False, + }), + ('clickatell://{}/{}'.format('a' * 32, '1' * 9), { + 'instance': NotifyClickatell, + # throw a bizarre code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('clickatell://{}@{}/{}'.format('a' * 32, '1' * 9, '1' * 9), { + 'instance': NotifyClickatell, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracefully handle them + 'test_requests_exceptions': True, + }), +) + + +def test_plugin_clickatell_urls(): + """ + NotifyClickatell() Apprise URLs + """ + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +def test_plugin_clickatell_edge_cases(mock_post): + """ + NotifyClickatell() Edge Cases + """ + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + + # Initialize some generic (but valid) tokens + api_token = 'b' * 32 + from_phone = '+1 (555) 123-3456' + + # No api_token specified + with pytest.raises(TypeError): + NotifyClickatell(api_token=None, from_phone=from_phone) + + # a error response + response.status_code = 400 + response.content = dumps({ + 'code': 21211, + 'message': "The 'To' number +1234567 is not a valid phone number.", + }) + mock_post.return_value = response + + # Initialize our object + obj = NotifyClickatell(api_token=api_token, from_phone=from_phone) + + # We will fail with the above error code + assert obj.notify('title', 'body', 'info') is False