diff --git a/changedetectionio/blueprint/ui/notification.py b/changedetectionio/blueprint/ui/notification.py
index a1585eede..2d6bf6bdc 100644
--- a/changedetectionio/blueprint/ui/notification.py
+++ b/changedetectionio/blueprint/ui/notification.py
@@ -26,6 +26,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# Watch_uuid could be unset in the case it`s used in tag editor, global settings
import apprise
+ from urllib.parse import urlparse
from changedetectionio.notification.apprise_plugin.assets import apprise_asset
# Necessary so that we import our custom handlers
@@ -39,8 +40,8 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# Use an existing random one on the global/main settings form
if not watch_uuid and is_global_settings_form and datastore.data.get('watching'):
- logger.debug(f"Send test notification - Choosing random Watch {watch_uuid}")
watch_uuid = random.choice(list(datastore.data['watching'].keys()))
+ logger.debug(f"Send test notification - Chose random watch UUID: {watch_uuid}")
if is_group_settings_form and datastore.data.get('watching'):
logger.debug(f"Send test notification - Choosing random Watch from group {watch_uuid}")
@@ -57,11 +58,24 @@ def construct_blueprint(datastore: ChangeDetectionStore):
watch = datastore.data['watching'].get(watch_uuid)
notification_urls = []
- if send_as_null_test:
- notification_urls.append('null://null-test-just-to-render-everything-on-the-same-codepath-and-get-preview')
- if request.form.get('notification_urls'):
- notification_urls += request.form['notification_urls'].strip().splitlines()
+ if send_as_null_test:
+ test_schema = ''
+ try:
+ if request.form.get('notification_urls') and '://' in request.form.get('notification_urls'):
+ first_test_notification_url = request.form['notification_urls'].strip().splitlines()[0]
+ test_schema = urlparse(first_test_notification_url).scheme.lower().strip()
+ except Exception as e:
+ logger.error(f"Error trying to get a test schema based on the first notification_url {str(e)}")
+
+ notification_urls = [
+ # Null lets us do the whole chain of the same code without any extra repeated code
+ f'null://null-test-just-to-render-everything-on-the-same-codepath-and-get-preview?test_schema={test_schema}'
+ ]
+
+ else:
+ if request.form.get('notification_urls'):
+ notification_urls += request.form['notification_urls'].strip().splitlines()
if not notification_urls:
logger.debug("Test notification - Trying by group/tag in the edit form if available")
@@ -83,7 +97,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
for n_url in notification_urls:
if len(n_url.strip()):
if not apobj.add(n_url):
- return f'Error: {n_url} is not a valid AppRise URL.'
+ return make_response(f'Error: {n_url} is not a valid AppRise URL.', 400)
try:
# use the same as when it is triggered, but then override it with the form test values
diff --git a/changedetectionio/notification/handler.py b/changedetectionio/notification/handler.py
index 3ea219e1c..5a6e3f8a9 100644
--- a/changedetectionio/notification/handler.py
+++ b/changedetectionio/notification/handler.py
@@ -5,7 +5,7 @@ import apprise
from loguru import logger
from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
from changedetectionio.safe_jinja import render as jinja_render
-
+from urllib.parse import urlparse
def _populate_notification_tokens(n_object, datastore):
"""
@@ -93,6 +93,40 @@ def _populate_notification_tokens(n_object, datastore):
if watch:
n_object.update(watch.extra_notification_token_values())
+def scan_notification_file_templates(url, datastore, n_body, notification_parameters):
+ import glob
+ from urllib.parse import urlparse, parse_qs
+
+ try:
+ scheme = urlparse(url).scheme.lower().strip()
+
+ # schema could be overriden dynamically
+ if scheme == 'null' and 'test_schema=' in url:
+ scheme = parse_qs(urlparse(url).query).get("test_schema", [None])[0]
+
+ logger.debug(f"Looking for '{scheme}' notification wrapper templates...")
+
+ # Try exact match first, then wildcard matches
+ candidates = [
+ os.path.join(datastore.datastore_path, f"notification-wrapper-{scheme}.html"),
+ *[f for f in glob.glob(os.path.join(datastore.datastore_path, "notification-wrapper-*--.html"))
+ if scheme.startswith(os.path.basename(f).replace("notification-wrapper-", "").replace("--.html", ""))]
+ ]
+
+ for tpl_name in candidates:
+ if os.path.isfile(tpl_name):
+ template_params = notification_parameters.copy()
+ template_params['notification_body'] = n_body
+
+ with open(tpl_name, 'r', encoding='utf-8') as f:
+ logger.info(f"Using HTML notification template wrapper from '{tpl_name}'")
+ return jinja_render(template_str=f.read(), **template_params)
+
+ except Exception as e:
+ logger.warning(f"Failed to load notification template: {e}")
+
+ return None
+
def process_notification(n_object, datastore):
from . import default_notification_format_for_watch, default_notification_format, valid_notification_formats
# be sure its registered
@@ -138,42 +172,27 @@ def process_notification(n_object, datastore):
if not n_object.get('notification_urls'):
return None
- # Check for notification.html template in datastore directory
- notification_template_path = os.path.join(datastore.datastore_path, 'notification.html')
- notification_template = None
- if os.path.exists(notification_template_path):
- try:
- with open(notification_template_path, 'r', encoding='utf-8') as f:
- notification_template = f.read()
- logger.info(f"Using notification template from {notification_template_path}")
- except Exception as e:
- logger.warning(f"Failed to load notification template {notification_template_path}: {e}")
-
with apprise.LogCapture(level=apprise.logging.DEBUG) as logs:
for url in n_object['notification_urls']:
+ # Commented out is OK
+ if url.startswith('#') or not url or not url.strip():
+ logger.trace(f"Skipping notification URL - '{url}'")
+ continue
# Get the notification body from datastore
n_body = jinja_render(template_str=n_object.get('notification_body', ''), **notification_parameters)
-
- # Apply notification template wrapper if it exists (the one from the disk)
- if notification_template:
- template_params = notification_parameters.copy()
- template_params['notification_body'] = n_body
- template_params['notification_url_current'] = url
- n_body = jinja_render(template_str=notification_template, **template_params)
-
if n_object.get('notification_format', '').startswith('HTML'):
n_body = n_body.replace("\n", '
')
n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)
- url = url.strip()
- if url.startswith('#'):
- logger.trace(f"Skipping commented out notification URL - {url}")
- continue
+ n_body_from_file_template = scan_notification_file_templates(url=url,
+ datastore=datastore,
+ n_body=n_body,
+ notification_parameters=notification_parameters)
+ if n_body_from_file_template:
+ n_body = n_body_from_file_template
+
- if not url:
- logger.warning(f"Process Notification: skipping empty notification URL.")
- continue
logger.info(f">> Process Notification: AppRise notifying {url}")
url = jinja_render(template_str=url, **notification_parameters)
diff --git a/changedetectionio/notification/notification-wrapper-HTML-mail--.html b/changedetectionio/notification/notification-wrapper-HTML-mail--.html
new file mode 100644
index 000000000..b8f01bdaa
--- /dev/null
+++ b/changedetectionio/notification/notification-wrapper-HTML-mail--.html
@@ -0,0 +1,15 @@
+{# Copy this to your data-store directory if you wish to enable it for HTML style notifications, applies to all as a wrapper :) #}
+
+
A change was detected on your web page watch for
{{ watch_html_link }}.
+ +[ view history ] [ pause checks ] [ mute notifications ] + +