# -*- coding: utf-8 -*- # BSD 2-Clause License # # Apprise - Push Notification Library. # Copyright (c) 2024, 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. # Youll need your own Revolt Bot and a Channel Id for the notifications to # be sent in since Revolt does not support webhooks yet. # # This plugin will simply work using the url of: # revolt://BOT_TOKEN/CHANNEL_ID # # API Documentation: # - https://api.revolt.chat/swagger/index.html # import requests from json import dumps, loads from datetime import timedelta from datetime import datetime from datetime import timezone from .NotifyBase import NotifyBase from ..common import NotifyImageSize from ..common import NotifyFormat from ..common import NotifyType from ..utils import validate_regex from ..utils import parse_list from ..AppriseLocale import gettext_lazy as _ class NotifyRevolt(NotifyBase): """ A wrapper for Revolt Notifications """ # The default descriptive name associated with the Notification service_name = 'Revolt' # The services URL service_url = 'https://revolt.chat/' # The default secure protocol secure_protocol = 'revolt' # A URL that takes you to the setup/help of the specific protocol setup_url = 'https://github.com/caronc/apprise/wiki/Notify_revolt' # Revolt Channel Message notify_url = 'https://api.revolt.chat/' # Revolt supports attachments but doesn't support it here (for now) attachment_support = False # Allows the user to specify the NotifyImageSize object image_size = NotifyImageSize.XY_256 # Revolt is kind enough to return how many more requests we're allowed to # continue to make within it's header response as: # X-RateLimit-Reset: The epoc time (in seconds) we can expect our # rate-limit to be reset. # X-RateLimit-Remaining: an integer identifying how many requests we're # still allow to make. request_rate_per_sec = 3 # Safety net clock_skew = timedelta(seconds=2) # The maximum allowable characters allowed in the body per message body_maxlen = 2000 # Title Maximum Length title_maxlen = 100 # Define object templates templates = ( '{schema}://{bot_token}/{targets}', ) # Defile out template tokens template_tokens = dict(NotifyBase.template_tokens, **{ 'bot_token': { 'name': _('Bot Token'), 'type': 'string', 'private': True, 'required': True, }, 'target_channel': { 'name': _('Channel ID'), 'type': 'string', 'map_to': 'targets', 'regex': (r'^[a-z0-9_-]+$', 'i'), 'private': True, 'required': True, }, 'targets': { 'name': _('Targets'), 'type': 'list:string', }, }) # Define our template arguments template_args = dict(NotifyBase.template_args, **{ 'channel': { 'alias_of': 'targets', }, 'bot_token': { 'alias_of': 'bot_token', }, 'icon_url': { 'name': _('Icon URL'), 'type': 'string' }, 'url': { 'name': _('Embed URL'), 'type': 'string', 'map_to': 'link', }, 'to': { 'alias_of': 'targets', }, }) def __init__(self, bot_token, targets, icon_url=None, link=None, **kwargs): super().__init__(**kwargs) # Bot Token self.bot_token = validate_regex(bot_token) if not self.bot_token: msg = 'An invalid Revolt Bot Token ' \ '({}) was specified.'.format(bot_token) self.logger.warning(msg) raise TypeError(msg) # Parse our Channel IDs self.targets = [] for target in parse_list(targets): results = validate_regex( target, *self.template_tokens['target_channel']['regex']) if not results: self.logger.warning( 'Dropped invalid Revolt channel ({}) specified.' .format(target), ) continue # Add our target self.targets.append(target) # Image for Embed self.icon_url = icon_url # Url for embed title self.link = link # For Tracking Purposes self.ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None) # Default to 1.0 self.ratelimit_remaining = 1.0 return def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): """ Perform Revolt Notification """ if len(self.targets) == 0: self.logger.warning('There were not Revolt channels to notify.') return False payload = {} # Acquire image_url image_url = self.icon_url \ if self.icon_url else self.image_url(notify_type) if self.notify_format == NotifyFormat.MARKDOWN: payload['embeds'] = [{ 'title': None if not title else title[0:self.title_maxlen], 'description': body, # Our color associated with our notification 'colour': self.color(notify_type), 'replies': None }] if image_url: payload['embeds'][0]['icon_url'] = image_url if self.link: payload['embeds'][0]['url'] = self.link else: payload['content'] = \ body if not title else "{}\n{}".format(title, body) has_error = False channel_ids = list(self.targets) for channel_id in channel_ids: postokay, response = self._send(payload, channel_id) if not postokay: # Failed to send message has_error = True return not has_error def _send(self, payload, channel_id, retries=1, **kwargs): """ Wrapper to the requests (post) object """ headers = { 'User-Agent': self.app_id, 'X-Bot-Token': self.bot_token, 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'application/json; charset=utf-8', } notify_url = '{0}channels/{1}/messages'.format( self.notify_url, channel_id ) self.logger.debug('Revolt POST URL: %s (cert_verify=%r)' % ( notify_url, self.verify_certificate )) self.logger.debug('Revolt Payload: %s' % str(payload)) # By default set wait to None wait = None now = datetime.now(timezone.utc).replace(tzinfo=None) if self.ratelimit_remaining <= 0.0: # Determine how long we should wait for or if we should wait at # all. This isn't fool-proof because we can't be sure the client # time (calling this script) is completely synced up with the # Discord server. One would hope we're on NTP and our clocks are # the same allowing this to role smoothly: if now < self.ratelimit_reset: # We need to throttle for the difference in seconds wait = abs( (self.ratelimit_reset - now + self.clock_skew) .total_seconds()) # Default content response object content = {} # Always call throttle before any remote server i/o is made; self.throttle(wait=wait) try: r = requests.post( notify_url, data=dumps(payload), headers=headers, verify=self.verify_certificate, timeout=self.request_timeout ) try: content = loads(r.content) except (AttributeError, TypeError, ValueError): # ValueError = r.content is Unparsable # TypeError = r.content is None # AttributeError = r is None content = {} # Handle rate limiting (if specified) try: # Store our rate limiting (if provided) self.ratelimit_remaining = \ int(r.headers.get('X-RateLimit-Remaining')) self.ratelimit_reset = \ now + timedelta(seconds=(int( r.headers.get('X-RateLimit-Reset-After')) / 1000)) except (TypeError, ValueError): # This is returned if we could not retrieve this # information gracefully accept this state and move on pass if r.status_code not in ( requests.codes.ok, requests.codes.no_content): # Some details to debug by self.logger.debug('Response Details:\r\n{}'.format( content if content else r.content)) # We had a problem status_str = \ NotifyBase.http_response_code_lookup(r.status_code) self.logger.warning( 'Revolt request limit reached; ' 'instructed to throttle for %.3fs', abs((self.ratelimit_reset - now + self.clock_skew) .total_seconds())) if r.status_code == requests.codes.too_many_requests \ and retries > 0: # Try again return self._send( payload=payload, channel_id=channel_id, retries=retries - 1, **kwargs) self.logger.warning( 'Failed to send to Revolt notification: ' '{}{}error={}.'.format( status_str, ', ' if status_str else '', r.status_code)) # Return; we're done return (False, content) else: self.logger.info('Sent Revolt notification.') except requests.RequestException as e: self.logger.warning( 'A Connection error occurred posting to Revolt.') self.logger.debug('Socket Exception: %s' % str(e)) return (False, content) return (True, content) def url(self, privacy=False, *args, **kwargs): """ Returns the URL built dynamically based on specified arguments. """ # Define any URL parameters params = {} if self.icon_url: params['icon_url'] = self.icon_url if self.link: params['url'] = self.link params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) return '{schema}://{bot_token}/{targets}/?{params}'.format( schema=self.secure_protocol, bot_token=self.pprint(self.bot_token, privacy, safe=''), targets='/'.join( [self.pprint(x, privacy, safe='') for x in self.targets]), params=NotifyRevolt.urlencode(params), ) def __len__(self): """ Returns the number of targets associated with this notification """ return 1 if not self.targets else len(self.targets) @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 # Store our bot token bot_token = NotifyRevolt.unquote(results['host']) # Now fetch the Channel IDs targets = NotifyRevolt.split_path(results['fullpath']) results['bot_token'] = bot_token results['targets'] = targets # Support the 'to' variable so that we can support rooms this way too # The 'to' makes it easier to use yaml configuration if 'to' in results['qsd'] and len(results['qsd']['to']): results['targets'] += \ NotifyRevolt.parse_list(results['qsd']['to']) # Support channel id on the URL string (if specified) if 'channel' in results['qsd']: results['targets'] += \ NotifyRevolt.parse_list(results['qsd']['channel']) # Support bot token on the URL string (if specified) if 'bot_token' in results['qsd']: results['bot_token'] = \ NotifyRevolt.unquote(results['qsd']['bot_token']) if 'icon_url' in results['qsd']: results['icon_url'] = \ NotifyRevolt.unquote(results['qsd']['icon_url']) if 'url' in results['qsd']: results['link'] = NotifyRevolt.unquote(results['qsd']['url']) if 'format' not in results['qsd'] and ( 'url' in results or 'icon_url' in results): # Markdown is implied results['format'] = NotifyFormat.MARKDOWN return results