From c2eb736051a754f2d8866b4e81935f96be310cf9 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Wed, 10 Sep 2025 14:54:24 +0200 Subject: [PATCH] WIP --- .../blueprint/ui/notification.py | 26 +++++-- changedetectionio/notification/handler.py | 73 ++++++++++++------- .../notification-wrapper-HTML-mail--.html | 15 ++++ changedetectionio/notification/readme.md | 17 +++++ changedetectionio/tests/test_notification.py | 14 ++++ 5 files changed, 112 insertions(+), 33 deletions(-) create mode 100644 changedetectionio/notification/notification-wrapper-HTML-mail--.html create mode 100644 changedetectionio/notification/readme.md 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 :) #} + + +Hello,
+ +

A change was detected on your web page watch for

{{ watch_html_link }}.

+ +[ view history ] [ pause checks ] [ mute notifications ] + +
+ {{ notification_body }} +
+ + + \ No newline at end of file diff --git a/changedetectionio/notification/readme.md b/changedetectionio/notification/readme.md new file mode 100644 index 000000000..1729d8bfd --- /dev/null +++ b/changedetectionio/notification/readme.md @@ -0,0 +1,17 @@ +## Notification syntax + +All notifications use the https://github.com/caronc/apprise syntax, there are some custom ones such as `posts` etc for general web-services usability. + + +## Template file notification wrappers + +You can by default wrap all notifications by creating a `notification-wrapper-HTML-schema.html` in your datastore directory. + +For example + +You can use "`--`" in the filename where the _schema_ is to symbolize a wildcard. For example `notification-wrapper-HTML-mail--.html` would +apply to `mail://` `mailto://` etc etc + +See is `notification-wrapper-HTML-mail--.html` which applies to `mail://`, `mailto://foobar..` etc notifications + + diff --git a/changedetectionio/tests/test_notification.py b/changedetectionio/tests/test_notification.py index 8640d2cca..4fcbab79f 100644 --- a/changedetectionio/tests/test_notification.py +++ b/changedetectionio/tests/test_notification.py @@ -289,6 +289,20 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me # test_endpoint - that sends the contents of a file # test_notification_endpoint - that takes a POST and writes it to file (test-datastore/notification.txt) + # Drop in a custom wrapping template + with open("test-datastore/notification-wrapper.html", "w" ) as f: + f.write(""" + + + A change was detected at {{watch_html_link}} + 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) + + """) + + # CUSTOM JSON BODY CHECK for POST:// set_original_response() # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#header-manipulation