Notifications fixes (#3534) #3531 #3530 #3529

2025-10-30 14:17:40 +00:00 · 2025-10-24 20:40:15 +02:00
parent 181d32e82a
commit e233d52931
8 changed files with 368 additions and 108 deletions
--- a/changedetectionio/diff.py
+++ b/changedetectionio/diff.py
@@ -1,8 +1,10 @@
 import difflib
 from typing import List, Iterator, Union

-HTML_REMOVED_STYLE = "background-color: #fadad7; color: #b30000;"
-HTML_ADDED_STYLE = "background-color: #eaf2c2; color: #406619;"
+# https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050
+HTML_REMOVED_STYLE = "background-color: #ffebe9; color: #82071e"
+HTML_ADDED_STYLE = "background-color: #dafbe1; color: #116329;"
+HTML_CHANGED_STYLE = "background-color: #ffd8b5; color: #953800;"

 # These get set to html or telegram type or discord compatible or whatever in handler.py
 REMOVED_PLACEMARKER_OPEN = '<<<removed_PLACEMARKER_OPEN'
--- a/changedetectionio/notification/apprise_plugin/custom_handlers.py
+++ b/changedetectionio/notification/apprise_plugin/custom_handlers.py
@@ -1,10 +1,61 @@
+"""
+Custom Apprise HTTP Handlers with format= Parameter Support
+
+IMPORTANT: This module works around a limitation in Apprise's @notify decorator.
+
+THE PROBLEM:
+-------------
+When using Apprise's @notify decorator to create custom notification handlers, the
+decorator creates a CustomNotifyPlugin that uses parse_url(..., simple=True) to parse
+URLs. This simple parsing mode does NOT extract the format= query parameter from the URL
+and set it as a top-level parameter that NotifyBase.__init__ can use to set notify_format.
+
+As a result:
+1. URL: post://example.com/webhook?format=html
+2. Apprise parses this and sees format=html in qsd (query string dictionary)
+3. But it does NOT extract it and pass it to NotifyBase.__init__
+4. NotifyBase defaults to notify_format=TEXT
+5. When you call apobj.notify(body="<html>...", body_format="html"):
+   - Apprise sees: input format = html, output format (notify_format) = text
+   - Apprise calls convert_between("html", "text", body)
+   - This strips all HTML tags, leaving only plain text
+6. Your custom handler receives stripped plain text instead of HTML
+
+THE SOLUTION:
+-------------
+Instead of using the @notify decorator directly, we:
+1. Manually register custom plugins using plugins.N_MGR.add()
+2. Create a CustomHTTPHandler class that extends CustomNotifyPlugin
+3. Override __init__ to extract format= from qsd and set it as kwargs['format']
+4. Call NotifyBase.__init__ which properly sets notify_format from kwargs['format']
+5. Set up _default_args like CustomNotifyPlugin does for compatibility
+
+This ensures that when format=html is in the URL:
+- notify_format is set to HTML
+- Apprise sees: input format = html, output format = html
+- No conversion happens (convert_between returns content unchanged)
+- Your custom handler receives the original HTML intact
+
+TESTING:
+--------
+To verify this works:
+>>> apobj = apprise.Apprise()
+>>> apobj.add('post://localhost:5005/test?format=html')
+>>> for server in apobj:
+...     print(server.notify_format)  # Should print: html (not text)
+>>> apobj.notify(body='<span>Test</span>', body_format='html')
+# Your handler should receive '<span>Test</span>' not 'Test'
+"""
+
 import json
 import re
 from urllib.parse import unquote_plus

 import requests
-from apprise.decorators import notify
-from apprise.utils.parse import parse_url as apprise_parse_url
+from apprise import plugins
+from apprise.decorators.base import CustomNotifyPlugin
+from apprise.utils.parse import parse_url as apprise_parse_url, url_assembly
+from apprise.utils.logic import dict_full_update
 from loguru import logger
 from requests.structures import CaseInsensitiveDict

@@ -12,13 +63,66 @@ SUPPORTED_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head"}


 def notify_supported_methods(func):
+    """Register custom HTTP method handlers that properly support format= parameter."""
    for method in SUPPORTED_HTTP_METHODS:
-        func = notify(on=method)(func)
-        # Add support for https, for each supported http method
-        func = notify(on=f"{method}s")(func)
+        _register_http_handler(method, func)
+        _register_http_handler(f"{method}s", func)
    return func


+def _register_http_handler(schema, send_func):
+    """Register a custom HTTP handler that extracts format= from URL query parameters."""
+
+    # Parse base URL
+    base_url = f"{schema}://"
+    base_args = apprise_parse_url(base_url, default_schema=schema, verify_host=False, simple=True)
+
+    class CustomHTTPHandler(CustomNotifyPlugin):
+        secure_protocol = schema
+        service_name = f"Custom HTTP - {schema.upper()}"
+        _base_args = base_args
+
+        def __init__(self, **kwargs):
+            # Extract format from qsd and set it as a top-level kwarg
+            # This allows NotifyBase.__init__ to properly set notify_format
+            if 'qsd' in kwargs and 'format' in kwargs['qsd']:
+                kwargs['format'] = kwargs['qsd']['format']
+
+            # Call NotifyBase.__init__ (skip CustomNotifyPlugin.__init__)
+            super(CustomNotifyPlugin, self).__init__(**kwargs)
+
+            # Set up _default_args like CustomNotifyPlugin does
+            self._default_args = {}
+            kwargs.pop("secure", None)
+            dict_full_update(self._default_args, self._base_args)
+            dict_full_update(self._default_args, kwargs)
+            self._default_args["url"] = url_assembly(**self._default_args)
+
+        __send = staticmethod(send_func)
+
+        def send(self, body, title="", notify_type="info", *args, **kwargs):
+            """Call the custom send function."""
+            try:
+                result = self.__send(
+                    body, title, notify_type,
+                    *args,
+                    meta=self._default_args,
+                    **kwargs
+                )
+                return True if result is None else bool(result)
+            except Exception as e:
+                self.logger.warning(f"Exception in custom HTTP handler: {e}")
+                return False
+
+    # Register the plugin
+    plugins.N_MGR.add(
+        plugin=CustomHTTPHandler,
+        schemas=schema,
+        send_func=send_func,
+        url=base_url,
+    )
+
+
 def _get_auth(parsed_url: dict) -> str | tuple[str, str]:
    user: str | None = parsed_url.get("user")
    password: str | None = parsed_url.get("password")
@@ -74,6 +178,8 @@ def apprise_http_custom_handler(
    *args,
    **kwargs,
 ) -> bool:
+
+
    url: str = meta.get("url")
    schema: str = meta.get("schema")
    method: str = re.sub(r"s$", "", schema).upper()
--- a/changedetectionio/notification/handler.py
+++ b/changedetectionio/notification/handler.py
@@ -8,9 +8,10 @@ from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
 from .apprise_plugin.custom_handlers import SUPPORTED_HTTP_METHODS
 from ..diff import HTML_REMOVED_STYLE, REMOVED_PLACEMARKER_OPEN, REMOVED_PLACEMARKER_CLOSED, ADDED_PLACEMARKER_OPEN, HTML_ADDED_STYLE, \
    ADDED_PLACEMARKER_CLOSED, CHANGED_INTO_PLACEMARKER_OPEN, CHANGED_INTO_PLACEMARKER_CLOSED, CHANGED_PLACEMARKER_OPEN, \
-    CHANGED_PLACEMARKER_CLOSED
+    CHANGED_PLACEMARKER_CLOSED, HTML_CHANGED_STYLE
 from ..notification_service import NotificationContextData

+CUSTOM_LINEBREAK_PLACEHOLDER='$$BR$$'

 def markup_text_links_to_html(body):
    """
@@ -156,16 +157,17 @@ def apply_service_tweaks(url, n_body, n_title, requested_output_format):

    # Is not discord/tgram and they want htmlcolor
    elif requested_output_format == 'htmlcolor':
-        n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}">')
+        # https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050
+        n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}" role="deletion" aria-label="Removed text" title="Removed text">')
        n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, f'</span>')
-        n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}">')
+        n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}" role="insertion" aria-label="Added text" title="Added text">')
        n_body = n_body.replace(ADDED_PLACEMARKER_CLOSED, f'</span>')
        # Handle changed/replaced lines (old → new)
-        n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}">')
+        n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'<span style="{HTML_CHANGED_STYLE}" role="note" aria-label="Changed text" title="Changed text">')
        n_body = n_body.replace(CHANGED_PLACEMARKER_CLOSED, f'</span>')
-        n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}">')
+        n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'<span style="{HTML_CHANGED_STYLE}" role="note" aria-label="Changed into" title="Changed into">')
        n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_CLOSED, f'</span>')
-        n_body = n_body.replace("\n", '<br>')
+        n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n')
    elif requested_output_format == 'html':
        n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ')
        n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, '')
@@ -175,7 +177,7 @@ def apply_service_tweaks(url, n_body, n_title, requested_output_format):
        n_body = n_body.replace(CHANGED_PLACEMARKER_CLOSED, f'')
        n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'(into) ')
        n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_CLOSED, f'')
-        n_body = n_body.replace("\n", '<br>')
+        n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n')

    else: #plaintext etc default
        n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ')
@@ -233,11 +235,6 @@ def process_notification(n_object: NotificationContextData, datastore):

    logger.trace(f"Complete notification body including Jinja and placeholders calculated in  {time.time() - now:.2f}s")

-    # If we have custom handlers, use invalid format to prevent conversion
-    # Otherwise use the proper format
-    if has_custom_handler:
-        input_format = 'raw-no-convert'
-
    # https://github.com/caronc/apprise/wiki/Development_LogCapture
    # Anything higher than or equal to WARNING (which covers things like Connection errors)
    # raise it as an exception
@@ -258,42 +255,15 @@ def process_notification(n_object: NotificationContextData, datastore):
    if not n_object.get('notification_urls'):
        return None

-    with apprise.LogCapture(level=apprise.logging.DEBUG) as logs:
+    with (apprise.LogCapture(level=apprise.logging.DEBUG) as logs):
        for url in n_object['notification_urls']:
-            parsed_url = urlparse(url)
-            prefix_add_to_url = '?' if not parsed_url.query else '&'

            # Get the notification body from datastore
            n_body = jinja_render(template_str=n_object.get('notification_body', ''), **notification_parameters)

-            if n_object.get('markup_text_to_html'):
+            if n_object.get('markup_text_links_to_html_links'):
                n_body = markup_text_links_to_html(body=n_body)

-            # This actually means we request "Markdown to HTML"
-            if requested_output_format == NotifyFormat.MARKDOWN.value:
-                output_format = NotifyFormat.HTML.value
-                input_format = NotifyFormat.MARKDOWN.value
-                if not 'format=' in url.lower():
-                    url = f"{url}{prefix_add_to_url}format={output_format}"
-
-            # Deviation from apprise.
-            # No conversion, its like they want to send raw HTML but we add linebreaks
-            elif requested_output_format == NotifyFormat.HTML.value:
-                # same in and out means apprise wont try to convert
-                input_format = output_format = NotifyFormat.HTML.value
-                if not 'format=' in url.lower():
-                    url = f"{url}{prefix_add_to_url}format={output_format}"
-
-            else:
-                # Nothing to be done, leave it as plaintext
-                # `body_format` Tell apprise what format the INPUT is in
-                # &format= in URL Tell apprise what format the OUTPUT should be in (it can convert between)
-                input_format = output_format = NotifyFormat.TEXT.value
-                if not 'format=' in url.lower():
-                    url = f"{url}{prefix_add_to_url}format={output_format}"
-
-            if has_custom_handler:
-                input_format='raw-no-convert'

            n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)

@@ -311,6 +281,41 @@ def process_notification(n_object: NotificationContextData, datastore):

            (url, n_body, n_title) = apply_service_tweaks(url=url, n_body=n_body, n_title=n_title, requested_output_format=requested_output_format_original)

+            apprise_input_format = "NO-THANKS-WE-WILL-MANAGE-ALL-OF-THIS"
+
+            if not 'format=' in url:
+                parsed_url = urlparse(url)
+                prefix_add_to_url = '?' if not parsed_url.query else '&'
+
+                # THIS IS THE TRICK HOW TO DISABLE APPRISE DOING WEIRD AUTO-CONVERSION WITH BREAKING BR TAGS ETC
+                if 'html' in requested_output_format:
+                    url = f"{url}{prefix_add_to_url}format={NotifyFormat.HTML.value}"
+                    apprise_input_format = NotifyFormat.HTML.value
+                elif 'text' in requested_output_format:
+                    url = f"{url}{prefix_add_to_url}format={NotifyFormat.TEXT.value}"
+                    apprise_input_format = NotifyFormat.TEXT.value
+
+                elif requested_output_format == NotifyFormat.MARKDOWN.value:
+                    # This actually means we request "Markdown to HTML", we want HTML output
+                    url = f"{url}{prefix_add_to_url}format={NotifyFormat.HTML.value}"
+                    requested_output_format = NotifyFormat.HTML.value
+                    apprise_input_format = NotifyFormat.MARKDOWN.value
+
+                # If it's a plaintext document, and they want HTML type email/alerts, so it needs to be escaped
+                watch_mime_type = n_object.get('watch_mime_type', '').lower()
+                if watch_mime_type and 'text/' in watch_mime_type and not 'html' in watch_mime_type:
+                    if 'html' in requested_output_format:
+                        from markupsafe import escape
+                        n_body = str(escape(n_body))
+
+                # Could have arrived at any stage, so we dont end up running .escape on it
+                if 'html' in requested_output_format:
+                    n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '<br>')
+                else:
+                    # Just incase
+                    n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '')
+
+
            apobj.add(url)

            sent_objs.append({'title': n_title,
@@ -320,9 +325,9 @@ def process_notification(n_object: NotificationContextData, datastore):
        apobj.notify(
            title=n_title,
            body=n_body,
-            # `body_format` Tell apprise what format the INPUT is in
+            # `body_format` Tell apprise what format the INPUT is in, specify a wrong/bad type and it will force skip conversion in apprise
            # &format= in URL Tell apprise what format the OUTPUT should be in (it can convert between)
-            body_format=input_format,
+            body_format=apprise_input_format,
            # False is not an option for AppRise, must be type None
            attach=n_object.get('screenshot', None)
        )
--- a/changedetectionio/notification_service.py
+++ b/changedetectionio/notification_service.py
@@ -31,7 +31,7 @@ class NotificationContextData(dict):
            'preview_url': None,
            'watch_tag': None,
            'watch_title': None,
-            'markup_text_to_html': False, # If automatic conversion of plaintext to HTML should happen
+            'markup_text_links_to_html_links': False, # If automatic conversion of plaintext to HTML should happen
        })

        # Apply any initial data passed in
@@ -131,6 +131,7 @@ class NotificationService:
            'uuid': watch.get('uuid') if watch else None,
            'watch_url': watch.get('url') if watch else None,
            'watch_uuid': watch.get('uuid') if watch else None,
+            'watch_mime_type': watch.get('content-type')
        })

        if watch:
@@ -228,7 +229,7 @@ class NotificationService:

        n_format = self.datastore.data['settings']['application'].get('notification_format', default_notification_format)
        filter_list = ", ".join(watch['include_filters'])
-        # @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_to_html' is not needed
+        # @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_links_to_html_links' is not needed
        body = f"""Hello,

 Your configured CSS/xPath filters of '{filter_list}' for {{{{watch_url}}}} did not appear on the page after {threshold} attempts.
@@ -244,7 +245,7 @@ Thanks - Your omniscient changedetection.io installation.
            'notification_title': 'Changedetection.io - Alert - CSS/xPath filter was not present in the page',
            'notification_body': body,
            'notification_format': n_format,
-            'markup_text_to_html': n_format.lower().startswith('html')
+            'markup_text_links_to_html_links': n_format.lower().startswith('html')
        })

        if len(watch['notification_urls']):
@@ -275,7 +276,7 @@ Thanks - Your omniscient changedetection.io installation.
        threshold = self.datastore.data['settings']['application'].get('filter_failure_notification_threshold_attempts')
        n_format = self.datastore.data['settings']['application'].get('notification_format', default_notification_format).lower()
        step = step_n + 1
-        # @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_to_html' is not needed
+        # @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_links_to_html_links' is not needed

        # {{{{ }}}} because this will be Jinja2 {{ }} tokens
        body = f"""Hello,
@@ -293,7 +294,7 @@ Thanks - Your omniscient changedetection.io installation.
            'notification_title': f"Changedetection.io - Alert - Browser step at position {step} could not be run",
            'notification_body': body,
            'notification_format': n_format,
-            'markup_text_to_html': n_format.lower().startswith('html')
+            'markup_text_links_to_html_links': n_format.lower().startswith('html')
        })

        if len(watch['notification_urls']):
--- a/changedetectionio/tests/smtp/smtp-test-server.py
+++ b/changedetectionio/tests/smtp/smtp-test-server.py
@@ -1,51 +1,76 @@
 #!/usr/bin/env python3
-import asyncio
+import threading
+import time
 from aiosmtpd.controller import Controller
-from aiosmtpd.smtp import SMTP
+from flask import Flask, Response

-# Accept a SMTP message and offer a way to retrieve the last message via TCP Socket
+# Accept a SMTP message and offer a way to retrieve the last message via HTTP

-last_received_message = b"Nothing"
+last_received_message = b"Nothing received yet."
+active_smtp_connections = 0
+smtp_lock = threading.Lock()


 class CustomSMTPHandler:
    async def handle_DATA(self, server, session, envelope):
-        global last_received_message
-        last_received_message = envelope.content
-        print('Receiving message from:', session.peer)
-        print('Message addressed from:', envelope.mail_from)
-        print('Message addressed to  :', envelope.rcpt_tos)
-        print('Message length        :', len(envelope.content))
-        print(envelope.content.decode('utf8'))
-        return '250 Message accepted for delivery'
+        global last_received_message, active_smtp_connections
+
+        with smtp_lock:
+            active_smtp_connections += 1
+
+        try:
+            last_received_message = envelope.content
+            print('Receiving message from:', session.peer)
+            print('Message addressed from:', envelope.mail_from)
+            print('Message addressed to  :', envelope.rcpt_tos)
+            print('Message length        :', len(envelope.content))
+            print('*******************************')
+            print(envelope.content.decode('utf8'))
+            print('*******************************')
+            return '250 Message accepted for delivery'
+        finally:
+            with smtp_lock:
+                active_smtp_connections -= 1


-class EchoServerProtocol(asyncio.Protocol):
-    def connection_made(self, transport):
-        global last_received_message
-        self.transport = transport
-        peername = transport.get_extra_info('peername')
-        print('Incoming connection from {}'.format(peername))
-        self.transport.write(last_received_message)
-
-        last_received_message = b''
-        self.transport.close()
+# Simple Flask HTTP server to echo back the last SMTP message
+app = Flask(__name__)


-async def main():
+@app.route('/')
+def echo_last_message():
+    global last_received_message, active_smtp_connections
+
+    # Wait for any in-progress SMTP connections to complete
+    max_wait = 5  # Maximum 5 seconds
+    wait_interval = 0.05  # Check every 50ms
+    elapsed = 0
+
+    while elapsed < max_wait:
+        with smtp_lock:
+            if active_smtp_connections == 0:
+                break
+        time.sleep(wait_interval)
+        elapsed += wait_interval
+
+    return Response(last_received_message, mimetype='text/plain')
+
+
+def run_flask():
+    app.run(host='0.0.0.0', port=11080, debug=False, use_reloader=False)
+
+
+if __name__ == "__main__":
    # Start the SMTP server
    controller = Controller(CustomSMTPHandler(), hostname='0.0.0.0', port=11025)
    controller.start()

-    # Start the TCP Echo server
-    loop = asyncio.get_running_loop()
-    server = await loop.create_server(
-        lambda: EchoServerProtocol(),
-        '0.0.0.0', 11080
-    )
-    async with server:
-        await server.serve_forever()
+    # Start the HTTP server in a separate thread
+    flask_thread = threading.Thread(target=run_flask, daemon=True)
+    flask_thread.start()

-
-if __name__ == "__main__":
-    asyncio.run(main())
+    # Keep the main thread alive
+    try:
+        flask_thread.join()
+    except KeyboardInterrupt:
+        print("Shutting down...")
--- a/changedetectionio/tests/smtp/test_notification_smtp.py
+++ b/changedetectionio/tests/smtp/test_notification_smtp.py
@@ -3,7 +3,7 @@ from flask import url_for
 from email import message_from_string
 from email.policy import default as email_policy

-from changedetectionio.diff import HTML_REMOVED_STYLE, HTML_ADDED_STYLE
+from changedetectionio.diff import HTML_REMOVED_STYLE, HTML_ADDED_STYLE, HTML_CHANGED_STYLE
 from changedetectionio.tests.util import set_original_response, set_modified_response, set_more_modified_response, live_server_setup, \
    wait_for_all_checks, \
    set_longer_modified_response, delete_all_watches
@@ -24,16 +24,14 @@ from changedetectionio.notification import (


 def get_last_message_from_smtp_server():
-    import socket
-    port = 11080  # socket server port number
-
-    client_socket = socket.socket()  # instantiate
-    client_socket.connect((smtp_test_server, port))  # connect to the server
-
-    data = client_socket.recv(50024).decode()  # receive response
+    import requests
+    time.sleep(1) # wait for any smtp connects to die off
+    port = 11080  # HTTP server port number
+    # Make HTTP GET request to Flask server
+    response = requests.get(f'http://{smtp_test_server}:{port}/')
+    data = response.text
    logging.info("get_last_message_from_smtp_server..")
    logging.info(data)
-    client_socket.close()  # close the connection
    return data


@@ -172,7 +170,7 @@ def test_check_notification_html_color_format(client, live_server, measure_memor
        url_for("settings.settings_page"),
        data={"application-notification_urls": notification_url,
              "application-notification_title": "fallback-title " + default_notification_title,
-              "application-notification_body": "some text\n" + default_notification_body,
+              "application-notification_body": "some text\n" + default_notification_body, #some text\n should get <br>
              "application-notification_format": 'HTML Color',
              "requests-time_between_check-minutes": 180,
              'application-fetch_backend': "html_requests"},
@@ -225,7 +223,7 @@ def test_check_notification_html_color_format(client, live_server, measure_memor
    html_part = parts[1]
    assert html_part.get_content_type() == 'text/html'
    html_content = html_part.get_content()
-    assert HTML_REMOVED_STYLE in html_content
+    assert HTML_CHANGED_STYLE or HTML_REMOVED_STYLE in html_content
    assert HTML_ADDED_STYLE in html_content

    assert 'some text<br>' in html_content
@@ -299,7 +297,7 @@ def test_check_notification_markdown_format(client, live_server, measure_memory_
    assert '(added) So let\'s see what happens.<br' in html_content
    delete_all_watches(client)

-
+# Custom notification body with HTML, that is either sent as HTML or rendered to plaintext and sent
 def test_check_notification_email_formats_default_Text_override_HTML(client, live_server, measure_memory_usage):

    # HTML problems? see this
@@ -334,7 +332,7 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
    assert b"Settings updated." in res.data

    # Add a watch and trigger a HTTP POST
-    test_url = url_for('test_endpoint', _external=True)
+    test_url = url_for('test_endpoint',content_type="text/html", _external=True)
    res = client.post(
        url_for("ui.ui_views.form_quick_watch_add"),
        data={"url": test_url, "tags": 'nice one'},
@@ -343,6 +341,7 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv

    assert b"Watch added" in res.data

+    #################################### FIRST SITUATION, PLAIN TEXT NOTIFICATION IS WANTED BUT WE HAVE HTML IN OUR TEMPLATE AND CONTENT ##########
    wait_for_all_checks(client)
    set_longer_modified_response()
    time.sleep(2)
@@ -365,7 +364,10 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
    # Get the plain text content
    text_content = msg.get_content()
    assert '(added) So let\'s see what happens.\r\n' in text_content  # The plaintext part
+    assert '<!DOCTYPE html>' in text_content # even tho they added html, they selected plaintext so it should have not got converted

+
+    #################################### SECOND SITUATION, HTML IS CORRECTLY PASSED THROUGH TO THE EMAIL ####################
    set_original_response()
    # Now override as HTML format
    res = client.post(
@@ -405,10 +407,130 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
    html_part = parts[1]
    assert html_part.get_content_type() == 'text/html'
    html_content = html_part.get_content()
-    assert '(removed) So let\'s see what happens.<br>' in html_content  # the html part
+    assert '(removed) So let\'s see what happens.' in html_content  # the html part
+    assert '&lt;!DOCTYPE html' not in html_content
+    assert '<!DOCTYPE html' in html_content # Our original template is working correctly

    # https://github.com/dgtlmoon/changedetection.io/issues/2103
    assert '<h1>Test</h1>' in html_content
    assert '&lt;' not in html_content

    delete_all_watches(client)
+
+def test_check_plaintext_document_plaintext_notification_smtp(client, live_server, measure_memory_usage):
+    """When following a plaintext document, notification in Plain Text format is sent correctly"""
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("Some nice plain text\nwhich we add some extra data\nover here\n")
+
+    notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
+    notification_body = f"""{default_notification_body}"""
+
+    #####################
+    # Set this up for when we remove the notification from the watch, it should fallback with these details
+    res = client.post(
+        url_for("settings.settings_page"),
+        data={"application-notification_urls": notification_url,
+              "application-notification_title": "fallback-title " + default_notification_title,
+              "application-notification_body": notification_body,
+              "application-notification_format": 'Plain Text',
+              "requests-time_between_check-minutes": 180,
+              'application-fetch_backend': "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Settings updated." in res.data
+
+    # Add our URL to the import page
+    test_url = url_for('test_endpoint', content_type="text/plain", _external=True)
+    uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
+    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
+    wait_for_all_checks(client)
+
+    # Change the content
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("Some nice plain text\nwhich we add some extra data\nAnd let's talk about <title> tags\nover here\n")
+
+
+    time.sleep(1)
+    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
+    wait_for_all_checks(client)
+
+    # Parse the email properly using Python's email library
+    msg = message_from_string(get_last_message_from_smtp_server(), policy=email_policy)
+
+    assert not msg.is_multipart()
+    assert msg.get_content_type() == 'text/plain'
+    body = msg.get_content()
+    # nothing is escaped, raw html stuff in text/plain
+    assert 'talk about <title> tags' in body
+    assert '(added)' in body
+    assert '<br' not in body
+
+    delete_all_watches(client)
+
+def test_check_plaintext_document_html_notifications(client, live_server, measure_memory_usage):
+    """When following a plaintext document, notification in Plain Text format is sent correctly"""
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("Some nice plain text\nwhich we add some extra data\nover here\n")
+
+    notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
+    notification_body = f"""{default_notification_body}"""
+
+    #####################
+    # Set this up for when we remove the notification from the watch, it should fallback with these details
+    res = client.post(
+        url_for("settings.settings_page"),
+        data={"application-notification_urls": notification_url,
+              "application-notification_title": "fallback-title " + default_notification_title,
+              "application-notification_body": notification_body,
+              "application-notification_format": 'HTML',
+              "requests-time_between_check-minutes": 180,
+              'application-fetch_backend': "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Settings updated." in res.data
+
+    # Add our URL to the import page
+    test_url = url_for('test_endpoint', content_type="text/plain", _external=True)
+    uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
+    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
+    wait_for_all_checks(client)
+
+    # Change the content
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("Some nice plain text\nwhich we add some extra data\nAnd let's talk about <title> tags\nover here\n")
+
+
+    time.sleep(1)
+    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
+    wait_for_all_checks(client)
+
+    # Parse the email properly using Python's email library
+    msg = message_from_string(get_last_message_from_smtp_server(), policy=email_policy)
+
+
+    # The email should have two bodies (multipart/alternative)
+    assert msg.is_multipart()
+    assert msg.get_content_type() == 'multipart/alternative'
+
+    # Get the parts
+    parts = list(msg.iter_parts())
+    assert len(parts) == 2
+
+    # First part should be text/plain
+    text_part = parts[0]
+    assert text_part.get_content_type() == 'text/plain'
+    text_content = text_part.get_content()
+
+
+    assert 'And let\'s talk about <title> tags\r\n' in text_content
+
+    # Second part should be text/html
+    html_part = parts[1]
+    assert html_part.get_content_type() == 'text/html'
+    html_content = html_part.get_content()
+    assert 'talk about <title>' not in html_content  # the html part, should have got marked up to &lt; etc
+    assert '<br>\r\n(added) And let&#39;s talk about &lt;title&gt; tags<br>' in html_content
+
+    delete_all_watches(client)
--- a/changedetectionio/tests/test_notification.py
+++ b/changedetectionio/tests/test_notification.py
@@ -16,7 +16,7 @@ from changedetectionio.notification import (
    default_notification_title,
    valid_notification_formats,
 )
-
+from ..diff import HTML_CHANGED_STYLE


 # Hard to just add more live server URLs when one test is already running (I think)
@@ -485,8 +485,6 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage

 def _test_color_notifications(client, notification_body_token):

-    from changedetectionio.diff import HTML_ADDED_STYLE, HTML_REMOVED_STYLE
-
    set_original_response()

    if os.path.isfile("test-datastore/notification.txt"):
@@ -533,7 +531,8 @@ def _test_color_notifications(client, notification_body_token):

    with open("test-datastore/notification.txt", 'r') as f:
        x = f.read()
-        assert f'<span style="{HTML_REMOVED_STYLE}">Which is across multiple lines' in x
+        s =  f'<span style="{HTML_CHANGED_STYLE}" role="note" aria-label="Changed text" title="Changed text">Which is across multiple lines'
+        assert s in x


    client.get(
--- a/changedetectionio/tests/util.py
+++ b/changedetectionio/tests/util.py
@@ -244,7 +244,7 @@ def new_live_server_setup(live_server):
        return request.method

    # Where we POST to as a notification, also use a space here to test URL escaping is OK across all tests that use this. ( #2868 )
-    @live_server.app.route('/test_notification endpoint', methods=['POST', 'GET'])
+    @live_server.app.route('/test_notification_endpoint', methods=['POST', 'GET'])
    def test_notification_endpoint():

        with open("test-datastore/notification.txt", "wb") as f: