Added webhook support for sending Notification results (#139)

This commit is contained in:
Chris Caron 2023-10-15 16:21:13 -04:00 committed by GitHub
parent cd2135bb46
commit 6a8099d79e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 664 additions and 81 deletions

View File

@ -7,6 +7,7 @@ omit =
*apps.py,
*/migrations/*,
*/core/settings/*,
*/*/tests/*,
lib/*,
lib64/*,
*urls.py,

View File

@ -299,6 +299,7 @@ The use of environment variables allow you to provide over-rides to default sett
| `ALLOWED_HOSTS` | A list of strings representing the host/domain names that this API can serve. This is a security measure to prevent HTTP Host header attacks, which are possible even under many seemingly-safe web server configurations. By default this is set to `*` allowing any host. Use space to delimit more than one host.
| `APPRISE_PLUGIN_PATHS` | Apprise supports the ability to define your own `schema://` definitions and load them. To read more about how you can create your own customizations, check out [this link here](https://github.com/caronc/apprise/wiki/decorator_notify). You may define one or more paths (separated by comma `,`) here. By default the `apprise_api/var/plugin` directory is scanned (which does not include anything). Feel free to set this to an empty string to disable any custom plugin loading.
| `APPRISE_RECURSION_MAX` | This defines the number of times one Apprise API Server can (recursively) call another. This is to both support and mitigate abuse through [the `apprise://` schema](https://github.com/caronc/apprise/wiki/Notify_apprise_api) for those who choose to use it. When leveraged properly, you can increase this (recursion max) value and successfully load balance the handling of many notification requests through many additional API Servers. By default this value is set to `1` (one).
| `APPRISE_WEBHOOK_URL` | Define a Webhook that Apprise should `POST` results to upon each notification call made. This must be in the format of an `http://` or `https://` URI. By default no URL is specified and no webhook is actioned.
| `APPRISE_WORKER_COUNT` | Over-ride the number of workers to run. by default this is `(2 * CPUS) + 1` as advised by Gunicorn's website.
| `APPRISE_WORKER_TIMEOUT` | Over-ride the worker timeout value; by default this is `300` (5 min) which should be more than enough time to send all pending notifications.
| `BASE_URL` | Those who are hosting the API behind a proxy that requires a subpath to gain access to this API should specify this path here as well. By default this is not set at all.

View File

@ -61,7 +61,8 @@ class AddTests(SimpleTestCase):
# However adding just 1 more character exceeds our limit and the save
# will fail
response = self.client.post(
'/add/{}'.format(key + 'x'), {'urls': 'mailto://user:pass@yahoo.ca'})
'/add/{}'.format(key + 'x'),
{'urls': 'mailto://user:pass@yahoo.ca'})
assert response.status_code == 404
@override_settings(APPRISE_CONFIG_LOCK=True)

View File

@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from django.test import SimpleTestCase
from unittest import mock
from ..utils import Attachment
from django.test.utils import override_settings
from tempfile import TemporaryDirectory
from shutil import rmtree
class AttachmentTests(SimpleTestCase):
def setUp(self):
# Prepare a temporary directory
self.tmp_dir = TemporaryDirectory()
def tearDown(self):
# Clear directory
try:
rmtree(self.tmp_dir.name)
except FileNotFoundError:
# no worries
pass
self.tmp_dir = None
def test_attachment_initialization(self):
"""
Test attachment handling
"""
with override_settings(APPRISE_ATTACH_DIR=self.tmp_dir.name):
with mock.patch('os.makedirs', side_effect=OSError):
with self.assertRaises(ValueError):
Attachment('file')
with mock.patch('tempfile.mkstemp', side_effect=FileNotFoundError):
with self.assertRaises(ValueError):
Attachment('file')
a = Attachment('file')
assert a.filename

View File

@ -23,7 +23,8 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from django.test import SimpleTestCase, override_settings
from unittest.mock import patch
from django.core.files.uploadedfile import SimpleUploadedFile
from unittest import mock
import requests
from ..forms import NotifyForm
import json
@ -36,7 +37,7 @@ class NotifyTests(SimpleTestCase):
Test notifications
"""
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_notify_by_loaded_urls(self, mock_notify):
"""
Test adding a simple notification and notifying it
@ -78,7 +79,146 @@ class NotifyTests(SimpleTestCase):
assert response.status_code == 200
assert mock_notify.call_count == 1
@patch('requests.post')
# Reset our mock object
mock_notify.reset_mock()
# Preare our form data
form_data = {
'body': 'test notifiction',
}
attach_data = {
'attachment': SimpleUploadedFile(
"attach.txt", b"content here", content_type="text/plain")
}
# At a minimum, just a body is required
form = NotifyForm(form_data, attach_data)
assert form.is_valid()
# Send our notification
response = self.client.post(
'/notify/{}'.format(key), form.cleaned_data)
assert response.status_code == 200
assert mock_notify.call_count == 1
# Reset our mock object
mock_notify.reset_mock()
# Test Headers
for level in ('CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG',
'TRACE', 'INVALID'):
# Preare our form data
form_data = {
'body': 'test notifiction',
}
attach_data = {
'attachment': SimpleUploadedFile(
"attach.txt", b"content here", content_type="text/plain")
}
# At a minimum, just a body is required
form = NotifyForm(form_data, attach_data)
assert form.is_valid()
# Prepare our header
headers = {
'HTTP_X-APPRISE-LOG-LEVEL': level,
}
# Send our notification
response = self.client.post(
'/notify/{}'.format(key), form.cleaned_data, **headers)
assert response.status_code == 200
assert mock_notify.call_count == 1
# Reset our mock object
mock_notify.reset_mock()
# Long Filename
attach_data = {
'attachment': SimpleUploadedFile(
"{}.txt".format('a' * 2000),
b"content here", content_type="text/plain")
}
# At a minimum, just a body is required
form = NotifyForm(form_data, attach_data)
assert form.is_valid()
# Send our notification
response = self.client.post(
'/notify/{}'.format(key), form.cleaned_data)
# We fail because the filename is too long
assert response.status_code == 400
assert mock_notify.call_count == 0
# Reset our mock object
mock_notify.reset_mock()
with override_settings(APPRISE_MAX_ATTACHMENTS=0):
# Preare our form data
form_data = {
'body': 'test notifiction',
}
attach_data = {
'attachment': SimpleUploadedFile(
"attach.txt", b"content here", content_type="text/plain")
}
# At a minimum, just a body is required
form = NotifyForm(form_data, attach_data)
assert form.is_valid()
# Send our notification
response = self.client.post(
'/notify/{}'.format(key), form.cleaned_data)
# No attachments allowed
assert response.status_code == 400
assert mock_notify.call_count == 0
# Reset our mock object
mock_notify.reset_mock()
# Test Webhooks
with mock.patch('requests.post') as mock_post:
# Response object
response = mock.Mock()
response.status_code = requests.codes.ok
mock_post.return_value = response
with override_settings(
APPRISE_WEBHOOK_URL='http://localhost/webhook/'):
# Preare our form data
form_data = {
'body': 'test notifiction',
}
# At a minimum, just a body is required
form = NotifyForm(data=form_data)
assert form.is_valid()
# Required to prevent None from being passed into
# self.client.post()
del form.cleaned_data['attachment']
# Send our notification
response = self.client.post(
'/notify/{}'.format(key), form.cleaned_data)
# Test our results
assert response.status_code == 200
assert mock_notify.call_count == 1
assert mock_post.call_count == 1
# Reset our mock object
mock_notify.reset_mock()
@mock.patch('requests.post')
def test_notify_with_tags(self, mock_post):
"""
Test notification handling when setting tags
@ -170,7 +310,7 @@ class NotifyTests(SimpleTestCase):
assert response['message'] == form_data['body']
assert response['type'] == apprise.NotifyType.WARNING
@patch('requests.post')
@mock.patch('requests.post')
def test_notify_with_tags_via_apprise(self, mock_post):
"""
Test notification handling when setting tags via the Apprise CLI
@ -272,7 +412,7 @@ class NotifyTests(SimpleTestCase):
assert response['message'] == form_data['body']
assert response['type'] == apprise.NotifyType.WARNING
@patch('requests.post')
@mock.patch('requests.post')
def test_advanced_notify_with_tags(self, mock_post):
"""
Test advanced notification handling when setting tags
@ -474,7 +614,7 @@ class NotifyTests(SimpleTestCase):
# We'll trigger on 2 entries
assert mock_post.call_count == 0
@patch('apprise.NotifyBase.notify')
@mock.patch('apprise.NotifyBase.notify')
def test_partial_notify_by_loaded_urls(self, mock_notify):
"""
Test notification handling when one or more of the services
@ -523,7 +663,7 @@ class NotifyTests(SimpleTestCase):
assert response.status_code == 424
assert mock_notify.call_count == 2
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_notify_by_loaded_urls_with_json(self, mock_notify):
"""
Test adding a simple notification and notifying it using JSON
@ -622,7 +762,7 @@ class NotifyTests(SimpleTestCase):
}
# Test the handling of underlining disk/write exceptions
with patch('gzip.open') as mock_open:
with mock.patch('gzip.open') as mock_open:
mock_open.side_effect = OSError()
# We'll fail to write our key now
response = self.client.post(
@ -746,7 +886,7 @@ class NotifyTests(SimpleTestCase):
assert mock_notify.call_count == 1
assert response['content-type'] == 'text/html'
@patch('apprise.plugins.NotifyEmail.NotifyEmail.send')
@mock.patch('apprise.plugins.NotifyEmail.NotifyEmail.send')
def test_notify_with_filters(self, mock_send):
"""
Test workings of APPRISE_DENY_SERVICES and APPRISE_ALLOW_SERVICES
@ -904,7 +1044,7 @@ class NotifyTests(SimpleTestCase):
apprise.common.NOTIFY_SCHEMA_MAP['mailto'].enabled is True
@override_settings(APPRISE_RECURSION_MAX=1)
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_stateful_notify_recursion(self, mock_notify):
"""
Test recursion an id header details as part of post

View File

@ -23,9 +23,11 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from django.test import SimpleTestCase
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test.utils import override_settings
from unittest.mock import patch
from unittest import mock
from ..forms import NotifyByUrlForm
import requests
import json
import apprise
@ -35,7 +37,7 @@ class StatelessNotifyTests(SimpleTestCase):
Test stateless notifications
"""
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_notify(self, mock_notify):
"""
Test sending a simple notification
@ -78,8 +80,102 @@ class StatelessNotifyTests(SimpleTestCase):
assert response.status_code == 200
assert mock_notify.call_count == 1
# Reset our mock object
mock_notify.reset_mock()
# Test Headers
for level in ('CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG',
'TRACE', 'INVALID'):
form_data = {
'urls': 'mailto://user:pass@hotmail.com',
'body': 'test notifiction',
'format': apprise.NotifyFormat.MARKDOWN,
}
attach_data = {
'attachment': SimpleUploadedFile(
"attach.txt", b"content here", content_type="text/plain")
}
# At a minimum, just a body is required
form = NotifyByUrlForm(form_data, attach_data)
assert form.is_valid()
# Prepare our header
headers = {
'HTTP_X-APPRISE-LOG-LEVEL': level,
}
# Send our notification
response = self.client.post(
'/notify', form.cleaned_data, **headers)
assert response.status_code == 200
assert mock_notify.call_count == 1
# Reset our mock object
mock_notify.reset_mock()
# Long Filename
attach_data = {
'attachment': SimpleUploadedFile(
"{}.txt".format('a' * 2000),
b"content here", content_type="text/plain")
}
# At a minimum, just a body is required
form = NotifyByUrlForm(form_data, attach_data)
assert form.is_valid()
# Send our notification
response = self.client.post('/notify', form.cleaned_data)
# We fail because the filename is too long
assert response.status_code == 400
assert mock_notify.call_count == 0
# Reset our mock object
mock_notify.reset_mock()
# Test Webhooks
with mock.patch('requests.post') as mock_post:
# Response object
response = mock.Mock()
response.status_code = requests.codes.ok
mock_post.return_value = response
with override_settings(
APPRISE_WEBHOOK_URL='http://localhost/webhook/'):
# Preare our form data
form_data = {
'urls': 'mailto://user:pass@hotmail.com',
'body': 'test notifiction',
'format': apprise.NotifyFormat.MARKDOWN,
}
# At a minimum, just a body is required
form = NotifyByUrlForm(data=form_data)
assert form.is_valid()
# Required to prevent None from being passed into
# self.client.post()
del form.cleaned_data['attachment']
# Send our notification
response = self.client.post('/notify', form.cleaned_data)
# Test our results
assert response.status_code == 200
assert mock_notify.call_count == 1
assert mock_post.call_count == 1
# Reset our mock object
mock_notify.reset_mock()
# Reset our count
mock_notify.reset_mock()
form_data = {
'urls': 'mailto://user:pass@hotmail.com',
'body': 'test notifiction',
@ -89,7 +185,17 @@ class StatelessNotifyTests(SimpleTestCase):
form = NotifyByUrlForm(data=form_data)
assert not form.is_valid()
@patch('apprise.NotifyBase.notify')
# Required to prevent None from being passed into self.client.post()
del form.cleaned_data['attachment']
# Send our notification
response = self.client.post('/notify', form.cleaned_data)
# Test our results
assert response.status_code == 200
assert mock_notify.call_count == 1
@mock.patch('apprise.NotifyBase.notify')
def test_partial_notify(self, mock_notify):
"""
Test sending multiple notifications where one fails
@ -123,7 +229,7 @@ class StatelessNotifyTests(SimpleTestCase):
assert mock_notify.call_count == 2
@override_settings(APPRISE_RECURSION_MAX=1)
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_stateless_notify_recursion(self, mock_notify):
"""
Test recursion an id header details as part of post
@ -216,7 +322,7 @@ class StatelessNotifyTests(SimpleTestCase):
assert mock_notify.call_count == 0
@override_settings(APPRISE_STATELESS_URLS="mailto://user:pass@localhost")
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_notify_default_urls(self, mock_notify):
"""
Test fallback to default URLS if none were otherwise specified
@ -244,7 +350,7 @@ class StatelessNotifyTests(SimpleTestCase):
assert response.status_code == 200
assert mock_notify.call_count == 1
@patch('apprise.Apprise.notify')
@mock.patch('apprise.Apprise.notify')
def test_notify_by_loaded_urls_with_json(self, mock_notify):
"""
Test sending a simple notification using JSON
@ -358,7 +464,7 @@ class StatelessNotifyTests(SimpleTestCase):
assert response.status_code == 400
assert mock_notify.call_count == 0
@patch('apprise.plugins.NotifyJSON.NotifyJSON.send')
@mock.patch('apprise.plugins.NotifyJSON.NotifyJSON.send')
def test_notify_with_filters(self, mock_send):
"""
Test workings of APPRISE_DENY_SERVICES and APPRISE_ALLOW_SERVICES

View File

@ -0,0 +1,120 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from django.test import SimpleTestCase
from unittest import mock
from json import loads
import requests
from ..utils import send_webhook
from django.test.utils import override_settings
class WebhookTests(SimpleTestCase):
@mock.patch('requests.post')
def test_webhook_testing(self, mock_post):
"""
Test webhook handling
"""
# Response object
response = mock.Mock()
response.status_code = requests.codes.ok
mock_post.return_value = response
with override_settings(
APPRISE_WEBHOOK_URL='https://'
'user:pass@localhost/webhook'):
send_webhook({})
assert mock_post.call_count == 1
details = mock_post.call_args_list[0]
assert details[0][0] == 'https://localhost/webhook'
assert loads(details[1]['data']) == {}
assert 'User-Agent' in details[1]['headers']
assert 'Content-Type' in details[1]['headers']
assert details[1]['headers']['User-Agent'] == 'Apprise-API'
assert details[1]['headers']['Content-Type'] == 'application/json'
assert details[1]['auth'] == ('user', 'pass')
assert details[1]['verify'] is True
assert details[1]['params'] == {}
assert details[1]['timeout'] == (4.0, 4.0)
mock_post.reset_mock()
with override_settings(
APPRISE_WEBHOOK_URL='http://'
'user@localhost/webhook/here'
'?verify=False&key=value&cto=2.0&rto=1.0'):
send_webhook({})
assert mock_post.call_count == 1
details = mock_post.call_args_list[0]
assert details[0][0] == 'http://localhost/webhook/here'
assert loads(details[1]['data']) == {}
assert 'User-Agent' in details[1]['headers']
assert 'Content-Type' in details[1]['headers']
assert details[1]['headers']['User-Agent'] == 'Apprise-API'
assert details[1]['headers']['Content-Type'] == 'application/json'
assert details[1]['auth'] == ('user', None)
assert details[1]['verify'] is False
assert details[1]['params'] == {'key': 'value'}
assert details[1]['timeout'] == (2.0, 1.0)
mock_post.reset_mock()
with override_settings(APPRISE_WEBHOOK_URL='invalid'):
# Invalid wbhook defined
send_webhook({})
assert mock_post.call_count == 0
mock_post.reset_mock()
with override_settings(
APPRISE_WEBHOOK_URL='invalid://hostname'):
# Invalid wbhook defined
send_webhook({})
assert mock_post.call_count == 0
mock_post.reset_mock()
# A valid URL with a bad server response:
response.status_code = requests.codes.internal_server_error
mock_post.return_value = response
with override_settings(
APPRISE_WEBHOOK_URL='http://localhost'):
send_webhook({})
assert mock_post.call_count == 1
mock_post.reset_mock()
# A valid URL with a bad server response:
mock_post.return_value = None
mock_post.side_effect = requests.RequestException("error")
with override_settings(
APPRISE_WEBHOOK_URL='http://localhost'):
send_webhook({})
assert mock_post.call_count == 1

View File

@ -32,7 +32,8 @@ import apprise
import hashlib
import errno
import base64
import requests
from json import dumps
from django.conf import settings
# import the logging library
@ -76,6 +77,7 @@ class Attachment(apprise.attachment.AttachFile):
Initialize our attachment
"""
self._filename = filename
self.delete = delete
try:
os.makedirs(settings.APPRISE_ATTACH_DIR, exist_ok=True)
@ -96,7 +98,6 @@ class Attachment(apprise.attachment.AttachFile):
filename, settings.APPRISE_ATTACH_DIR))
self._path = path
self.delete = delete
# Prepare our item
super().__init__(path=self._path, name=filename)
@ -144,11 +145,13 @@ def parse_attachments(attachment_payload, files_request):
0 if not isinstance(files_request, dict) else len(files_request),
])
if settings.APPRISE_MAX_ATTACHMENTS > 0 and \
count > settings.APPRISE_MAX_ATTACHMENTS:
if settings.APPRISE_MAX_ATTACHMENTS <= 0 or \
(settings.APPRISE_MAX_ATTACHMENTS > 0 and
count > settings.APPRISE_MAX_ATTACHMENTS):
raise ValueError(
"There is a maximum of %d attachments" %
settings.APPRISE_MAX_ATTACHMENTS)
settings.APPRISE_MAX_ATTACHMENTS
if settings.APPRISE_MAX_ATTACHMENTS > 0 else 0)
if isinstance(attachment_payload, (tuple, list)):
for no, entry in enumerate(attachment_payload, start=1):
@ -583,3 +586,59 @@ def apply_global_filters():
logger.warning(
'APPRISE_DENY_SERVICES plugin %s:// was not found -'
' ignoring.', name)
def send_webhook(payload):
"""
POST our webhook results
"""
# Prepare HTTP Headers
headers = {
'User-Agent': 'Apprise-API',
'Content-Type': 'application/json',
}
if not apprise.utils.VALID_URL_RE.match(
settings.APPRISE_WEBHOOK_URL):
logger.warning(
'The Apprise Webhook Result URL is not a valid web based URI')
return
# Parse our URL
results = apprise.URLBase.parse_url(settings.APPRISE_WEBHOOK_URL)
if not results:
logger.warning('The Apprise Webhook Result URL is not parseable')
return
if results['schema'] not in ('http', 'https'):
logger.warning(
'The Apprise Webhook Result URL is not using the HTTP protocol')
return
# Load our URL
base = apprise.URLBase(**results)
# Our Query String Dictionary; we use this to track arguments
# specified that aren't otherwise part of this class
params = {k: v for k, v in results.get('qsd', {}).items()
if k not in base.template_args}
try:
requests.post(
base.request_url,
data=dumps(payload),
params=params,
headers=headers,
auth=base.request_auth,
verify=base.verify_certificate,
timeout=base.request_timeout,
)
except requests.RequestException as e:
logger.warning(
'A Connection error occurred sending the Apprise Webhook '
'results to %s.' % base.url(privacy=True))
logger.debug('Socket Exception: %s' % str(e))
return

View File

@ -37,6 +37,7 @@ from django.core.serializers.json import DjangoJSONEncoder
from .utils import parse_attachments
from .utils import ConfigCache
from .utils import apply_global_filters
from .utils import send_webhook
from .forms import AddByUrlForm
from .forms import AddByConfigForm
from .forms import NotifyForm
@ -44,13 +45,11 @@ from .forms import NotifyByUrlForm
from .forms import CONFIG_FORMATS
from .forms import AUTO_DETECT_CONFIG_KEYWORD
import logging
import apprise
import json
import re
# import the logging library
import logging
# Get an instance of a logger
logger = logging.getLogger('django')
@ -557,7 +556,7 @@ class GetView(View):
@method_decorator((gzip_page, never_cache), name='dispatch')
class NotifyView(View):
"""
A Django view for sending a notification
A Django view for sending a notification in a stateful manner
"""
def post(self, request, key):
"""
@ -608,9 +607,15 @@ class NotifyView(View):
# Handle Attachments
attach = None
if 'attachments' in content or request.FILES:
attach = parse_attachments(
content.get('attachments'), request.FILES)
if 'attachment' in content or request.FILES:
try:
attach = parse_attachments(
content.get('attachment'), request.FILES)
except (TypeError, ValueError):
return HttpResponse(
_('Bad attachment'),
status=ResponseCode.bad_request)
#
# Allow 'tag' value to be specified as part of the URL parameters
@ -823,51 +828,45 @@ class NotifyView(View):
level = request.headers.get(
'X-Apprise-Log-Level',
settings.LOGGING['loggers']['apprise']['level']).upper()
if level not in (
'CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG', 'TRACE'):
level = settings.LOGGING['loggers']['apprise']['level'].upper()
# Convert level to it's integer value
if level == 'CRITICAL':
level = logging.CRITICAL
elif level == 'ERROR':
level = logging.ERROR
elif level == 'WARNING':
level = logging.WARNING
elif level == 'INFO':
level = logging.INFO
elif level == 'DEBUG':
level = logging.DEBUG
elif level == 'TRACE':
level = logging.DEBUG - 1
# Initialize our response object
response = None
if level in ('CRITICAL', 'ERROR' 'WARNING', 'INFO', 'DEBUG'):
level = getattr(apprise.logging, level)
esc = '<!!-!ESC!-!!>'
esc = '<!!-!ESC!-!!>'
fmt = '<li class="log_%(levelname)s">' \
'<div class="log_time">%(asctime)s</div>' \
'<div class="log_level">%(levelname)s</div>' \
f'<div class="log_msg">{esc}%(message)s{esc}</div></li>' \
if content_type == 'text/html' else \
settings.LOGGING['formatters']['standard']['format']
# Format is only updated if the content_type is html
fmt = '<li class="log_%(levelname)s">' \
'<div class="log_time">%(asctime)s</div>' \
'<div class="log_level">%(levelname)s</div>' \
f'<div class="log_msg">{esc}%(message)s{esc}</div></li>' \
if content_type == 'text/html' else \
settings.LOGGING['formatters']['standard']['format']
# Now specify our format (and over-ride the default):
with apprise.LogCapture(level=level, fmt=fmt) as logs:
# Perform our notification at this point
result = a_obj.notify(
content.get('body'),
title=content.get('title', ''),
notify_type=content.get('type', apprise.NotifyType.INFO),
tag=content.get('tag'),
attach=attach,
)
if content_type == 'text/html':
# Iterate over our entries so that we can prepare to escape
# things to be presented as HTML
esc = re.escape(esc)
entries = re.findall(
r'(?P<head><li .+?){}(?P<to_escape>.*?)'
r'{}(?P<tail>.+li>$)(?=$|<li .+{})'.format(
esc, esc, esc), logs.getvalue(),
re.DOTALL)
# Wrap logs in `<ul>` tag and escape our message body:
response = '<ul class="logs">{}</ul>'.format(
''.join([e[0] + escape(e[1]) + e[2] for e in entries]))
else: # content_type == 'text/plain'
response = logs.getvalue()
else:
# Perform our notification at this point without logging
# Now specify our format (and over-ride the default):
with apprise.LogCapture(level=level, fmt=fmt) as logs:
# Perform our notification at this point
result = a_obj.notify(
content.get('body'),
title=content.get('title', ''),
@ -876,6 +875,33 @@ class NotifyView(View):
attach=attach,
)
if content_type == 'text/html':
# Iterate over our entries so that we can prepare to escape
# things to be presented as HTML
esc = re.escape(esc)
entries = re.findall(
r'(?P<head><li .+?){}(?P<to_escape>.*?)'
r'{}(?P<tail>.+li>$)(?=$|<li .+{})'.format(
esc, esc, esc), logs.getvalue(),
re.DOTALL)
# Wrap logs in `<ul>` tag and escape our message body:
response = '<ul class="logs">{}</ul>'.format(
''.join([e[0] + escape(e[1]) + e[2] for e in entries]))
else: # content_type == 'text/plain'
response = logs.getvalue()
if settings.APPRISE_WEBHOOK_URL:
webhook_payload = {
'source': request.META['REMOTE_ADDR'],
'status': 0 if result else 1,
'output': response,
}
# Send our webhook (pass or fail)
send_webhook(webhook_payload)
if not result:
# If at least one notification couldn't be sent; change up
# the response to a 424 error code
@ -1013,18 +1039,76 @@ class StatelessNotifyView(View):
# Handle Attachments
attach = None
if 'attachments' in content or request.FILES:
attach = parse_attachments(
content.get('attachments'), request.FILES)
if 'attachment' in content or request.FILES:
try:
attach = parse_attachments(
content.get('attachment'), request.FILES)
# Perform our notification at this point
result = a_obj.notify(
content.get('body'),
title=content.get('title', ''),
notify_type=content.get('type', apprise.NotifyType.INFO),
tag='all',
attach=attach,
)
except (TypeError, ValueError):
return HttpResponse(
_('Bad attachment'),
status=ResponseCode.bad_request)
# Acquire our log level from headers if defined, otherwise use
# the global one set in the settings
level = request.headers.get(
'X-Apprise-Log-Level',
settings.LOGGING['loggers']['apprise']['level']).upper()
if level not in (
'CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG', 'TRACE'):
level = settings.LOGGING['loggers']['apprise']['level'].upper()
# Convert level to it's integer value
if level == 'CRITICAL':
level = logging.CRITICAL
elif level == 'ERROR':
level = logging.ERROR
elif level == 'WARNING':
level = logging.WARNING
elif level == 'INFO':
level = logging.INFO
elif level == 'DEBUG':
level = logging.DEBUG
elif level == 'TRACE':
level = logging.DEBUG - 1
if settings.APPRISE_WEBHOOK_URL:
fmt = settings.LOGGING['formatters']['standard']['format']
with apprise.LogCapture(level=level, fmt=fmt) as logs:
# Perform our notification at this point
result = a_obj.notify(
content.get('body'),
title=content.get('title', ''),
notify_type=content.get('type', apprise.NotifyType.INFO),
tag='all',
attach=attach,
)
response = logs.getvalue()
webhook_payload = {
'source': request.META['REMOTE_ADDR'],
'status': 0 if result else 1,
'output': response,
}
# Send our webhook (pass or fail)
send_webhook(webhook_payload)
else:
# Perform our notification at this point
result = a_obj.notify(
content.get('body'),
title=content.get('title', ''),
notify_type=content.get('type', apprise.NotifyType.INFO),
tag='all',
attach=attach,
)
if not result:
# If at least one notification couldn't be sent; change up the

View File

@ -127,6 +127,11 @@ STATIC_URL = BASE_URL + '/s/'
APPRISE_DEFAULT_THEME = \
os.environ.get('APPRISE_DEFAULT_THEME', SiteTheme.LIGHT)
# Webhook that is posted to upon executed results
# Set it to something like https://myserver.com/path/
# Requets are done as a POST
APPRISE_WEBHOOK_URL = os.environ.get('APPRISE_WEBHOOK_URL', '')
# The location to store Apprise configuration files
APPRISE_CONFIG_DIR = os.environ.get(
'APPRISE_CONFIG_DIR', os.path.join(BASE_DIR, 'var', 'config'))

View File

@ -10,13 +10,16 @@
# apprise @ git+https://github.com/caronc/apprise@custom-tag-or-version
## 3. The below grabs our stable version (generally the best choice):
apprise == 1.5.0
apprise == 1.6.0
## Apprise API Minimum Requirements
django
gevent
gunicorn
## for webhook support
requests
## 3rd Party Service support
paho-mqtt
gntp