mirror of
https://github.com/dgtlmoon/changedetection.io.git
synced 2025-11-20 16:36:08 +00:00
Compare commits
6 Commits
rss-reader
...
email-noti
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6c54b4344f | ||
|
|
c30da09861 | ||
|
|
c0d51402ab | ||
|
|
7ab68380aa | ||
|
|
74792685c9 | ||
|
|
6647d0bb09 |
@@ -1,6 +1,5 @@
|
||||
from changedetectionio.model import default_notification_format_for_watch
|
||||
|
||||
ult_notification_format_for_watch = 'System default'
|
||||
default_notification_format = 'HTML Color'
|
||||
default_notification_body = '{{watch_url}} had a change.\n---\n{{diff}}\n---\n'
|
||||
default_notification_title = 'ChangeDetection.io Notification - {{watch_url}}'
|
||||
|
||||
@@ -70,6 +70,7 @@ def apprise_http_custom_handler(
|
||||
title: str,
|
||||
notify_type: str,
|
||||
meta: dict,
|
||||
body_format: str = None,
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> bool:
|
||||
|
||||
@@ -3,7 +3,9 @@ import time
|
||||
import apprise
|
||||
from apprise import NotifyFormat
|
||||
from loguru import logger
|
||||
from urllib.parse import urlparse
|
||||
from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
|
||||
from .apprise_plugin.custom_handlers import SUPPORTED_HTTP_METHODS
|
||||
from ..notification_service import NotificationContextData
|
||||
|
||||
|
||||
@@ -57,18 +59,17 @@ def notification_format_align_with_apprise(n_format : str):
|
||||
|
||||
if n_format.lower().startswith('html'):
|
||||
# Apprise only knows 'html' not 'htmlcolor' etc, which shouldnt matter here
|
||||
n_format = NotifyFormat.HTML
|
||||
n_format = NotifyFormat.HTML.value
|
||||
elif n_format.lower().startswith('markdown'):
|
||||
# probably the same but just to be safe
|
||||
n_format = NotifyFormat.MARKDOWN
|
||||
n_format = NotifyFormat.MARKDOWN.value
|
||||
elif n_format.lower().startswith('text'):
|
||||
# probably the same but just to be safe
|
||||
n_format = NotifyFormat.TEXT
|
||||
n_format = NotifyFormat.TEXT.value
|
||||
else:
|
||||
n_format = NotifyFormat.TEXT
|
||||
n_format = NotifyFormat.TEXT.value
|
||||
|
||||
# Must be str for apprise notify body_format
|
||||
return str(n_format)
|
||||
return n_format
|
||||
|
||||
def process_notification(n_object: NotificationContextData, datastore):
|
||||
from changedetectionio.jinja2_custom import render as jinja_render
|
||||
@@ -123,7 +124,7 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
if n_object.get('markup_text_to_html'):
|
||||
n_body = markup_text_links_to_html(body=n_body)
|
||||
|
||||
if n_format == str(NotifyFormat.HTML):
|
||||
if n_format == NotifyFormat.HTML.value:
|
||||
n_body = n_body.replace("\n", '<br>')
|
||||
|
||||
n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)
|
||||
@@ -147,7 +148,8 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
# Length of URL - Incase they specify a longer custom avatar_url
|
||||
|
||||
# So if no avatar_url is specified, add one so it can be correctly calculated into the total payload
|
||||
k = '?' if not '?' in url else '&'
|
||||
parsed = urlparse(url)
|
||||
k = '?' if not parsed.query else '&'
|
||||
if not 'avatar_url' in url \
|
||||
and not url.startswith('mail') \
|
||||
and not url.startswith('post') \
|
||||
@@ -176,16 +178,15 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
n_title = n_title[0:payload_max_size]
|
||||
n_body = n_body[0:body_limit]
|
||||
|
||||
elif url.startswith('mailto'):
|
||||
# Apprise will default to HTML, so we need to override it
|
||||
# So that whats' generated in n_body is in line with what is going to be sent.
|
||||
# https://github.com/caronc/apprise/issues/633#issuecomment-1191449321
|
||||
if not 'format=' in url and (n_format == 'Text' or n_format == 'Markdown'):
|
||||
prefix = '?' if not '?' in url else '&'
|
||||
# Apprise format is lowercase text https://github.com/caronc/apprise/issues/633
|
||||
n_format = n_format.lower()
|
||||
url = f"{url}{prefix}format={n_format}"
|
||||
# If n_format == HTML, then apprise email should default to text/html and we should be sending HTML only
|
||||
# Add format parameter to mailto URLs to ensure proper text/html handling
|
||||
# https://github.com/caronc/apprise/issues/633#issuecomment-1191449321
|
||||
# Note: Custom handlers (post://, get://, etc.) don't need this as we handle them
|
||||
# differently by passing an invalid body_format to prevent HTML conversion
|
||||
if not 'format=' in url and url.startswith(('mailto', 'mailtos')):
|
||||
parsed = urlparse(url)
|
||||
prefix = '?' if not parsed.query else '&'
|
||||
# Apprise format is already lowercase from notification_format_align_with_apprise()
|
||||
url = f"{url}{prefix}format={n_format}"
|
||||
|
||||
apobj.add(url)
|
||||
|
||||
@@ -195,10 +196,28 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
'body_format': n_format})
|
||||
|
||||
# Blast off the notifications tht are set in .add()
|
||||
# Check if we have any custom HTTP handlers (post://, get://, etc.)
|
||||
# These handlers created with @notify decorator don't handle format conversion properly
|
||||
# and will strip HTML if we pass a valid format. So we pass an invalid format string
|
||||
# to prevent Apprise from converting HTML->TEXT
|
||||
|
||||
# Create list of custom handler protocols (both http and https versions)
|
||||
custom_handler_protocols = [f"{method}://" for method in SUPPORTED_HTTP_METHODS]
|
||||
custom_handler_protocols += [f"{method}s://" for method in SUPPORTED_HTTP_METHODS]
|
||||
|
||||
has_custom_handler = any(
|
||||
url.startswith(tuple(custom_handler_protocols))
|
||||
for url in n_object['notification_urls']
|
||||
)
|
||||
|
||||
# If we have custom handlers, use invalid format to prevent conversion
|
||||
# Otherwise use the proper format
|
||||
notify_format = 'raw-no-convert' if has_custom_handler else n_format
|
||||
|
||||
apobj.notify(
|
||||
title=n_title,
|
||||
body=n_body,
|
||||
body_format=n_format,
|
||||
body_format=notify_format,
|
||||
# False is not an option for AppRise, must be type None
|
||||
attach=n_object.get('screenshot', None)
|
||||
)
|
||||
@@ -207,7 +226,7 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
# Returns empty string if nothing found, multi-line string otherwise
|
||||
log_value = logs.getvalue()
|
||||
|
||||
if log_value and 'WARNING' in log_value or 'ERROR' in log_value:
|
||||
if log_value and ('WARNING' in log_value or 'ERROR' in log_value):
|
||||
logger.critical(log_value)
|
||||
raise Exception(log_value)
|
||||
|
||||
|
||||
@@ -3,6 +3,10 @@ import os
|
||||
import time
|
||||
import re
|
||||
from flask import url_for
|
||||
from email import message_from_string
|
||||
from email.policy import default as email_policy
|
||||
|
||||
from changedetectionio.diff import REMOVED_STYLE, ADDED_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
|
||||
@@ -50,7 +54,7 @@ def test_check_notification_email_formats_default_HTML(client, live_server, meas
|
||||
url_for("settings.settings_page"),
|
||||
data={"application-notification_urls": notification_url,
|
||||
"application-notification_title": "fallback-title " + default_notification_title,
|
||||
"application-notification_body": "fallback-body<br> " + default_notification_body,
|
||||
"application-notification_body": "some text\nfallback-body<br> " + default_notification_body,
|
||||
"application-notification_format": 'HTML',
|
||||
"requests-time_between_check-minutes": 180,
|
||||
'application-fetch_backend': "html_requests"},
|
||||
@@ -77,14 +81,164 @@ def test_check_notification_email_formats_default_HTML(client, live_server, meas
|
||||
|
||||
time.sleep(3)
|
||||
|
||||
msg = get_last_message_from_smtp_server()
|
||||
assert len(msg) >= 1
|
||||
msg_raw = get_last_message_from_smtp_server()
|
||||
assert len(msg_raw) >= 1
|
||||
|
||||
# The email should have two bodies, and the text/html part should be <br>
|
||||
assert 'Content-Type: text/plain' in msg
|
||||
assert '(added) So let\'s see what happens.\r\n' in msg # The plaintext part with \r\n
|
||||
assert 'Content-Type: text/html' in msg
|
||||
assert '(added) So let\'s see what happens.<br>' in msg # the html part
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(msg_raw, policy=email_policy)
|
||||
|
||||
# The email should have two bodies (multipart/alternative with text/plain and text/html)
|
||||
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 (the auto-generated plaintext version)
|
||||
text_part = parts[0]
|
||||
assert text_part.get_content_type() == 'text/plain'
|
||||
text_content = text_part.get_content()
|
||||
assert '(added) So let\'s see what happens.\r\n' in text_content # The plaintext part
|
||||
assert 'fallback-body\r\n' in text_content # The plaintext part
|
||||
|
||||
# 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 'some text<br>' in html_content # We converted \n from the notification body
|
||||
assert 'fallback-body<br>' in html_content # kept the original <br>
|
||||
assert '(added) So let\'s see what happens.<br>' in html_content # the html part
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
def test_check_notification_plaintext_format(client, live_server, measure_memory_usage):
|
||||
set_original_response()
|
||||
|
||||
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
|
||||
|
||||
#####################
|
||||
# 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": "some text\n" + default_notification_body,
|
||||
"application-notification_format": 'Text',
|
||||
"requests-time_between_check-minutes": 180,
|
||||
'application-fetch_backend': "html_requests"},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
assert b"Settings updated." in res.data
|
||||
|
||||
# Add a watch and trigger a HTTP POST
|
||||
test_url = url_for('test_endpoint', _external=True)
|
||||
res = client.post(
|
||||
url_for("ui.ui_views.form_quick_watch_add"),
|
||||
data={"url": test_url, "tags": 'nice one'},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
assert b"Watch added" in res.data
|
||||
|
||||
wait_for_all_checks(client)
|
||||
set_longer_modified_response()
|
||||
time.sleep(2)
|
||||
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
time.sleep(3)
|
||||
|
||||
msg_raw = get_last_message_from_smtp_server()
|
||||
assert len(msg_raw) >= 1
|
||||
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(msg_raw, policy=email_policy)
|
||||
|
||||
# The email should be plain text only (not multipart)
|
||||
assert not msg.is_multipart()
|
||||
assert msg.get_content_type() == 'text/plain'
|
||||
|
||||
# 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
|
||||
|
||||
# Should NOT contain HTML
|
||||
assert '<br>' not in text_content # We should not have HTML in plain text
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
|
||||
def test_check_notification_html_color_format(client, live_server, measure_memory_usage):
|
||||
set_original_response()
|
||||
|
||||
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
|
||||
|
||||
#####################
|
||||
# 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": "some text\n" + default_notification_body,
|
||||
"application-notification_format": 'HTML Color',
|
||||
"requests-time_between_check-minutes": 180,
|
||||
'application-fetch_backend': "html_requests"},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
assert b"Settings updated." in res.data
|
||||
|
||||
# Add a watch and trigger a HTTP POST
|
||||
test_url = url_for('test_endpoint', _external=True)
|
||||
res = client.post(
|
||||
url_for("ui.ui_views.form_quick_watch_add"),
|
||||
data={"url": test_url, "tags": 'nice one'},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
assert b"Watch added" in res.data
|
||||
|
||||
wait_for_all_checks(client)
|
||||
set_longer_modified_response()
|
||||
time.sleep(2)
|
||||
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
time.sleep(3)
|
||||
|
||||
msg_raw = get_last_message_from_smtp_server()
|
||||
assert len(msg_raw) >= 1
|
||||
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(msg_raw, policy=email_policy)
|
||||
|
||||
# The email should have two bodies (multipart/alternative with text/plain and text/html)
|
||||
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 (the auto-generated plaintext version)
|
||||
text_part = parts[0]
|
||||
assert text_part.get_content_type() == 'text/plain'
|
||||
text_content = text_part.get_content()
|
||||
assert 'So let\'s see what happens.\r\n' in text_content # The plaintext part
|
||||
assert '(added)' not in text_content # Because apprise only dumb converts the html to text
|
||||
|
||||
# Second part should be text/html with color styling
|
||||
html_part = parts[1]
|
||||
assert html_part.get_content_type() == 'text/html'
|
||||
html_content = html_part.get_content()
|
||||
assert REMOVED_STYLE in html_content
|
||||
assert ADDED_STYLE in html_content
|
||||
|
||||
assert 'some text<br>' in html_content
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
@@ -139,15 +293,21 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
|
||||
wait_for_all_checks(client)
|
||||
|
||||
time.sleep(3)
|
||||
msg = get_last_message_from_smtp_server()
|
||||
assert len(msg) >= 1
|
||||
msg_raw = get_last_message_from_smtp_server()
|
||||
assert len(msg_raw) >= 1
|
||||
# with open('/tmp/m.txt', 'w') as f:
|
||||
# f.write(msg)
|
||||
# f.write(msg_raw)
|
||||
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(msg_raw, policy=email_policy)
|
||||
|
||||
# The email should not have two bodies, should be TEXT only
|
||||
assert not msg.is_multipart()
|
||||
assert msg.get_content_type() == 'text/plain'
|
||||
|
||||
assert 'Content-Type: text/plain' in msg
|
||||
assert '(added) So let\'s see what happens.\r\n' in msg # The plaintext part with \r\n
|
||||
# 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
|
||||
|
||||
set_original_response()
|
||||
# Now override as HTML format
|
||||
@@ -164,18 +324,34 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
|
||||
wait_for_all_checks(client)
|
||||
|
||||
time.sleep(3)
|
||||
msg = get_last_message_from_smtp_server()
|
||||
assert len(msg) >= 1
|
||||
msg_raw = get_last_message_from_smtp_server()
|
||||
assert len(msg_raw) >= 1
|
||||
|
||||
# The email should have two bodies, and the text/html part should be <br>
|
||||
assert 'Content-Type: text/plain' in msg
|
||||
assert '(removed) So let\'s see what happens.\r\n' in msg # The plaintext part with \n
|
||||
assert 'Content-Type: text/html' in msg
|
||||
assert '(removed) So let\'s see what happens.<br>' in msg # the html part
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(msg_raw, 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 '(removed) So let\'s see what happens.\r\n' in text_content # The plaintext part
|
||||
|
||||
# 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 '(removed) So let\'s see what happens.<br>' in html_content # the html part
|
||||
|
||||
# https://github.com/dgtlmoon/changedetection.io/issues/2103
|
||||
assert '<h1>Test</h1>' in msg
|
||||
assert '<' not in msg
|
||||
assert 'Content-Type: text/html' in msg
|
||||
assert '<h1>Test</h1>' in html_content
|
||||
assert '<' not in html_content
|
||||
|
||||
delete_all_watches(client)
|
||||
|
||||
@@ -541,9 +541,7 @@ def _test_color_notifications(client, notification_body_token):
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Just checks the format of the colour notifications was correct
|
||||
def test_html_color_notifications(client, live_server, measure_memory_usage):
|
||||
|
||||
|
||||
_test_color_notifications(client, '{{diff}}')
|
||||
_test_color_notifications(client, '{{diff_full}}')
|
||||
|
||||
Reference in New Issue
Block a user