WIP

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

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", '<br>')
             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)

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 :) #}
+<html>
+<body>
+Hello,<br>
+
+<p>A change was detected on your web page watch for <p>{{ watch_html_link }}.</p>
+
+[ view history ] [ pause checks ] [ mute notifications ]
+
+<div>
+    {{  notification_body }}
+</div>
+
+</body>
+</html>

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
+
+

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("""<html>
+        <body id="notification-wrapper">
+        
+        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)
+        </body>
+        """)
+
+
     # CUSTOM JSON BODY CHECK for POST://
     set_original_response()
     # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#header-manipulation