extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-10-05 09:42:21 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #5 from caronc/4-discord-support

Browse Source 
Added Discord Support; refs #4
This commit is contained in:
lead2gold 2018-02-25 20:40:46 -05:00 committed by GitHub
commit e544260ec5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 490 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -22,6 +22,7 @@ The table below identifies the services this tool supports and some example serv
| Notification Service | Service ID | Default Port | Example Syntax |
| -------------------- | ---------- | ------------ | -------------- |
| [Boxcar](https://github.com/caronc/apprise/wiki/Notify_boxcar) | boxcar:// | (TCP) 443 | boxcar://hostname<br />boxcar://hostname/@tag<br/>boxcar://hostname/device_token<br />boxcar://hostname/device_token1/device_token2/device_tokenN<br />boxcar://hostname/@tag/@tag2/device_token
| [Discord](https://github.com/caronc/apprise/wiki/Notify_discord) | discord:// | (TCP) 443 | discord://webhook_id/webhook_token<br />discord://avatar@webhook_id/webhook_token
| [Faast](https://github.com/caronc/apprise/wiki/Notify_faast) | faast:// | (TCP) 443 | faast://authorizationtoken
| [Growl](https://github.com/caronc/apprise/wiki/Notify_growl) | growl:// | (UDP) 23053 | growl://hostname<br />growl://hostname:portno<br />growl://password@hostname<br />growl://password@hostname:port</br>_Note: you can also use the get parameter _version_ which can allow the growl request to behave using the older v1.x protocol. An example would look like: growl://hostname?version=1
| [Join](https://github.com/caronc/apprise/wiki/Notify_join) | join:// | (TCP) 443 | join://apikey/device<br />join://apikey/device1/device2/deviceN/<br />join://apikey/group<br />join://apikey/groupA/groupB/groupN<br />join://apikey/DeviceA/groupA/groupN/DeviceN/

76
apprise/AppriseAsset.py
View File

@ -33,6 +33,15 @@ class AppriseAsset(object):
URL masks.
"""
# Application Identifier
app_id = 'Apprise'
# Application Description
app_desc = 'Apprise Notifications'
# Provider URL
app_url = 'https://github.com/caronc/apprise'
# A Simple Mapping of Colors; For every NOTIFY_TYPE identified,
# there should be a mapping to it's color here:
html_notify_map = {
@ -52,6 +61,10 @@ class AppriseAsset(object):
image_url_mask = \
'http://nuxref.com/apprise/themes/{THEME}/apprise-{TYPE}-{XY}.png'
# Application Logo
image_url_logo = \
'http://nuxref.com/apprise/themes/{THEME}/apprise-logo.png'
# Image Path Mask
image_path_mask = abspath(join(
dirname(__file__),
@ -76,20 +89,48 @@ class AppriseAsset(object):
if image_url_mask is not None:
self.image_url_mask = image_url_mask
def html_color(self, notify_type):
def color(self, notify_type, color_type=None):
"""
Returns an HTML mapped color based on passed in notify type
if color_type is:
None then a standard hex string is returned as
a string format ('#000000').
int then the integer representation is returned
tuple then the the red, green, blue is returned in a tuple
"""
# Attempt to get the type, otherwise return a default grey
# if we couldn't look up the entry
return self.html_notify_map.get(notify_type, self.default_html_color)
color = self.html_notify_map.get(notify_type, self.default_html_color)
if color_type is None:
# This is the default return type
return color
def image_url(self, notify_type, image_size):
elif color_type is int:
# Convert the color to integer
return AppriseAsset.hex_to_int(color)
# The only other type is tuple
elif color_type is tuple:
return AppriseAsset.hex_to_rgb(color)
# Unsupported type
raise ValueError(
'AppriseAsset html_color(): An invalid color_type was specified.')
def image_url(self, notify_type, image_size, logo=False):
"""
Apply our mask to our image URL
if logo is set to True, then the logo_url is used instead
"""
if not self.image_url_mask:
url_mask = self.image_url_logo if logo else self.image_url_mask
if not url_mask:
# No image to return
return None
@ -105,7 +146,7 @@ class AppriseAsset(object):
re.IGNORECASE,
)
return re_table.sub(lambda x: re_map[x.group()], self.image_url_mask)
return re_table.sub(lambda x: re_map[x.group()], url_mask)
def image_path(self, notify_type, image_size, must_exist=True):
"""
@ -154,3 +195,28 @@ class AppriseAsset(object):
return None
return None
@staticmethod
def hex_to_rgb(value):
"""
Takes a hex string (such as #00ff00) and returns a tuple in the form
of (red, green, blue)
eg: #00ff00 becomes : (0, 65535, 0)
"""
value = value.lstrip('#')
lv = len(value)
return tuple(int(value[i:i + lv // 3], 16)
for i in range(0, lv, lv // 3))
@staticmethod
def hex_to_int(value):
"""
Takes a hex string (such as #00ff00) and returns its integer
equivalent
eg: #00000f becomes : 15
"""
return int(value.lstrip('#'), 16)

BIN
apprise/assets/themes/default/apprise-logo.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 42 KiB

33
apprise/plugins/NotifyBase.py
View File

@ -95,12 +95,6 @@ class NotifyBase(object):
# This value can be the same as the defined protocol.
secure_protocol = ''
# our Application identifier
app_id = 'Apprise'
# our Application description
app_desc = 'Apprise Notifications'
# Most Servers do not like more then 1 request per 5 seconds, so 5.5 gives
# us a safe play range...
throttle_attempt = 5.5
@ -177,7 +171,7 @@ class NotifyBase(object):
return
def image_url(self, notify_type):
def image_url(self, notify_type, logo=False):
"""
Returns Image URL if possible
"""
@ -191,6 +185,7 @@ class NotifyBase(object):
return self.asset.image_url(
notify_type=notify_type,
image_size=self.image_size,
logo=logo,
)
def image_path(self, notify_type):
@ -223,6 +218,30 @@ class NotifyBase(object):
image_size=self.image_size,
)
def color(self, notify_type, color_type=None):
"""
Returns the html color (hex code) associated with the notify_type
"""
if notify_type not in NOTIFY_TYPES:
return None
return self.asset.color(
notify_type=notify_type,
color_type=color_type,
)
@property
def app_id(self):
return self.asset.app_id
@property
def app_desc(self):
return self.asset.app_desc
@property
def app_url(self):
return self.asset.app_url
@staticmethod
def escape_html(html, convert_new_lines=False):
"""

251
apprise/plugins/NotifyDiscord.py Normal file
View File

@ -0,0 +1,251 @@
# -*- coding: utf-8 -*-
#
# Discord Notify Wrapper
#
# Copyright (C) 2018 Chris Caron <lead2gold@gmail.com>
#
# This file is part of apprise.
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
# For this to work correctly you need to create a webhook. To do this just
# click on the little gear icon next to the channel you're part of. From
# here you'll be able to access the Webhooks menu and create a new one.
#
# When you've completed, you'll get a URL that looks a little like this:
# https://discordapp.com/api/webhooks/417429632418316298/\
# JHZ7lQml277CDHmQKMHI8qBe7bk2ZwO5UKjCiOAF7711o33MyqU344Qpgv7YTpadV_js
#
# Simplified, it looks like this:
# https://discordapp.com/api/webhooks/WEBHOOK_ID/WEBHOOK_TOKEN
#
# This plugin will simply work using the url of:
# discord://WEBHOOK_ID/WEBHOOK_TOKEN
#
# API Documentation on Webhooks:
# - https://discordapp.com/developers/docs/resources/webhook
#
import requests
from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import parse_bool
# Image Support (256x256)
DISCORD_IMAGE_XY = NotifyImageSize.XY_256
class NotifyDiscord(NotifyBase):
"""
A wrapper to Discord Notifications
"""
# The default secure protocol
secure_protocol = 'discord'
# Discord Webhook
notify_url = 'https://discordapp.com/api/webhooks'
def __init__(self, webhook_id, webhook_token, tts=False, avatar=True,
footer=False, thumbnail=True, **kwargs):
"""
Initialize Discord Object
"""
super(NotifyDiscord, self).__init__(
title_maxlen=250, body_maxlen=2000,
image_size=DISCORD_IMAGE_XY, **kwargs)
if not webhook_id:
raise TypeError(
'An invalid Client ID was specified.'
)
if not webhook_token:
raise TypeError(
'An invalid Webhook Token was specified.'
)
# Store our data
self.webhook_id = webhook_id
self.webhook_token = webhook_token
# Text To Speech
self.tts = tts
# Over-ride Avatar Icon
self.avatar = avatar
# Place a footer icon
self.footer = footer
# Place a thumbnail image inline with the message body
self.thumbnail = thumbnail
return
def notify(self, title, body, notify_type, **kwargs):
"""
Perform Discord Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'multipart/form-data',
}
# Prepare JSON Object
payload = {
# Text-To-Speech
'tts': self.tts,
# If Text-To-Speech is set to True, then we do not want to wait
# for the whole message before continuing. Otherwise, we wait
'wait': self.tts is False,
# Our color associated with our notification
'color': self.color(notify_type, int),
'embeds': [{
'provider': {
'name': self.app_id,
'url': self.app_url,
},
'title': title,
'type': 'rich',
'description': body,
}]
}
if self.footer:
logo_url = self.image_url(notify_type, logo=True)
payload['embeds'][0]['footer'] = {
'text': self.app_desc,
}
if logo_url:
payload['embeds'][0]['footer']['icon_url'] = logo_url
image_url = self.image_url(notify_type)
if image_url:
if self.thumbnail:
payload['embeds'][0]['thumbnail'] = {
'url': image_url,
'height': 256,
'width': 256,
}
if self.avatar:
payload['avatar_url'] = image_url
if self.user:
# Optionally override the default username of the webhook
payload['username'] = self.user
# Construct Notify URL
notify_url = '{0}/{1}/{2}'.format(
self.notify_url,
self.webhook_id,
self.webhook_token,
)
self.logger.debug('Discord POST URL: %s (cert_verify=%r)' % (
notify_url, self.verify_certificate,
))
self.logger.debug('Discord Payload: %s' % str(payload))
try:
r = requests.post(
notify_url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
)
if r.status_code not in (
requests.codes.ok, requests.codes.no_content):
# We had a problem
try:
self.logger.warning(
'Failed to send Discord notification: '
'%s (error=%s).' % (
HTTP_ERROR_MAP[r.status_code],
r.status_code))
except KeyError:
self.logger.warning(
'Failed to send Discord notification '
'(error=%s).' % r.status_code)
self.logger.debug('Response Details: %s' % r.raw.read())
# Return; we're done
return False
else:
self.logger.info('Sent Discord notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending Discord '
'notification.'
)
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 substantiate this object.
Syntax:
discord://webhook_id/webhook_token
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Store our webhook ID
webhook_id = results['host']
# Now fetch our tokens
try:
webhook_token = [x for x in filter(bool, NotifyBase.split_path(
results['fullpath']))][0]
except (ValueError, AttributeError, IndexError):
# Force some bad values that will get caught
# in parsing later
webhook_token = None
results['webhook_id'] = webhook_id
results['webhook_token'] = webhook_token
# Text To Speech
results['tts'] = parse_bool(results['qsd'].get('tts', False))
# Use Footer
results['footer'] = parse_bool(results['qsd'].get('footer', False))
# Update Avatar Icon
results['avatar'] = parse_bool(results['qsd'].get('avatar', True))
# Use Thumbnail
results['thumbnail'] = \
parse_bool(results['qsd'].get('thumbnail', True))
return results

2
apprise/plugins/NotifySlack.py
View File

@ -220,7 +220,7 @@ class NotifySlack(NotifyBase):
'attachments': [{
'title': title,
'text': body,
'color': self.asset.html_color(notify_type),
'color': self.color(notify_type),
# Time
'ts': time(),
'footer': self.app_id,

3
apprise/plugins/__init__.py
View File

@ -21,6 +21,7 @@
from . import NotifyEmail as NotifyEmailBase
from .NotifyBoxcar import NotifyBoxcar
from .NotifyDiscord import NotifyDiscord
from .NotifyEmail import NotifyEmail
from .NotifyFaast import NotifyFaast
from .NotifyGrowl.NotifyGrowl import NotifyGrowl
@ -55,7 +56,7 @@ __all__ = [
'NotifyMyAndroid', 'NotifyProwl', 'NotifyPushalot', 'NotifyPushBullet',
'NotifyPushover', 'NotifyRocketChat', 'NotifyToasty', 'NotifyTwitter',
'NotifyXBMC', 'NotifyXML', 'NotifySlack', 'NotifyJoin', 'NotifyTelegram',
'NotifyMatterMost', 'NotifyPushjet',
'NotifyMatterMost', 'NotifyPushjet', 'NotifyDiscord',
# Reference
'NotifyImageSize', 'NOTIFY_IMAGE_SIZES', 'NotifyType', 'NOTIFY_TYPES',

26
test/test_api.py
View File

@ -298,8 +298,30 @@ def test_apprise_asset(tmpdir):
a.default_html_color = '#abcabc'
a.html_notify_map[NotifyType.INFO] = '#aaaaaa'
assert(a.html_color('invalid') == '#abcabc')
assert(a.html_color(NotifyType.INFO) == '#aaaaaa')
assert(a.color('invalid', tuple) == (171, 202, 188))
assert(a.color(NotifyType.INFO, tuple) == (170, 170, 170))
assert(a.color('invalid', int) == 11258556)
assert(a.color(NotifyType.INFO, int) == 11184810)
assert(a.color('invalid', None) == '#abcabc')
assert(a.color(NotifyType.INFO, None) == '#aaaaaa')
# None is the default
assert(a.color(NotifyType.INFO) == '#aaaaaa')
# Invalid Type
try:
a.color(NotifyType.INFO, dict)
# We should not get here (exception should be thrown)
assert(False)
except ValueError:
# The exception we expect since dict is not supported
assert(True)
except:
# Any other exception is not good
assert(False)
assert(a.image_url(NotifyType.INFO, NotifyImageSize.XY_256) ==
'http://localhost/dark/info-256x256.png')

10
test/test_notify_base.py
View File

@ -20,6 +20,7 @@ from apprise.plugins.NotifyBase import NotifyBase
from apprise import NotifyType
from apprise import NotifyImageSize
from timeit import default_timer
from apprise.utils import compat_is_basestring
def test_notify_base():
@ -75,6 +76,15 @@ def test_notify_base():
assert nb.image_path(notify_type=NotifyType.INFO) is None
assert nb.image_raw(notify_type=NotifyType.INFO) is None
# Color handling
assert nb.color(notify_type='invalid') is None
assert compat_is_basestring(
nb.color(notify_type=NotifyType.INFO, color_type=None))
assert isinstance(
nb.color(notify_type=NotifyType.INFO, color_type=int), int)
assert isinstance(
nb.color(notify_type=NotifyType.INFO, color_type=tuple), tuple)
# Create an object with an ImageSize loaded into it
nb = NotifyBase(image_size=NotifyImageSize.XY_256)

104
test/test_rest_plugins.py
View File

@ -87,6 +87,65 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifyDiscord
##################################
('discord://', {
'instance': None,
}),
# No webhook_token specified
('discord://%s' % ('i' * 24), {
'instance': TypeError,
}),
# Provide both an webhook id and a webhook token
('discord://%s/%s' % ('i' * 24, 't' * 64), {
'instance': plugins.NotifyDiscord,
'requests_response_code': requests.codes.no_content,
}),
# Provide a temporary username
('discord://l2g@%s/%s' % ('i' * 24, 't' * 64), {
'instance': plugins.NotifyDiscord,
'requests_response_code': requests.codes.no_content,
}),
# Enable other options
('discord://%s/%s?footer=Yes&thumbnail=Yes' % ('i' * 24, 't' * 64), {
'instance': plugins.NotifyDiscord,
'requests_response_code': requests.codes.no_content,
}),
('discord://%s/%s?avatar=No&footer=No' % ('i' * 24, 't' * 64), {
'instance': plugins.NotifyDiscord,
'requests_response_code': requests.codes.no_content,
}),
# Test without image set
('discord://%s/%s' % ('i' * 24, 't' * 64), {
'instance': plugins.NotifyDiscord,
'requests_response_code': requests.codes.no_content,
# don't include an image by default
'include_image': False,
}),
# An invalid url
('discord://:@/', {
'instance': None,
}),
('discord://%s/%s/' % ('a' * 24, 'b' * 64), {
'instance': plugins.NotifyDiscord,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('discord://%s/%s/' % ('a' * 24, 'b' * 64), {
'instance': plugins.NotifyDiscord,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('discord://%s/%s/' % ('a' * 24, 'b' * 64), {
'instance': plugins.NotifyDiscord,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyFaast
##################################
@ -1384,6 +1443,51 @@ def test_notify_boxcar_plugin(mock_post, mock_get):
p.notify(body=None, title=None, notify_type=NotifyType.INFO) is True
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_discord_plugin(mock_post, mock_get):
"""
API: NotifyDiscord() Extra Checks
"""
# Initialize some generic (but valid) tokens
webhook_id = 'A' * 24
webhook_token = 'B' * 64
# Prepare Mock
mock_get.return_value = requests.Request()
mock_post.return_value = requests.Request()
mock_post.return_value.status_code = requests.codes.ok
mock_get.return_value.status_code = requests.codes.ok
# Empty Channel list
try:
plugins.NotifyDiscord(webhook_id=None, webhook_token=webhook_token)
assert(False)
except TypeError:
# we'll thrown because no webhook_id was specified
assert(True)
obj = plugins.NotifyDiscord(
webhook_id=webhook_id,
webhook_token=webhook_token,
footer=True, thumbnail=False)
# Disable throttling to speed up unit tests
obj.throttle_attempt = 0
# This call includes an image with it's payload:
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
# Toggle our logo availability
obj.asset.image_url_logo = None
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_join_plugin(mock_post, mock_get):