diff options
Diffstat (limited to 'libs/apprise/plugins/NotifyPushed.py')
| -rw-r--r-- | libs/apprise/plugins/NotifyPushed.py | 372 |
1 files changed, 372 insertions, 0 deletions
diff --git a/libs/apprise/plugins/NotifyPushed.py b/libs/apprise/plugins/NotifyPushed.py new file mode 100644 index 000000000..a50970f99 --- /dev/null +++ b/libs/apprise/plugins/NotifyPushed.py @@ -0,0 +1,372 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2024, Chris Caron <[email protected]> +# +# 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 re +import requests +from json import dumps +from itertools import chain + +from .NotifyBase import NotifyBase +from ..URLBase import PrivacyMode +from ..common import NotifyType +from ..utils import parse_list +from ..utils import validate_regex +from ..AppriseLocale import gettext_lazy as _ + +# Used to detect and parse channels +IS_CHANNEL = re.compile(r'^#?(?P<name>[A-Za-z0-9]+)$') + +# Used to detect and parse a users push id +IS_USER_PUSHED_ID = re.compile(r'^@(?P<name>[A-Za-z0-9]+)$') + + +class NotifyPushed(NotifyBase): + """ + A wrapper to Pushed Notifications + + """ + + # The default descriptive name associated with the Notification + service_name = 'Pushed' + + # The services URL + service_url = 'https://pushed.co/' + + # The default secure protocol + secure_protocol = 'pushed' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushed' + + # Pushed uses the http protocol with JSON requests + notify_url = 'https://api.pushed.co/1/push' + + # A title can not be used for Pushed Messages. Setting this to zero will + # cause any title (if defined) to get placed into the message body. + title_maxlen = 0 + + # The maximum allowable characters allowed in the body per message + body_maxlen = 160 + + # Define object templates + templates = ( + '{schema}://{app_key}/{app_secret}', + '{schema}://{app_key}/{app_secret}@{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'app_key': { + 'name': _('Application Key'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'app_secret': { + 'name': _('Application Secret'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'target_user': { + 'name': _('Target User'), + 'prefix': '@', + 'type': 'string', + 'map_to': 'targets', + }, + 'target_channel': { + 'name': _('Target Channel'), + 'type': 'string', + 'prefix': '#', + 'map_to': 'targets', + }, + 'targets': { + 'name': _('Targets'), + 'type': 'list:string', + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'to': { + 'alias_of': 'targets', + }, + }) + + def __init__(self, app_key, app_secret, targets=None, **kwargs): + """ + Initialize Pushed Object + + """ + super().__init__(**kwargs) + + # Application Key (associated with project) + self.app_key = validate_regex(app_key) + if not self.app_key: + msg = 'An invalid Pushed Application Key ' \ + '({}) was specified.'.format(app_key) + self.logger.warning(msg) + raise TypeError(msg) + + # Access Secret (associated with project) + self.app_secret = validate_regex(app_secret) + if not self.app_secret: + msg = 'An invalid Pushed Application Secret ' \ + '({}) was specified.'.format(app_secret) + self.logger.warning(msg) + raise TypeError(msg) + + # Initialize channel list + self.channels = list() + + # Initialize user list + self.users = list() + + # Get our targets + targets = parse_list(targets) + if targets: + # Validate recipients and drop bad ones: + for target in targets: + result = IS_CHANNEL.match(target) + if result: + # store valid device + self.channels.append(result.group('name')) + continue + + result = IS_USER_PUSHED_ID.match(target) + if result: + # store valid room + self.users.append(result.group('name')) + continue + + self.logger.warning( + 'Dropped invalid channel/userid ' + '(%s) specified.' % target, + ) + + if len(self.channels) + len(self.users) == 0: + # We have no valid channels or users to notify after + # explicitly identifying at least one. + msg = 'No Pushed targets to notify.' + self.logger.warning(msg) + raise TypeError(msg) + + return + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform Pushed Notification + """ + + # Initiaize our error tracking + has_error = False + + # prepare JSON Object + payload = { + 'app_key': self.app_key, + 'app_secret': self.app_secret, + 'target_type': 'app', + 'content': body, + } + + # So the logic is as follows: + # - if no user/channel was specified, then we just simply notify the + # app. + # - if there are user/channels specified, then we only alert them + # while respecting throttle limits (in the event there are a lot of + # entries. + + if len(self.channels) + len(self.users) == 0: + # Just notify the app + return self._send( + payload=payload, notify_type=notify_type, **kwargs) + + # If our code reaches here, we want to target channels and users (by + # their Pushed_ID instead... + + # Generate a copy of our original list + channels = list(self.channels) + users = list(self.users) + + # Copy our payload + _payload = dict(payload) + _payload['target_type'] = 'channel' + + while len(channels) > 0: + # Get Channel + _payload['target_alias'] = channels.pop(0) + + if not self._send( + payload=_payload, notify_type=notify_type, **kwargs): + + # toggle flag + has_error = True + + # Copy our payload + _payload = dict(payload) + _payload['target_type'] = 'pushed_id' + + # Send all our defined User Pushed ID's + while len(users): + # Get User's Pushed ID + _payload['pushed_id'] = users.pop(0) + + if not self._send( + payload=_payload, notify_type=notify_type, **kwargs): + + # toggle flag + has_error = True + + return not has_error + + def _send(self, payload, notify_type, **kwargs): + """ + A lower level call that directly pushes a payload to the Pushed + Notification servers. This should never be called directly; it is + referenced automatically through the send() function. + """ + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json' + } + + self.logger.debug('Pushed POST URL: %s (cert_verify=%r)' % ( + self.notify_url, self.verify_certificate, + )) + self.logger.debug('Pushed Payload: %s' % str(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + self.notify_url, + data=dumps(payload), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifyPushed.http_response_code_lookup(r.status_code) + + self.logger.warning( + 'Failed to send Pushed notification:' + '{}{}error={}.'.format( + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug('Response Details:\r\n{}'.format(r.content)) + + # Return; we're done + return False + + else: + self.logger.info('Sent Pushed notification.') + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending Pushed notification.') + self.logger.debug('Socket Exception: %s' % str(e)) + + # Return; we're done + return False + + return True + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Our URL parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + return '{schema}://{app_key}/{app_secret}/{targets}/?{params}'.format( + schema=self.secure_protocol, + app_key=self.pprint(self.app_key, privacy, safe=''), + app_secret=self.pprint( + self.app_secret, privacy, mode=PrivacyMode.Secret, safe=''), + targets='/'.join( + [NotifyPushed.quote(x) for x in chain( + # Channels are prefixed with a pound/hashtag symbol + ['#{}'.format(x) for x in self.channels], + # Users are prefixed with an @ symbol + ['@{}'.format(x) for x in self.users], + )]), + params=NotifyPushed.urlencode(params)) + + def __len__(self): + """ + Returns the number of targets associated with this notification + """ + targets = len(self.channels) + len(self.users) + return targets if targets > 0 else 1 + + @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 load the results + return results + + # The first token is stored in the hostname + app_key = NotifyPushed.unquote(results['host']) + + entries = NotifyPushed.split_path(results['fullpath']) + # Now fetch the remaining tokens + try: + app_secret = entries.pop(0) + + except IndexError: + # Force some bad values that will get caught + # in parsing later + app_secret = None + app_key = None + + # Get our recipients (based on remaining entries) + results['targets'] = entries + + # The 'to' makes it easier to use yaml configuration + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'] += \ + NotifyPushed.parse_list(results['qsd']['to']) + + results['app_key'] = app_key + results['app_secret'] = app_secret + + return results |