Preserve whitespace's in HTML style notifications

changedetectionio/diff.py

@@ -2,10 +2,17 @@ import difflib
 from typing import List, Iterator, Union
 
 # https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050
-HTML_ADDED_STYLE = "background-color: #d2f7c2; color: #255d00;"
-HTML_CHANGED_INTO_STYLE = "background-color: #dafbe1; color: #116329;"
-HTML_CHANGED_STYLE = "background-color: #ffd6cc; color: #7a2000;"
-HTML_REMOVED_STYLE = "background-color: #ffebe9; color: #82071e;"
+#HTML_ADDED_STYLE = "background-color: #d2f7c2; color: #255d00;"
+#HTML_CHANGED_INTO_STYLE = "background-color: #dafbe1; color: #116329;"
+#HTML_CHANGED_STYLE = "background-color: #ffd6cc; color: #7a2000;"
+#HTML_REMOVED_STYLE = "background-color: #ffebe9; color: #82071e;"
+
+# @todo - In the future we can make this configurable
+HTML_ADDED_STYLE = "background-color: #eaf2c2; color: #406619"
+HTML_REMOVED_STYLE = "background-color: #fadad7; color: #b30000"
+HTML_CHANGED_STYLE = HTML_REMOVED_STYLE
+HTML_CHANGED_INTO_STYLE = HTML_ADDED_STYLE
+
 
 # These get set to html or telegram type or discord compatible or whatever in handler.py
 # Something that cant get escaped to HTML by accident

changedetectionio/notification/apprise_plugin/custom_handlers.py

@@ -195,25 +195,16 @@ def apprise_http_custom_handler(
 
     url = re.sub(rf"^{schema}", "https" if schema.endswith("s") else "http", parsed_url.get("url"))
 
-    try:
-        response = requests.request(
-            method=method,
-            url=url,
-            auth=auth,
-            headers=headers,
-            params=params,
-            data=body.encode("utf-8") if isinstance(body, str) else body,
-        )
+    response = requests.request(
+        method=method,
+        url=url,
+        auth=auth,
+        headers=headers,
+        params=params,
+        data=body.encode("utf-8") if isinstance(body, str) else body,
+    )
 
-        response.raise_for_status()
+    response.raise_for_status()
 
-        logger.info(f"Successfully sent custom notification to {url}")
-        return True
-
-    except requests.RequestException as e:
-        logger.error(f"Remote host error while sending custom notification to {url}: {e}")
-        return False
-
-    except Exception as e:
-        logger.error(f"Unexpected error occurred while sending custom notification to {url}: {e}")
-        return False
+    logger.info(f"Successfully sent custom notification to {url}")
+    return True

changedetectionio/notification/handler.py

@@ -257,7 +257,6 @@ def process_notification(n_object: NotificationContextData, datastore):
             if n_object.get('markup_text_links_to_html_links'):
                 n_body = markup_text_links_to_html(body=n_body)
 
-
             n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)
 
             url = url.strip()
@@ -279,6 +278,11 @@ def process_notification(n_object: NotificationContextData, datastore):
                     from markupsafe import escape
                     n_body = str(escape(n_body))
 
+            if 'html' in requested_output_format:
+                # Since the n_body is always some kind of text from the 'diff' engine, attempt to preserve whitespaces that get sent to the HTML output
+                # But only where its more than 1 consecutive whitespace, otherwise "and this" becomes "and this" etc which is too much.
+                n_body = n_body.replace('  ', '  ')
+
             (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"
@@ -306,8 +310,8 @@ def process_notification(n_object: NotificationContextData, datastore):
                 if 'html' in requested_output_format:
                     n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '<br>\n')
                 else:
-                    # Just incase
-                    n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '')
+                    # Markup, text types etc
+                    n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '\r\n')
 
             sent_objs.append({'title': n_title,
                               'body': n_body,

changedetectionio/tests/smtp/smtp-test-server.py

@@ -32,6 +32,8 @@ class CustomSMTPHandler:
 
             # Parse the email message
             msg = message_from_bytes(envelope.content, policy=default)
+            with open('/tmp/last.eml', 'wb') as f:
+                f.write(envelope.content)
 
             # Write parts to files based on content type
             if msg.is_multipart():

changedetectionio/tests/smtp/test_notification_smtp.py

@@ -477,7 +477,7 @@ def test_check_plaintext_document_html_notifications(client, live_server, measur
     """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")
+        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}"""
@@ -504,7 +504,7 @@ def test_check_plaintext_document_html_notifications(client, live_server, measur
 
     # 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")
+        f.write("    Some nice plain text\nwhich we add some extra data\nAnd let's talk about <title> tags\nover here\n")
 
 
     time.sleep(2)
@@ -544,6 +544,10 @@ def test_check_plaintext_document_html_notifications(client, live_server, measur
     assert '<br>\r\n(added) And let&#39;s talk about &lt;title&gt; tags<br>' in html_content
     assert '&lt;br' not in html_content
 
+    # And now for the whitespace retention
+    assert '&nbsp;&nbsp;&nbsp;&nbsp;Some nice plain text' in html_content
+    assert '(added) And let' in html_content # just to show a single whitespace didnt get touched
+
     delete_all_watches(client)
 
 
@@ -618,3 +622,60 @@ def test_check_plaintext_document_html_color_notifications(client, live_server,
     assert '<br>' in html_content
 
     delete_all_watches(client)
+
+def test_check_html_document_plaintext_notification(client, live_server, measure_memory_usage):
+    """When following a HTML document, notification in Plain Text format is sent correctly"""
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("<html><body>some stuff<br>and more stuff<br>and even more stuff<br></body></html>")
+
+    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": f"{notification_body}\nMore output test\n{ALL_MARKUP_TOKENS}",
+              "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/html", _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)
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("<html><body>sxome stuff<br>and more stuff<br>lets slip this in<br>and this in<br>and even more stuff<br>&lt;tag&gt;</body></html>")
+
+    time.sleep(0.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()
+
+    assert '<tag>' in body # Should have got converted from original HTML to plaintext
+    assert '(changed) some stuff\r\n' in body
+    assert '(into) sxome stuff\r\n' in body
+    assert '(added) lets slip this in\r\n' in body
+    assert '(added) and this in\r\n' in body
+    assert '&nbsp;' not in body
+
+
+    delete_all_watches(client)
+
+

changedetectionio/tests/test_notification.py

@@ -467,6 +467,25 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
         # Should come from notification.py default handler when there is no notification body to pull from
         assert 'change detection is cool 网站监测 内容更新了' in x
 
+    ## Check that 'test' catches errors
+    test_notification_url = 'post://akjsdfkjasdkfjasdkfjasdkjfas232323/should-error'
+
+    ######### Test global/system settings
+    res = client.post(
+        url_for("ui.ui_notification.ajax_callback_send_notification_test")+"?mode=global-settings",
+        data={"notification_urls": test_notification_url},
+        follow_redirects=True
+    )
+    assert res.status_code == 400
+    assert (
+        b"No address found" in res.data or
+        b"Name or service not known" in res.data or
+        b"nodename nor servname provided" in res.data or
+        b"Temporary failure in name resolution" in res.data or
+        b"Failed to establish a new connection" in res.data or
+        b"Connection error occurred" in res.data
+    )
+    
     client.get(
         url_for("ui.form_delete", uuid="all"),
         follow_redirects=True
@@ -483,6 +502,7 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
     assert b"Error: You must have atleast one watch configured for 'test notification' to work" in res.data
 
 
+
 def _test_color_notifications(client, notification_body_token):
 
     set_original_response()