mirror of
https://github.com/dgtlmoon/changedetection.io.git
synced 2025-11-23 18:06:09 +00:00
Compare commits
16 Commits
rss-per-gr
...
rss-reader
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
78647d308d | ||
|
|
00d28c6c40 | ||
|
|
91729ae724 | ||
|
|
e09c86dd13 | ||
|
|
90d68f7ca7 | ||
|
|
b6b733a1fa | ||
|
|
0be5005776 | ||
|
|
12ce03c0bb | ||
|
|
3767a2d5b9 | ||
|
|
71c8d8b1b1 | ||
|
|
20cbe6f510 | ||
|
|
3a6e1f908f | ||
|
|
73fdbf24e3 | ||
|
|
629f939224 | ||
|
|
48299e5738 | ||
|
|
5b1b70b8ab |
@@ -1,8 +1,8 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
# Read more https://github.com/dgtlmoon/changedetection.io/wiki
|
||||
|
||||
__version__ = '0.50.43'
|
||||
# Semver means never use .01, or 00. Should be .1.
|
||||
__version__ = '0.51.2'
|
||||
|
||||
from changedetectionio.strtobool import strtobool
|
||||
from json.decoder import JSONDecodeError
|
||||
|
||||
@@ -3,6 +3,7 @@ from loguru import logger
|
||||
|
||||
from changedetectionio.model import USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH
|
||||
from changedetectionio.notification import valid_notification_formats
|
||||
|
||||
RSS_CONTENT_FORMAT_DEFAULT = 'text'
|
||||
|
||||
# Some stuff not related
|
||||
@@ -15,3 +16,12 @@ if RSS_FORMAT_TYPES.get(USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH):
|
||||
|
||||
if not RSS_FORMAT_TYPES.get(RSS_CONTENT_FORMAT_DEFAULT):
|
||||
logger.critical(f"RSS_CONTENT_FORMAT_DEFAULT not in the acceptable list {RSS_CONTENT_FORMAT_DEFAULT}")
|
||||
|
||||
RSS_TEMPLATE_TYPE_OPTIONS = {'system_default': 'System default', 'notification_body': 'Notification body'}
|
||||
|
||||
# @note: We use <pre> because nearly all RSS readers render only HTML (Thunderbird for example cant do just plaintext)
|
||||
RSS_TEMPLATE_PLAINTEXT_DEFAULT = "<pre>{{watch_label}} had a change.\n\n{{diff}}\n</pre>"
|
||||
|
||||
# @todo add some [edit]/[history]/[goto] etc links
|
||||
# @todo need {{watch_edit_link}} + delete + history link token
|
||||
RSS_TEMPLATE_HTML_DEFAULT = "<html><body>\n<h4><a href=\"{{watch_url}}\">{{watch_label}}</a></h4>\n<p>{{diff}}</p>\n</body></html>\n"
|
||||
|
||||
@@ -2,9 +2,11 @@
|
||||
Utility functions for RSS feed generation.
|
||||
"""
|
||||
|
||||
from changedetectionio.jinja2_custom import render as jinja_render
|
||||
from changedetectionio.notification.handler import apply_service_tweaks
|
||||
from changedetectionio.notification.handler import process_notification
|
||||
from changedetectionio.notification_service import NotificationContextData, _check_cascading_vars
|
||||
from loguru import logger
|
||||
import datetime
|
||||
import pytz
|
||||
import re
|
||||
|
||||
|
||||
@@ -39,57 +41,116 @@ def clean_entry_content(content):
|
||||
return cleaned
|
||||
|
||||
|
||||
def generate_watch_guid(watch):
|
||||
def generate_watch_guid(watch, timestamp):
|
||||
"""
|
||||
Generate a unique GUID for a watch RSS entry.
|
||||
"""
|
||||
return f"{watch['uuid']}/{watch.last_changed}"
|
||||
|
||||
|
||||
def generate_watch_diff_content(watch, dates, rss_content_format, datastore):
|
||||
"""
|
||||
Generate HTML diff content for a watch given its history dates.
|
||||
Returns tuple of (content, watch_label).
|
||||
|
||||
Args:
|
||||
watch: The watch object
|
||||
dates: List of history snapshot dates
|
||||
rss_content_format: Format for RSS content (html or text)
|
||||
datastore: The ChangeDetectionStore instance
|
||||
timestamp: The timestamp of the specific change this entry represents
|
||||
"""
|
||||
return f"{watch['uuid']}/{timestamp}"
|
||||
|
||||
|
||||
def validate_rss_token(datastore, request):
|
||||
"""
|
||||
Validate the RSS access token from the request.
|
||||
|
||||
Returns:
|
||||
Tuple of (content, watch_label) - the rendered HTML content and watch label
|
||||
tuple: (is_valid, error_response) where error_response is None if valid
|
||||
"""
|
||||
from changedetectionio import diff
|
||||
app_rss_token = datastore.data['settings']['application'].get('rss_access_token')
|
||||
rss_url_token = request.args.get('token')
|
||||
|
||||
# Same logic as watch-overview.html
|
||||
if datastore.data['settings']['application']['ui'].get('use_page_title_in_list') or watch.get('use_page_title_in_list'):
|
||||
watch_label = watch.label
|
||||
if rss_url_token != app_rss_token:
|
||||
return False, ("Access denied, bad token", 403)
|
||||
|
||||
return True, None
|
||||
|
||||
|
||||
def get_rss_template(datastore, watch, rss_content_format, default_html, default_plaintext):
|
||||
"""Get the appropriate template for RSS content."""
|
||||
if datastore.data['settings']['application'].get('rss_template_type') == 'notification_body':
|
||||
return _check_cascading_vars(datastore=datastore, var_name='notification_body', watch=watch)
|
||||
|
||||
override = datastore.data['settings']['application'].get('rss_template_override')
|
||||
if override and override.strip():
|
||||
return override
|
||||
elif 'text' in rss_content_format:
|
||||
return default_plaintext
|
||||
else:
|
||||
watch_label = watch.get('url')
|
||||
return default_html
|
||||
|
||||
try:
|
||||
html_diff = diff.render_diff(
|
||||
previous_version_file_contents=watch.get_history_snapshot(timestamp=dates[-2]),
|
||||
newest_version_file_contents=watch.get_history_snapshot(timestamp=dates[-1]),
|
||||
include_equal=False
|
||||
)
|
||||
|
||||
requested_output_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
url, html_diff, n_title = apply_service_tweaks(url='', n_body=html_diff, n_title=None, requested_output_format=requested_output_format)
|
||||
def get_watch_label(datastore, watch):
|
||||
"""Get the label for a watch based on settings."""
|
||||
if datastore.data['settings']['application']['ui'].get('use_page_title_in_list') or watch.get('use_page_title_in_list'):
|
||||
return watch.label
|
||||
else:
|
||||
return watch.get('url')
|
||||
|
||||
except FileNotFoundError as e:
|
||||
html_diff = f"History snapshot file for watch {watch.get('uuid')}@{watch.last_changed} - '{watch.get('title')} not found."
|
||||
|
||||
# @note: We use <pre> because nearly all RSS readers render only HTML (Thunderbird for example cant do just plaintext)
|
||||
rss_template = "<pre>{{watch_label}} had a change.\n\n{{html_diff}}\n</pre>"
|
||||
if 'html' in rss_content_format:
|
||||
rss_template = "<html><body>\n<h4><a href=\"{{watch_url}}\">{{watch_label}}</a></h4>\n<p>{{html_diff}}</p>\n</body></html>\n"
|
||||
def add_watch_categories(fe, watch, datastore):
|
||||
"""Add category tags to a feed entry based on watch tags."""
|
||||
for tag_uuid in watch.get('tags', []):
|
||||
tag = datastore.data['settings']['application'].get('tags', {}).get(tag_uuid)
|
||||
if tag and tag.get('title'):
|
||||
fe.category(term=tag.get('title'))
|
||||
|
||||
content = jinja_render(template_str=rss_template, watch_label=watch_label, html_diff=html_diff, watch_url=watch.link)
|
||||
|
||||
# Out of range chars could also break feedgen
|
||||
def build_notification_context(watch, timestamp_from, timestamp_to, watch_label,
|
||||
n_body_template, rss_content_format):
|
||||
"""Build the notification context object."""
|
||||
return NotificationContextData(initial_data={
|
||||
'notification_urls': ['null://just-sending-a-null-test-for-the-render-in-RSS'],
|
||||
'notification_body': n_body_template,
|
||||
'timestamp_to': timestamp_to,
|
||||
'timestamp_from': timestamp_from,
|
||||
'watch_label': watch_label,
|
||||
'notification_format': rss_content_format
|
||||
})
|
||||
|
||||
|
||||
def render_notification(n_object, notification_service, watch, datastore,
|
||||
date_index_from=None, date_index_to=None):
|
||||
"""Process and render the notification content."""
|
||||
kwargs = {'n_object': n_object, 'watch': watch}
|
||||
|
||||
if date_index_from is not None and date_index_to is not None:
|
||||
kwargs['date_index_from'] = date_index_from
|
||||
kwargs['date_index_to'] = date_index_to
|
||||
|
||||
n_object = notification_service.queue_notification_for_watch(**kwargs)
|
||||
n_object['watch_mime_type'] = None
|
||||
|
||||
res = process_notification(n_object=n_object, datastore=datastore)
|
||||
return res[0]
|
||||
|
||||
|
||||
def populate_feed_entry(fe, watch, content, guid, timestamp, link=None, title_suffix=None):
|
||||
"""Populate a feed entry with content and metadata."""
|
||||
watch_label = watch.get('url') # Already determined by caller
|
||||
|
||||
# Set link
|
||||
if link:
|
||||
fe.link(link=link)
|
||||
|
||||
# Set title
|
||||
if title_suffix:
|
||||
fe.title(title=f"{watch_label} - {title_suffix}")
|
||||
else:
|
||||
fe.title(title=watch_label)
|
||||
|
||||
# Clean and set content
|
||||
if scan_invalid_chars_in_rss(content):
|
||||
content = clean_entry_content(content)
|
||||
fe.content(content=content, type='CDATA')
|
||||
|
||||
# Set GUID
|
||||
fe.guid(guid, permalink=False)
|
||||
|
||||
# Set pubDate using the timestamp of this specific change
|
||||
dt = datetime.datetime.fromtimestamp(int(timestamp))
|
||||
dt = dt.replace(tzinfo=pytz.UTC)
|
||||
fe.pubDate(dt)
|
||||
|
||||
return content, watch_label
|
||||
|
||||
@@ -1,11 +1,5 @@
|
||||
from flask import make_response, request, url_for, redirect
|
||||
from feedgen.feed import FeedGenerator
|
||||
from loguru import logger
|
||||
import datetime
|
||||
import pytz
|
||||
import time
|
||||
|
||||
from ._util import generate_watch_guid, generate_watch_diff_content
|
||||
|
||||
|
||||
def construct_main_feed_routes(rss_blueprint, datastore):
|
||||
@@ -27,14 +21,24 @@ def construct_main_feed_routes(rss_blueprint, datastore):
|
||||
# from changedetectionio.auth_decorator import login_optionally_required
|
||||
@rss_blueprint.route("", methods=['GET'])
|
||||
def feed():
|
||||
now = time.time()
|
||||
# Always requires token set
|
||||
app_rss_token = datastore.data['settings']['application'].get('rss_access_token')
|
||||
rss_url_token = request.args.get('token')
|
||||
rss_content_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
from feedgen.feed import FeedGenerator
|
||||
from loguru import logger
|
||||
import time
|
||||
|
||||
if rss_url_token != app_rss_token:
|
||||
return "Access denied, bad token", 403
|
||||
from . import RSS_TEMPLATE_HTML_DEFAULT, RSS_TEMPLATE_PLAINTEXT_DEFAULT
|
||||
from ._util import (validate_rss_token, generate_watch_guid, get_rss_template,
|
||||
get_watch_label, build_notification_context, render_notification,
|
||||
populate_feed_entry, add_watch_categories)
|
||||
from ...notification_service import NotificationService
|
||||
|
||||
now = time.time()
|
||||
|
||||
# Validate token
|
||||
is_valid, error = validate_rss_token(datastore, request)
|
||||
if not is_valid:
|
||||
return error
|
||||
|
||||
rss_content_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
|
||||
limit_tag = request.args.get('tag', '').lower().strip()
|
||||
# Be sure limit_tag is a uuid
|
||||
@@ -52,7 +56,6 @@ def construct_main_feed_routes(rss_blueprint, datastore):
|
||||
continue
|
||||
if limit_tag and not limit_tag in watch['tags']:
|
||||
continue
|
||||
watch['uuid'] = uuid
|
||||
sorted_watches.append(watch)
|
||||
|
||||
sorted_watches.sort(key=lambda x: x.last_changed, reverse=False)
|
||||
@@ -61,6 +64,7 @@ def construct_main_feed_routes(rss_blueprint, datastore):
|
||||
fg.title('changedetection.io')
|
||||
fg.description('Feed description')
|
||||
fg.link(href='https://changedetection.io')
|
||||
notification_service = NotificationService(datastore=datastore, notification_q=False)
|
||||
|
||||
for watch in sorted_watches:
|
||||
|
||||
@@ -72,28 +76,28 @@ def construct_main_feed_routes(rss_blueprint, datastore):
|
||||
if not watch.viewed:
|
||||
# Re #239 - GUID needs to be individual for each event
|
||||
# @todo In the future make this a configurable link back (see work on BASE_URL https://github.com/dgtlmoon/changedetection.io/pull/228)
|
||||
guid = generate_watch_guid(watch)
|
||||
fe = fg.add_entry()
|
||||
|
||||
# Include a link to the diff page, they will have to login here to see if password protection is enabled.
|
||||
# Description is the page you watch, link takes you to the diff JS UI page
|
||||
# Dict val base_url will get overriden with the env var if it is set.
|
||||
ext_base_url = datastore.data['settings']['application'].get('active_base_url')
|
||||
# @todo fix
|
||||
|
||||
# Because we are called via whatever web server, flask should figure out the right path (
|
||||
watch_label = get_watch_label(datastore, watch)
|
||||
timestamp_to = dates[-1]
|
||||
timestamp_from = dates[-2]
|
||||
guid = generate_watch_guid(watch, timestamp_to)
|
||||
# Because we are called via whatever web server, flask should figure out the right path
|
||||
diff_link = {'href': url_for('ui.ui_views.diff_history_page', uuid=watch['uuid'], _external=True)}
|
||||
|
||||
fe.link(link=diff_link)
|
||||
# Get template and build notification context
|
||||
n_body_template = get_rss_template(datastore, watch, rss_content_format,
|
||||
RSS_TEMPLATE_HTML_DEFAULT, RSS_TEMPLATE_PLAINTEXT_DEFAULT)
|
||||
|
||||
content, watch_label = generate_watch_diff_content(watch, dates, rss_content_format, datastore)
|
||||
n_object = build_notification_context(watch, timestamp_from, timestamp_to,
|
||||
watch_label, n_body_template, rss_content_format)
|
||||
|
||||
fe.title(title=watch_label)
|
||||
fe.content(content=content, type='CDATA')
|
||||
fe.guid(guid, permalink=False)
|
||||
dt = datetime.datetime.fromtimestamp(int(watch.newest_history_key))
|
||||
dt = dt.replace(tzinfo=pytz.UTC)
|
||||
fe.pubDate(dt)
|
||||
# Render notification
|
||||
res = render_notification(n_object, notification_service, watch, datastore)
|
||||
|
||||
# Create and populate feed entry
|
||||
fe = fg.add_entry()
|
||||
populate_feed_entry(fe, watch, res['body'], guid, timestamp_to, link=diff_link)
|
||||
fe.title(title=watch_label) # Override title to not include suffix
|
||||
add_watch_categories(fe, watch, datastore)
|
||||
|
||||
response = make_response(fg.rss_str())
|
||||
response.headers.set('Content-Type', 'application/rss+xml;charset=utf-8')
|
||||
|
||||
@@ -1,9 +1,3 @@
|
||||
from flask import make_response, request, url_for
|
||||
from feedgen.feed import FeedGenerator
|
||||
import datetime
|
||||
import pytz
|
||||
|
||||
from ._util import generate_watch_guid, generate_watch_diff_content
|
||||
|
||||
|
||||
def construct_single_watch_routes(rss_blueprint, datastore):
|
||||
@@ -17,17 +11,31 @@ def construct_single_watch_routes(rss_blueprint, datastore):
|
||||
|
||||
@rss_blueprint.route("/watch/<string:uuid>", methods=['GET'])
|
||||
def rss_single_watch(uuid):
|
||||
"""
|
||||
Display the most recent change for a single watch as RSS feed.
|
||||
Returns RSS XML with a single entry showing the diff between the last two snapshots.
|
||||
"""
|
||||
# Always requires token set
|
||||
app_rss_token = datastore.data['settings']['application'].get('rss_access_token')
|
||||
rss_url_token = request.args.get('token')
|
||||
rss_content_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
import time
|
||||
|
||||
if rss_url_token != app_rss_token:
|
||||
return "Access denied, bad token", 403
|
||||
from flask import make_response, request
|
||||
from feedgen.feed import FeedGenerator
|
||||
from loguru import logger
|
||||
|
||||
from . import RSS_TEMPLATE_HTML_DEFAULT, RSS_TEMPLATE_PLAINTEXT_DEFAULT
|
||||
from ._util import (validate_rss_token, get_rss_template, get_watch_label,
|
||||
build_notification_context, render_notification,
|
||||
populate_feed_entry, add_watch_categories)
|
||||
from ...notification_service import NotificationService
|
||||
|
||||
"""
|
||||
Display the most recent changes for a single watch as RSS feed.
|
||||
Returns RSS XML with multiple entries showing diffs between consecutive snapshots.
|
||||
The number of entries is controlled by the rss_diff_length setting.
|
||||
"""
|
||||
now = time.time()
|
||||
|
||||
# Validate token
|
||||
is_valid, error = validate_rss_token(datastore, request)
|
||||
if not is_valid:
|
||||
return error
|
||||
|
||||
rss_content_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
|
||||
# Get the watch by UUID
|
||||
watch = datastore.data['watching'].get(uuid)
|
||||
@@ -42,30 +50,66 @@ def construct_single_watch_routes(rss_blueprint, datastore):
|
||||
# Add uuid to watch for proper functioning
|
||||
watch['uuid'] = uuid
|
||||
|
||||
# Generate the diff content using the shared helper function
|
||||
content, watch_label = generate_watch_diff_content(watch, dates, rss_content_format, datastore)
|
||||
# Get the number of diffs to include (default: 5)
|
||||
rss_diff_length = datastore.data['settings']['application'].get('rss_diff_length', 5)
|
||||
|
||||
# Create RSS feed with single entry
|
||||
# Calculate how many diffs we can actually show (limited by available history)
|
||||
# We need at least 2 snapshots to create 1 diff
|
||||
max_possible_diffs = len(dates) - 1
|
||||
num_diffs = min(rss_diff_length, max_possible_diffs) if rss_diff_length > 0 else max_possible_diffs
|
||||
|
||||
# Create RSS feed
|
||||
fg = FeedGenerator()
|
||||
fg.title(f'changedetection.io - {watch.label}')
|
||||
|
||||
# Set title: use "label (url)" if label differs from url, otherwise just url
|
||||
watch_url = watch.get('url', '')
|
||||
watch_label = get_watch_label(datastore, watch)
|
||||
|
||||
if watch_label != watch_url:
|
||||
feed_title = f'changedetection.io - {watch_label} ({watch_url})'
|
||||
else:
|
||||
feed_title = f'changedetection.io - {watch_url}'
|
||||
|
||||
fg.title(feed_title)
|
||||
fg.description('Changes')
|
||||
fg.link(href='https://changedetection.io')
|
||||
|
||||
# Add single entry for this watch
|
||||
guid = generate_watch_guid(watch)
|
||||
# Loop through history and create RSS entries for each diff
|
||||
# Add entries in reverse order because feedgen reverses them
|
||||
# This way, the newest change appears first in the final RSS
|
||||
|
||||
notification_service = NotificationService(datastore=datastore, notification_q=False)
|
||||
for i in range(num_diffs - 1, -1, -1):
|
||||
# Calculate indices for this diff (working backwards from newest)
|
||||
# i=0: compare dates[-2] to dates[-1] (most recent change)
|
||||
# i=1: compare dates[-3] to dates[-2] (previous change)
|
||||
# etc.
|
||||
date_index_to = -(i + 1)
|
||||
date_index_from = -(i + 2)
|
||||
timestamp_to = dates[date_index_to]
|
||||
timestamp_from = dates[date_index_from]
|
||||
|
||||
# Get template and build notification context
|
||||
n_body_template = get_rss_template(datastore, watch, rss_content_format,
|
||||
RSS_TEMPLATE_HTML_DEFAULT, RSS_TEMPLATE_PLAINTEXT_DEFAULT)
|
||||
|
||||
n_object = build_notification_context(watch, timestamp_from, timestamp_to,
|
||||
watch_label, n_body_template, rss_content_format)
|
||||
|
||||
# Render notification with date indices
|
||||
res = render_notification(n_object, notification_service, watch, datastore,
|
||||
date_index_from, date_index_to)
|
||||
|
||||
# Create and populate feed entry
|
||||
guid = f"{watch['uuid']}/{timestamp_to}"
|
||||
fe = fg.add_entry()
|
||||
|
||||
# Include a link to the diff page
|
||||
diff_link = {'href': url_for('ui.ui_views.diff_history_page', uuid=watch['uuid'], _external=True)}
|
||||
fe.link(link=diff_link)
|
||||
|
||||
fe.title(title=watch_label)
|
||||
fe.content(content=content, type='CDATA')
|
||||
fe.guid(guid, permalink=False)
|
||||
dt = datetime.datetime.fromtimestamp(int(watch.newest_history_key))
|
||||
dt = dt.replace(tzinfo=pytz.UTC)
|
||||
fe.pubDate(dt)
|
||||
title_suffix = f"Change @ {res['original_context']['change_datetime']}"
|
||||
populate_feed_entry(fe, watch, res.get('body', ''), guid, timestamp_to,
|
||||
link={'href': watch.get('url')}, title_suffix=title_suffix)
|
||||
add_watch_categories(fe, watch, datastore)
|
||||
|
||||
response = make_response(fg.rss_str())
|
||||
response.headers.set('Content-Type', 'application/rss+xml;charset=utf-8')
|
||||
logger.debug(f"RSS Single watch built in {time.time()-now:.2f}s")
|
||||
|
||||
return response
|
||||
|
||||
@@ -1,12 +1,3 @@
|
||||
from flask import make_response, request, url_for
|
||||
from feedgen.feed import FeedGenerator
|
||||
import datetime
|
||||
import pytz
|
||||
|
||||
|
||||
from ._util import generate_watch_guid, generate_watch_diff_content
|
||||
|
||||
|
||||
def construct_tag_routes(rss_blueprint, datastore):
|
||||
"""
|
||||
Construct RSS feed routes for tags.
|
||||
@@ -18,17 +9,26 @@ def construct_tag_routes(rss_blueprint, datastore):
|
||||
|
||||
@rss_blueprint.route("/tag/<string:tag_uuid>", methods=['GET'])
|
||||
def rss_tag_feed(tag_uuid):
|
||||
|
||||
from flask import make_response, request, url_for
|
||||
from feedgen.feed import FeedGenerator
|
||||
|
||||
from . import RSS_TEMPLATE_HTML_DEFAULT, RSS_TEMPLATE_PLAINTEXT_DEFAULT
|
||||
from ._util import (validate_rss_token, generate_watch_guid, get_rss_template,
|
||||
get_watch_label, build_notification_context, render_notification,
|
||||
populate_feed_entry, add_watch_categories)
|
||||
from ...notification_service import NotificationService
|
||||
|
||||
"""
|
||||
Display an RSS feed for all unviewed watches that belong to a specific tag.
|
||||
Returns RSS XML with entries for each unviewed watch with sufficient history.
|
||||
"""
|
||||
# Always requires token set
|
||||
app_rss_token = datastore.data['settings']['application'].get('rss_access_token')
|
||||
rss_url_token = request.args.get('token')
|
||||
rss_content_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
# Validate token
|
||||
is_valid, error = validate_rss_token(datastore, request)
|
||||
if not is_valid:
|
||||
return error
|
||||
|
||||
if rss_url_token != app_rss_token:
|
||||
return "Access denied, bad token", 403
|
||||
rss_content_format = datastore.data['settings']['application'].get('rss_content_format')
|
||||
|
||||
# Verify tag exists
|
||||
tag = datastore.data['settings']['application'].get('tags', {}).get(tag_uuid)
|
||||
@@ -42,9 +42,12 @@ def construct_tag_routes(rss_blueprint, datastore):
|
||||
fg.title(f'changedetection.io - {tag_title}')
|
||||
fg.description(f'Changes for watches tagged with {tag_title}')
|
||||
fg.link(href='https://changedetection.io')
|
||||
|
||||
notification_service = NotificationService(datastore=datastore, notification_q=False)
|
||||
# Find all watches with this tag
|
||||
for uuid, watch in datastore.data['watching'].items():
|
||||
#@todo This is wrong, it needs to sort by most recently changed and then limit it datastore.data['watching'].items().sorted(?)
|
||||
# So get all watches in this tag then sort
|
||||
|
||||
# Skip if watch doesn't have this tag
|
||||
if tag_uuid not in watch.get('tags', []):
|
||||
continue
|
||||
@@ -63,23 +66,32 @@ def construct_tag_routes(rss_blueprint, datastore):
|
||||
# Add uuid to watch for proper functioning
|
||||
watch['uuid'] = uuid
|
||||
|
||||
# Generate GUID for this entry
|
||||
guid = generate_watch_guid(watch)
|
||||
fe = fg.add_entry()
|
||||
|
||||
# Include a link to the diff page
|
||||
diff_link = {'href': url_for('ui.ui_views.diff_history_page', uuid=watch['uuid'], _external=True)}
|
||||
fe.link(link=diff_link)
|
||||
|
||||
# Generate diff content
|
||||
content, watch_label = generate_watch_diff_content(watch, dates, rss_content_format, datastore)
|
||||
# Get watch label
|
||||
watch_label = get_watch_label(datastore, watch)
|
||||
|
||||
fe.title(title=watch_label)
|
||||
fe.content(content=content, type='CDATA')
|
||||
fe.guid(guid, permalink=False)
|
||||
dt = datetime.datetime.fromtimestamp(int(watch.newest_history_key))
|
||||
dt = dt.replace(tzinfo=pytz.UTC)
|
||||
fe.pubDate(dt)
|
||||
# Get template and build notification context
|
||||
timestamp_to = dates[-1]
|
||||
timestamp_from = dates[-2]
|
||||
|
||||
# Generate GUID for this entry
|
||||
guid = generate_watch_guid(watch, timestamp_to)
|
||||
n_body_template = get_rss_template(datastore, watch, rss_content_format,
|
||||
RSS_TEMPLATE_HTML_DEFAULT, RSS_TEMPLATE_PLAINTEXT_DEFAULT)
|
||||
|
||||
n_object = build_notification_context(watch, timestamp_from, timestamp_to,
|
||||
watch_label, n_body_template, rss_content_format)
|
||||
|
||||
# Render notification
|
||||
res = render_notification(n_object, notification_service, watch, datastore)
|
||||
|
||||
# Create and populate feed entry
|
||||
fe = fg.add_entry()
|
||||
title_suffix = f"Change @ {res['original_context']['change_datetime']}"
|
||||
populate_feed_entry(fe, watch, res['body'], guid, timestamp_to, link=diff_link, title_suffix=title_suffix)
|
||||
add_watch_categories(fe, watch, datastore)
|
||||
|
||||
response = make_response(fg.rss_str())
|
||||
response.headers.set('Content-Type', 'application/rss+xml;charset=utf-8')
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
{% block content %}
|
||||
{% from '_helpers.html' import render_field, render_checkbox_field, render_button, render_time_schedule_form, render_ternary_field, render_fieldlist_with_inline_errors %}
|
||||
{% from '_common_fields.html' import render_common_settings_form %}
|
||||
{% from '_common_fields.html' import render_common_settings_form, show_token_placeholders %}
|
||||
<script>
|
||||
const notification_base_url="{{url_for('ui.ui_notification.ajax_callback_send_notification_test', mode="global-settings")}}";
|
||||
{% if emailprefix %}
|
||||
@@ -24,6 +24,7 @@
|
||||
<li class="tab"><a href="#filters">Global Filters</a></li>
|
||||
<li class="tab"><a href="#ui-options">UI Options</a></li>
|
||||
<li class="tab"><a href="#api">API</a></li>
|
||||
<li class="tab"><a href="#rss">RSS</a></li>
|
||||
<li class="tab"><a href="#timedate">Time & Date</a></li>
|
||||
<li class="tab"><a href="#proxies">CAPTCHA & Proxies</a></li>
|
||||
</ul>
|
||||
@@ -65,26 +66,13 @@
|
||||
|
||||
<div class="pure-control-group">
|
||||
{{ render_checkbox_field(form.application.form.shared_diff_access, class="shared_diff_access") }}
|
||||
<span class="pure-form-message-inline">Allow access to view watch diff page when password is enabled (Good for sharing the diff page)
|
||||
<span class="pure-form-message-inline">Allow access to the watch change history page when password is enabled (Good for sharing the diff page)
|
||||
</span>
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_checkbox_field(form.application.form.empty_pages_are_a_change) }}
|
||||
<span class="pure-form-message-inline">When a request returns no content, or the HTML does not contain any text, is this considered a change?</span>
|
||||
</div>
|
||||
<div class="grey-form-border">
|
||||
<div class="pure-control-group">
|
||||
{{ render_checkbox_field(form.application.form.rss_hide_muted_watches) }}
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.application.form.rss_content_format) }}
|
||||
<span class="pure-form-message-inline">Love RSS? Does your reader support HTML? Set it here</span>
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_checkbox_field(form.application.form.rss_reader_mode) }}
|
||||
<span class="pure-form-message-inline">When watching RSS/Atom feeds, convert them into clean text for better change detection.</span>
|
||||
</div>
|
||||
</div>
|
||||
</fieldset>
|
||||
</div>
|
||||
|
||||
@@ -229,6 +217,36 @@ nav
|
||||
</a>
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
<div class="tab-pane-inner" id="rss">
|
||||
<div class="pure-control-group">
|
||||
{{ render_checkbox_field(form.application.form.rss_hide_muted_watches) }}
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.application.form.rss_diff_length) }}
|
||||
<span class="pure-form-message-inline">Maximum number of history snapshots to include in the watch specific RSS feed.</span>
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_checkbox_field(form.application.form.rss_reader_mode) }}
|
||||
<span class="pure-form-message-inline">For watching other RSS feeds - When watching RSS/Atom feeds, convert them into clean text for better change detection.</span>
|
||||
</div>
|
||||
<div class="pure-control-group grey-form-border">
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.application.form.rss_content_format) }}
|
||||
<span class="pure-form-message-inline">Does your reader support HTML? Set it here</span>
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.application.form.rss_template_type) }}
|
||||
<span class="pure-form-message-inline">'System default' for the same template for all items, or re-use your "Notification Body" as the template.</span>
|
||||
</div>
|
||||
<div>
|
||||
{{ render_field(form.application.form.rss_template_override) }}
|
||||
{{ show_token_placeholders(extra_notification_token_placeholder_info=extra_notification_token_placeholder_info, suffix="-rss") }}
|
||||
</div>
|
||||
</div>
|
||||
<br>
|
||||
|
||||
|
||||
</div>
|
||||
<div class="tab-pane-inner" id="timedate">
|
||||
<div class="pure-control-group">
|
||||
|
||||
@@ -2,7 +2,6 @@ from flask import Blueprint, request, make_response
|
||||
import random
|
||||
from loguru import logger
|
||||
|
||||
from changedetectionio.notification_service import NotificationContextData, set_basic_notification_vars
|
||||
from changedetectionio.store import ChangeDetectionStore
|
||||
from changedetectionio.auth_decorator import login_optionally_required
|
||||
|
||||
@@ -15,7 +14,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
|
||||
@notification_blueprint.route("/notification/send-test/", methods=['POST'])
|
||||
@login_optionally_required
|
||||
def ajax_callback_send_notification_test(watch_uuid=None):
|
||||
|
||||
from changedetectionio.notification_service import NotificationContextData, set_basic_notification_vars
|
||||
# Watch_uuid could be unset in the case it`s used in tag editor, global settings
|
||||
import apprise
|
||||
from changedetectionio.notification.handler import process_notification
|
||||
@@ -97,31 +96,14 @@ def construct_blueprint(datastore: ChangeDetectionStore):
|
||||
n_object['as_async'] = False
|
||||
|
||||
# Same like in notification service, should be refactored
|
||||
dates = []
|
||||
dates = list(watch.history.keys())
|
||||
trigger_text = ''
|
||||
snapshot_contents = ''
|
||||
if watch:
|
||||
watch_history = watch.history
|
||||
dates = list(watch_history.keys())
|
||||
trigger_text = watch.get('trigger_text', [])
|
||||
# Add text that was triggered
|
||||
if len(dates):
|
||||
snapshot_contents = watch.get_history_snapshot(timestamp=dates[-1])
|
||||
else:
|
||||
snapshot_contents = "No snapshot/history available, the watch should fetch atleast once."
|
||||
|
||||
if len(trigger_text):
|
||||
from . import html_tools
|
||||
triggered_text = html_tools.get_triggered_text(content=snapshot_contents, trigger_text=trigger_text)
|
||||
if triggered_text:
|
||||
triggered_text = '\n'.join(triggered_text)
|
||||
|
||||
# Could be called as a 'test notification' with only 1 snapshot available
|
||||
prev_snapshot = "Example text: example test\nExample text: change detection is cool\nExample text: some more examples\n"
|
||||
current_snapshot = "Example text: example test\nExample text: change detection is fantastic\nExample text: even more examples\nExample text: a lot more examples"
|
||||
|
||||
|
||||
|
||||
if len(dates) > 1:
|
||||
prev_snapshot = watch.get_history_snapshot(timestamp=dates[-2])
|
||||
current_snapshot = watch.get_history_snapshot(timestamp=dates[-1])
|
||||
@@ -130,7 +112,8 @@ def construct_blueprint(datastore: ChangeDetectionStore):
|
||||
current_snapshot=current_snapshot,
|
||||
prev_snapshot=prev_snapshot,
|
||||
watch=watch,
|
||||
triggered_text=trigger_text))
|
||||
triggered_text=trigger_text,
|
||||
timestamp_changed=dates[-1] if dates else None))
|
||||
|
||||
|
||||
sent_obj = process_notification(n_object, datastore)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from loguru import logger
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import asyncio
|
||||
from changedetectionio import strtobool
|
||||
from changedetectionio.content_fetchers.exceptions import BrowserStepsInUnsupportedFetcher, EmptyReply, Non200ErrorCodeReceived
|
||||
@@ -76,6 +77,19 @@ class fetcher(Fetcher):
|
||||
if not is_binary:
|
||||
# Don't run this for PDF (and requests identified as binary) takes a _long_ time
|
||||
if not r.headers.get('content-type') or not 'charset=' in r.headers.get('content-type'):
|
||||
# For XML/RSS feeds, check the XML declaration for encoding attribute
|
||||
# This is more reliable than chardet which can misdetect UTF-8 as MacRoman
|
||||
content_type = r.headers.get('content-type', '').lower()
|
||||
if 'xml' in content_type or 'rss' in content_type:
|
||||
# Look for <?xml version="1.0" encoding="UTF-8"?>
|
||||
xml_encoding_match = re.search(rb'<\?xml[^>]+encoding=["\']([^"\']+)["\']', r.content[:200])
|
||||
if xml_encoding_match:
|
||||
r.encoding = xml_encoding_match.group(1).decode('ascii')
|
||||
else:
|
||||
# Default to UTF-8 for XML if no encoding found
|
||||
r.encoding = 'utf-8'
|
||||
else:
|
||||
# For other content types, use chardet
|
||||
encoding = chardet.detect(r.content)['encoding']
|
||||
if encoding:
|
||||
r.encoding = encoding
|
||||
|
||||
@@ -3,7 +3,7 @@ import re
|
||||
from loguru import logger
|
||||
from wtforms.widgets.core import TimeInput
|
||||
|
||||
from changedetectionio.blueprint.rss import RSS_FORMAT_TYPES
|
||||
from changedetectionio.blueprint.rss import RSS_FORMAT_TYPES, RSS_TEMPLATE_TYPE_OPTIONS, RSS_TEMPLATE_HTML_DEFAULT
|
||||
from changedetectionio.conditions.form import ConditionFormRow
|
||||
from changedetectionio.notification_service import NotificationContextData
|
||||
from changedetectionio.strtobool import strtobool
|
||||
@@ -1001,6 +1001,8 @@ class globalSettingsApplicationForm(commonSettingsForm):
|
||||
message="Should be atleast zero (disabled)")])
|
||||
|
||||
rss_content_format = SelectField('RSS Content format', choices=list(RSS_FORMAT_TYPES.items()))
|
||||
rss_template_type = SelectField('RSS <description> body built from', choices=list(RSS_TEMPLATE_TYPE_OPTIONS.items()))
|
||||
rss_template_override = TextAreaField('RSS "System default" template override', render_kw={"rows": "5", "placeholder": RSS_TEMPLATE_HTML_DEFAULT}, validators=[validators.Optional(), ValidateJinja2Template()])
|
||||
|
||||
removepassword_button = SubmitField('Remove password', render_kw={"class": "pure-button pure-button-primary"})
|
||||
render_anchor_tag_content = BooleanField('Render anchor tag content', default=False)
|
||||
@@ -1009,8 +1011,10 @@ class globalSettingsApplicationForm(commonSettingsForm):
|
||||
rss_hide_muted_watches = BooleanField('Hide muted watches from RSS feed', default=True,
|
||||
validators=[validators.Optional()])
|
||||
|
||||
rss_reader_mode = BooleanField('RSS reader mode ', default=False,
|
||||
validators=[validators.Optional()])
|
||||
rss_reader_mode = BooleanField('Enable RSS reader mode ', default=False, validators=[validators.Optional()])
|
||||
rss_diff_length = IntegerField(label='Number of changes to show in watch RSS feed',
|
||||
render_kw={"style": "width: 5em;"},
|
||||
validators=[validators.NumberRange(min=0, message="Should contain zero or more attempts")])
|
||||
|
||||
filter_failure_notification_threshold_attempts = IntegerField('Number of times the filter can be missing before sending a notification',
|
||||
render_kw={"style": "width: 5em;"},
|
||||
|
||||
@@ -55,6 +55,9 @@ class model(dict):
|
||||
'render_anchor_tag_content': False,
|
||||
'rss_access_token': None,
|
||||
'rss_content_format': RSS_CONTENT_FORMAT_DEFAULT,
|
||||
'rss_template_type': 'system_default',
|
||||
'rss_template_override': None,
|
||||
'rss_diff_length': 5,
|
||||
'rss_hide_muted_watches': True,
|
||||
'rss_reader_mode': False,
|
||||
'scheduler_timezone_default': None, # Default IANA timezone name
|
||||
|
||||
@@ -399,10 +399,14 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
apprise_input_format = NotifyFormat.TEXT.value
|
||||
requested_output_format = NotifyFormat.TEXT.value
|
||||
|
||||
|
||||
#@todo on null:// (only if its a 1 url with null) probably doesnt need to actually .add/setup/etc
|
||||
sent_objs.append({'title': n_title,
|
||||
'body': n_body,
|
||||
'url': url})
|
||||
'url': url,
|
||||
# So that we can do a null:// call and get back exactly what would have been sent
|
||||
'original_context': n_object })
|
||||
|
||||
if not url.startswith('null://'):
|
||||
apobj.add(url)
|
||||
|
||||
# Since the output is always based on the plaintext of the 'diff' engine, wrap it nicely.
|
||||
@@ -411,6 +415,7 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
if not '<pre' in n_body and not '<body' in n_body: # No custom HTML-ish body was setup already
|
||||
n_body = as_monospaced_html_email(content=n_body, title=n_title)
|
||||
|
||||
if not url.startswith('null://'):
|
||||
apobj.notify(
|
||||
title=n_title,
|
||||
body=n_body,
|
||||
@@ -438,6 +443,8 @@ def create_notification_parameters(n_object: NotificationContextData, datastore)
|
||||
if not isinstance(n_object, NotificationContextData):
|
||||
raise TypeError(f"Expected NotificationContextData, got {type(n_object)}")
|
||||
|
||||
ext_base_url = datastore.data['settings']['application'].get('active_base_url').strip('/')+'/'
|
||||
|
||||
watch = datastore.data['watching'].get(n_object['uuid'])
|
||||
if watch:
|
||||
watch_title = datastore.data['watching'][n_object['uuid']].label
|
||||
@@ -451,20 +458,29 @@ def create_notification_parameters(n_object: NotificationContextData, datastore)
|
||||
watch_title = 'Change Detection'
|
||||
watch_tag = ''
|
||||
|
||||
# Create URLs to customise the notification with
|
||||
# active_base_url - set in store.py data property
|
||||
base_url = datastore.data['settings']['application'].get('active_base_url')
|
||||
|
||||
watch_url = n_object['watch_url']
|
||||
|
||||
diff_url = "{}/diff/{}".format(base_url, n_object['uuid'])
|
||||
preview_url = "{}/preview/{}".format(base_url, n_object['uuid'])
|
||||
# Build URLs manually instead of using url_for() to avoid requiring a request context
|
||||
# This allows notifications to be processed in background threads
|
||||
uuid = n_object['uuid']
|
||||
|
||||
if n_object.get('timestamp_from') and n_object.get('timestamp_to'):
|
||||
# Include a link to the diff page with specific versions
|
||||
diff_url = f"{ext_base_url}diff/{uuid}?from_version={n_object['timestamp_from']}&to_version={n_object['timestamp_to']}"
|
||||
else:
|
||||
diff_url = f"{ext_base_url}diff/{uuid}"
|
||||
|
||||
preview_url = f"{ext_base_url}preview/{uuid}"
|
||||
edit_url = f"{ext_base_url}edit/{uuid}"
|
||||
|
||||
# @todo test that preview_url is correct when running in not-null mode?
|
||||
# if not, first time app loads i think it can set a flask context
|
||||
n_object.update(
|
||||
{
|
||||
'base_url': base_url,
|
||||
'base_url': ext_base_url,
|
||||
'diff_url': diff_url,
|
||||
'preview_url': preview_url,
|
||||
'preview_url': preview_url, #@todo include 'version='
|
||||
'edit_url': edit_url, #@todo also pause, also mute link
|
||||
'watch_tag': watch_tag if watch_tag is not None else '',
|
||||
'watch_title': watch_title if watch_title is not None else '',
|
||||
'watch_url': watch_url,
|
||||
|
||||
@@ -5,13 +5,54 @@ Notification Service Module
|
||||
Extracted from update_worker.py to provide standalone notification functionality
|
||||
for both sync and async workers
|
||||
"""
|
||||
import datetime
|
||||
|
||||
import pytz
|
||||
from loguru import logger
|
||||
import time
|
||||
|
||||
from changedetectionio.notification import default_notification_format, valid_notification_formats
|
||||
|
||||
|
||||
def _check_cascading_vars(datastore, var_name, watch):
|
||||
"""
|
||||
Check notification variables in cascading priority:
|
||||
Individual watch settings > Tag settings > Global settings
|
||||
"""
|
||||
from changedetectionio.notification import (
|
||||
USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH,
|
||||
default_notification_body,
|
||||
default_notification_title
|
||||
)
|
||||
|
||||
# Would be better if this was some kind of Object where Watch can reference the parent datastore etc
|
||||
v = watch.get(var_name)
|
||||
if v and not watch.get('notification_muted'):
|
||||
if var_name == 'notification_format' and v == USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH:
|
||||
return datastore.data['settings']['application'].get('notification_format')
|
||||
|
||||
return v
|
||||
|
||||
tags = datastore.get_all_tags_for_watch(uuid=watch.get('uuid'))
|
||||
if tags:
|
||||
for tag_uuid, tag in tags.items():
|
||||
v = tag.get(var_name)
|
||||
if v and not tag.get('notification_muted'):
|
||||
return v
|
||||
|
||||
if datastore.data['settings']['application'].get(var_name):
|
||||
return datastore.data['settings']['application'].get(var_name)
|
||||
|
||||
# Otherwise could be defaults
|
||||
if var_name == 'notification_format':
|
||||
return USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH
|
||||
if var_name == 'notification_body':
|
||||
return default_notification_body
|
||||
if var_name == 'notification_title':
|
||||
return default_notification_title
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# What is passed around as notification context, also used as the complete list of valid {{ tokens }}
|
||||
class NotificationContextData(dict):
|
||||
@@ -34,6 +75,8 @@ class NotificationContextData(dict):
|
||||
'preview_url': None,
|
||||
'screenshot': None,
|
||||
'triggered_text': None,
|
||||
'timestamp_from': None,
|
||||
'timestamp_to': None,
|
||||
'uuid': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', # Converted to 'watch_uuid' in create_notification_parameters
|
||||
'watch_mime_type': None,
|
||||
'watch_tag': None,
|
||||
@@ -73,7 +116,25 @@ class NotificationContextData(dict):
|
||||
super().__setitem__(key, value)
|
||||
|
||||
|
||||
def set_basic_notification_vars(snapshot_contents, current_snapshot, prev_snapshot, watch, triggered_text):
|
||||
def timestamp_to_localtime(timestamp):
|
||||
# Format the date using locale-aware formatting with timezone
|
||||
dt = datetime.datetime.fromtimestamp(int(timestamp))
|
||||
dt = dt.replace(tzinfo=pytz.UTC)
|
||||
|
||||
# Get local timezone-aware datetime
|
||||
local_tz = datetime.datetime.now().astimezone().tzinfo
|
||||
local_dt = dt.astimezone(local_tz)
|
||||
|
||||
# Format date with timezone - using strftime for locale awareness
|
||||
try:
|
||||
formatted_date = local_dt.strftime('%Y-%m-%d %H:%M:%S %Z')
|
||||
except:
|
||||
# Fallback if locale issues
|
||||
formatted_date = local_dt.isoformat()
|
||||
|
||||
return formatted_date
|
||||
|
||||
def set_basic_notification_vars(snapshot_contents, current_snapshot, prev_snapshot, watch, triggered_text, timestamp_changed=None):
|
||||
now = time.time()
|
||||
from changedetectionio import diff
|
||||
|
||||
@@ -89,6 +150,7 @@ def set_basic_notification_vars(snapshot_contents, current_snapshot, prev_snapsh
|
||||
'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False),
|
||||
'diff_removed_clean': diff.render_diff(prev_snapshot, current_snapshot, include_added=False, include_change_type_prefix=False),
|
||||
'screenshot': watch.get_screenshot() if watch and watch.get('notification_screenshot') else None,
|
||||
'change_datetime': timestamp_to_localtime(timestamp_changed) if timestamp_changed else None,
|
||||
'triggered_text': triggered_text,
|
||||
'uuid': watch.get('uuid') if watch else None,
|
||||
'watch_url': watch.get('url') if watch else None,
|
||||
@@ -114,7 +176,7 @@ class NotificationService:
|
||||
self.datastore = datastore
|
||||
self.notification_q = notification_q
|
||||
|
||||
def queue_notification_for_watch(self, n_object: NotificationContextData, watch):
|
||||
def queue_notification_for_watch(self, n_object: NotificationContextData, watch, date_index_from=-2, date_index_to=-1):
|
||||
"""
|
||||
Queue a notification for a watch with full diff rendering and template variables
|
||||
"""
|
||||
@@ -154,57 +216,23 @@ class NotificationService:
|
||||
current_snapshot = "Example text: example test\nExample text: change detection is fantastic\nExample text: even more examples\nExample text: a lot more examples"
|
||||
|
||||
if len(dates) > 1:
|
||||
prev_snapshot = watch.get_history_snapshot(timestamp=dates[-2])
|
||||
current_snapshot = watch.get_history_snapshot(timestamp=dates[-1])
|
||||
prev_snapshot = watch.get_history_snapshot(timestamp=dates[date_index_from])
|
||||
current_snapshot = watch.get_history_snapshot(timestamp=dates[date_index_to])
|
||||
|
||||
|
||||
n_object.update(set_basic_notification_vars(snapshot_contents=snapshot_contents,
|
||||
current_snapshot=current_snapshot,
|
||||
prev_snapshot=prev_snapshot,
|
||||
watch=watch,
|
||||
triggered_text=triggered_text))
|
||||
triggered_text=triggered_text,
|
||||
timestamp_changed=dates[date_index_to]))
|
||||
|
||||
if self.notification_q:
|
||||
logger.debug("Queued notification for sending")
|
||||
self.notification_q.put(n_object)
|
||||
|
||||
def _check_cascading_vars(self, var_name, watch):
|
||||
"""
|
||||
Check notification variables in cascading priority:
|
||||
Individual watch settings > Tag settings > Global settings
|
||||
"""
|
||||
from changedetectionio.notification import (
|
||||
USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH,
|
||||
default_notification_body,
|
||||
default_notification_title
|
||||
)
|
||||
|
||||
# Would be better if this was some kind of Object where Watch can reference the parent datastore etc
|
||||
v = watch.get(var_name)
|
||||
if v and not watch.get('notification_muted'):
|
||||
if var_name == 'notification_format' and v == USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH:
|
||||
return self.datastore.data['settings']['application'].get('notification_format')
|
||||
|
||||
return v
|
||||
|
||||
tags = self.datastore.get_all_tags_for_watch(uuid=watch.get('uuid'))
|
||||
if tags:
|
||||
for tag_uuid, tag in tags.items():
|
||||
v = tag.get(var_name)
|
||||
if v and not tag.get('notification_muted'):
|
||||
return v
|
||||
|
||||
if self.datastore.data['settings']['application'].get(var_name):
|
||||
return self.datastore.data['settings']['application'].get(var_name)
|
||||
|
||||
# Otherwise could be defaults
|
||||
if var_name == 'notification_format':
|
||||
return USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH
|
||||
if var_name == 'notification_body':
|
||||
return default_notification_body
|
||||
if var_name == 'notification_title':
|
||||
return default_notification_title
|
||||
|
||||
return None
|
||||
else:
|
||||
logger.debug("Not queued, no queue defined. Just returning processed data")
|
||||
return n_object
|
||||
|
||||
def send_content_changed_notification(self, watch_uuid):
|
||||
"""
|
||||
@@ -227,10 +255,11 @@ class NotificationService:
|
||||
# Should be a better parent getter in the model object
|
||||
|
||||
# Prefer - Individual watch settings > Tag settings > Global settings (in that order)
|
||||
n_object['notification_urls'] = self._check_cascading_vars('notification_urls', watch)
|
||||
n_object['notification_title'] = self._check_cascading_vars('notification_title', watch)
|
||||
n_object['notification_body'] = self._check_cascading_vars('notification_body', watch)
|
||||
n_object['notification_format'] = self._check_cascading_vars('notification_format', watch)
|
||||
# this change probably not needed?
|
||||
n_object['notification_urls'] = _check_cascading_vars(self.datastore, 'notification_urls', watch)
|
||||
n_object['notification_title'] = _check_cascading_vars(self.datastore,'notification_title', watch)
|
||||
n_object['notification_body'] = _check_cascading_vars(self.datastore,'notification_body', watch)
|
||||
n_object['notification_format'] = _check_cascading_vars(self.datastore,'notification_format', watch)
|
||||
|
||||
# (Individual watch) Only prepare to notify if the rules above matched
|
||||
queued = False
|
||||
@@ -269,7 +298,7 @@ Thanks - Your omniscient changedetection.io installation.
|
||||
n_object = NotificationContextData({
|
||||
'notification_title': 'Changedetection.io - Alert - CSS/xPath filter was not present in the page',
|
||||
'notification_body': body,
|
||||
'notification_format': self._check_cascading_vars('notification_format', watch),
|
||||
'notification_format': _check_cascading_vars(self.datastore, 'notification_format', watch),
|
||||
})
|
||||
n_object['markup_text_links_to_html_links'] = n_object.get('notification_format').startswith('html')
|
||||
|
||||
|
||||
@@ -29,16 +29,135 @@ def cdata_in_document_to_text(html_content: str, render_anchor_tag_content=False
|
||||
return re.sub(pattern, repl, html_content)
|
||||
|
||||
|
||||
# Jinja2 template for formatting RSS/Atom feed entries
|
||||
# Covers all common feedparser entry fields including namespaced elements
|
||||
# Outputs HTML that will be converted to text via html_to_text
|
||||
# @todo - This could be a UI setting in the future
|
||||
RSS_ENTRY_TEMPLATE = """<article class="rss-item" id="{{ entry.id|replace('"', '')|replace(' ', '-') }}">{%- if entry.title -%}Title: {{ entry.title }}<br>{%- endif -%}
|
||||
{%- if entry.link -%}<strong>Link:</strong> <a href="{{ entry.link }}">{{ entry.link }}</a><br>
|
||||
{%- endif -%}
|
||||
{%- if entry.id -%}
|
||||
<strong>Guid:</strong> {{ entry.id }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.published -%}
|
||||
<strong>PubDate:</strong> {{ entry.published }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.updated and entry.updated != entry.published -%}
|
||||
<strong>Updated:</strong> {{ entry.updated }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.author -%}
|
||||
<strong>Author:</strong> {{ entry.author }}<br>
|
||||
{%- elif entry.author_detail and entry.author_detail.name -%}
|
||||
<strong>Author:</strong> {{ entry.author_detail.name }}
|
||||
{%- if entry.author_detail.email %} ({{ entry.author_detail.email }}){% endif -%}
|
||||
<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.contributors -%}
|
||||
<strong>Contributors:</strong> {% for contributor in entry.contributors -%}
|
||||
{{ contributor.name if contributor.name else contributor }}
|
||||
{%- if not loop.last %}, {% endif -%}
|
||||
{%- endfor %}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.publisher -%}
|
||||
<strong>Publisher:</strong> {{ entry.publisher }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.rights -%}
|
||||
<strong>Rights:</strong> {{ entry.rights }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.license -%}
|
||||
<strong>License:</strong> {{ entry.license }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.language -%}
|
||||
<strong>Language:</strong> {{ entry.language }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.tags -%}
|
||||
<strong>Tags:</strong> {% for tag in entry.tags -%}
|
||||
{{ tag.term if tag.term else tag }}
|
||||
{%- if not loop.last %}, {% endif -%}
|
||||
{%- endfor %}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.category -%}
|
||||
<strong>Category:</strong> {{ entry.category }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.comments -%}
|
||||
<strong>Comments:</strong> <a href="{{ entry.comments }}">{{ entry.comments }}</a><br>
|
||||
{%- endif -%}
|
||||
{%- if entry.slash_comments -%}
|
||||
<strong>Comment Count:</strong> {{ entry.slash_comments }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.enclosures -%}
|
||||
<strong>Enclosures:</strong><br>
|
||||
{%- for enclosure in entry.enclosures %}
|
||||
- <a href="{{ enclosure.href }}">{{ enclosure.href }}</a> ({{ enclosure.type if enclosure.type else 'unknown type' }}
|
||||
{%- if enclosure.length %}, {{ enclosure.length }} bytes{% endif -%}
|
||||
)<br>
|
||||
{%- endfor -%}
|
||||
{%- endif -%}
|
||||
{%- if entry.media_content -%}
|
||||
<strong>Media:</strong><br>
|
||||
{%- for media in entry.media_content %}
|
||||
- <a href="{{ media.url }}">{{ media.url }}</a>
|
||||
{%- if media.type %} ({{ media.type }}){% endif -%}
|
||||
{%- if media.width and media.height %} {{ media.width }}x{{ media.height }}{% endif -%}
|
||||
<br>
|
||||
{%- endfor -%}
|
||||
{%- endif -%}
|
||||
{%- if entry.media_thumbnail -%}
|
||||
<strong>Thumbnail:</strong> <a href="{{ entry.media_thumbnail[0].url if entry.media_thumbnail[0].url else entry.media_thumbnail[0] }}">{{ entry.media_thumbnail[0].url if entry.media_thumbnail[0].url else entry.media_thumbnail[0] }}</a><br>
|
||||
{%- endif -%}
|
||||
{%- if entry.media_description -%}
|
||||
<strong>Media Description:</strong> {{ entry.media_description }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.itunes_duration -%}
|
||||
<strong>Duration:</strong> {{ entry.itunes_duration }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.itunes_author -%}
|
||||
<strong>Podcast Author:</strong> {{ entry.itunes_author }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_identifier -%}
|
||||
<strong>Identifier:</strong> {{ entry.dc_identifier }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_source -%}
|
||||
<strong>DC Source:</strong> {{ entry.dc_source }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_type -%}
|
||||
<strong>Type:</strong> {{ entry.dc_type }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_format -%}
|
||||
<strong>Format:</strong> {{ entry.dc_format }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_relation -%}
|
||||
<strong>Related:</strong> {{ entry.dc_relation }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_coverage -%}
|
||||
<strong>Coverage:</strong> {{ entry.dc_coverage }}<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.source and entry.source.title -%}
|
||||
<strong>Source:</strong> {{ entry.source.title }}
|
||||
{%- if entry.source.link %} (<a href="{{ entry.source.link }}">{{ entry.source.link }}</a>){% endif -%}
|
||||
<br>
|
||||
{%- endif -%}
|
||||
{%- if entry.dc_content -%}
|
||||
<strong>Content:</strong> {{ entry.dc_content | safe }}
|
||||
{%- elif entry.content and entry.content[0].value -%}
|
||||
<strong>Content:</strong> {{ entry.content[0].value | safe }}
|
||||
{%- elif entry.summary -%}
|
||||
<strong>Summary:</strong> {{ entry.summary | safe }}
|
||||
{%- endif -%}</article>
|
||||
"""
|
||||
|
||||
|
||||
def format_rss_items(rss_content: str, render_anchor_tag_content=False) -> str:
|
||||
"""
|
||||
Format RSS/Atom feed items in a readable text format using feedparser.
|
||||
Format RSS/Atom feed items in a readable text format using feedparser and Jinja2.
|
||||
|
||||
Converts RSS <item> or Atom <entry> elements to formatted text with:
|
||||
- <title> → <h1>Title</h1>
|
||||
- <link> → Link: [url]
|
||||
- <guid> → Guid: [id]
|
||||
- <pubDate> → PubDate: [date]
|
||||
- <description> or <content> → Raw HTML content (CDATA and entities automatically handled)
|
||||
Converts RSS <item> or Atom <entry> elements to formatted text with all available fields:
|
||||
- Basic fields: title, link, id/guid, published date, updated date
|
||||
- Author fields: author, author_detail, contributors, publisher
|
||||
- Content fields: content, summary, description
|
||||
- Metadata: tags, category, rights, license
|
||||
- Media: enclosures, media_content, media_thumbnail
|
||||
- Dublin Core elements: dc:creator, dc:date, dc:publisher, etc. (mapped by feedparser)
|
||||
|
||||
Args:
|
||||
rss_content: The RSS/Atom feed content
|
||||
@@ -49,65 +168,19 @@ def format_rss_items(rss_content: str, render_anchor_tag_content=False) -> str:
|
||||
"""
|
||||
try:
|
||||
import feedparser
|
||||
from xml.sax.saxutils import escape as xml_escape
|
||||
from changedetectionio.jinja2_custom import safe_jinja
|
||||
|
||||
# Parse the feed - feedparser handles all RSS/Atom variants, CDATA, entity unescaping, etc.
|
||||
feed = feedparser.parse(rss_content)
|
||||
|
||||
formatted_items = []
|
||||
|
||||
# Determine feed type for appropriate labels when fields are missing
|
||||
# feedparser sets feed.version to things like 'rss20', 'atom10', etc.
|
||||
# Determine feed type for appropriate labels
|
||||
is_atom = feed.version and 'atom' in feed.version
|
||||
|
||||
formatted_items = []
|
||||
for entry in feed.entries:
|
||||
item_parts = []
|
||||
|
||||
# Title - feedparser handles CDATA and entity unescaping automatically
|
||||
if hasattr(entry, 'title') and entry.title:
|
||||
item_parts.append(f'<h1>{xml_escape(entry.title)}</h1>')
|
||||
|
||||
# Link
|
||||
if hasattr(entry, 'link') and entry.link:
|
||||
item_parts.append(f'Link: {xml_escape(entry.link)}<br>')
|
||||
|
||||
# GUID/ID
|
||||
if hasattr(entry, 'id') and entry.id:
|
||||
item_parts.append(f'Guid: {xml_escape(entry.id)}<br>')
|
||||
|
||||
# Date - feedparser normalizes all date field names to 'published'
|
||||
if hasattr(entry, 'published') and entry.published:
|
||||
item_parts.append(f'PubDate: {xml_escape(entry.published)}<br>')
|
||||
|
||||
# Description/Content - feedparser handles CDATA and entity unescaping automatically
|
||||
# Only add "Summary:" label for Atom <summary> tags
|
||||
content = None
|
||||
add_label = False
|
||||
|
||||
if hasattr(entry, 'content') and entry.content:
|
||||
# Atom <content> - no label, just content
|
||||
content = entry.content[0].value if entry.content[0].value else None
|
||||
elif hasattr(entry, 'summary'):
|
||||
# Could be RSS <description> or Atom <summary>
|
||||
# feedparser maps both to entry.summary
|
||||
content = entry.summary if entry.summary else None
|
||||
# Only add "Summary:" label for Atom feeds (which use <summary> tag)
|
||||
if is_atom:
|
||||
add_label = True
|
||||
|
||||
# Add content with or without label
|
||||
if content:
|
||||
if add_label:
|
||||
item_parts.append(f'Summary:<br>{content}')
|
||||
else:
|
||||
item_parts.append(content)
|
||||
else:
|
||||
# No content - just show <none>
|
||||
item_parts.append('<none>')
|
||||
|
||||
# Join all parts of this item
|
||||
if item_parts:
|
||||
formatted_items.append('\n'.join(item_parts))
|
||||
# Render the entry using Jinja2 template
|
||||
rendered = safe_jinja.render(RSS_ENTRY_TEMPLATE, entry=entry, is_atom=is_atom)
|
||||
formatted_items.append(rendered.strip())
|
||||
|
||||
# Wrap each item in a div with classes (first, last, item-N)
|
||||
items_html = []
|
||||
@@ -122,7 +195,8 @@ def format_rss_items(rss_content: str, render_anchor_tag_content=False) -> str:
|
||||
|
||||
class_str = ' '.join(classes)
|
||||
items_html.append(f'<div class="{class_str}">{item}</div>')
|
||||
return '<html><body>\n'+"\n<br><br>".join(items_html)+'\n</body></html>'
|
||||
|
||||
return '<html><body>\n' + "\n<br>".join(items_html) + '\n</body></html>'
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Error formatting RSS items: {str(e)}")
|
||||
|
||||
@@ -70,7 +70,7 @@ echo "RUNNING WITH BASE_URL SET"
|
||||
# Now re-run some tests with BASE_URL enabled
|
||||
# Re #65 - Ability to include a link back to the installation, in the notification.
|
||||
export BASE_URL="https://really-unique-domain.io"
|
||||
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest -vv --maxfail=1 tests/test_notification.py
|
||||
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest -vv -s --maxfail=1 tests/test_notification.py
|
||||
|
||||
|
||||
# Re-run with HIDE_REFERER set - could affect login
|
||||
@@ -90,7 +90,7 @@ FETCH_WORKERS=130 pytest tests/test_history_consistency.py -v -l
|
||||
|
||||
# Check file:// will pickup a file when enabled
|
||||
echo "Hello world" > /tmp/test-file.txt
|
||||
ALLOW_FILE_URI=yes pytest tests/test_security.py
|
||||
ALLOW_FILE_URI=yes pytest -vv -s tests/test_security.py
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,53 +1,18 @@
|
||||
|
||||
{% from '_helpers.html' import render_field %}
|
||||
|
||||
{% macro render_common_settings_form(form, emailprefix, settings_application, extra_notification_token_placeholder_info) %}
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.notification_urls, rows=5, placeholder="Examples:
|
||||
Gitter - gitter://token/room
|
||||
Office365 - o365://TenantID:AccountEmail/ClientID/ClientSecret/TargetEmail
|
||||
AWS SNS - sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo
|
||||
SMTPS - mailtos://user:pass@mail.domain.com?to=receivingAddress@example.com",
|
||||
class="notification-urls" )
|
||||
}}
|
||||
<div class="pure-form-message-inline">
|
||||
<p>
|
||||
<strong>Tip:</strong> Use <a target="newwindow" href="https://github.com/caronc/apprise">AppRise Notification URLs</a> for notification to just about any service! <i><a target="newwindow" href="https://github.com/dgtlmoon/changedetection.io/wiki/Notification-configuration-notes">Please read the notification services wiki here for important configuration notes</a></i>.<br>
|
||||
</p>
|
||||
<div data-target="#advanced-help-notifications" class="toggle-show pure-button button-tag button-xsmall">Show advanced help and tips</div>
|
||||
<ul style="display: none" id="advanced-help-notifications">
|
||||
<li><code><a target="newwindow" href="https://github.com/caronc/apprise/wiki/Notify_discord">discord://</a></code> (or <code>https://discord.com/api/webhooks...</code>)) only supports a maximum <strong>2,000 characters</strong> of notification text, including the title.</li>
|
||||
<li><code><a target="newwindow" href="https://github.com/caronc/apprise/wiki/Notify_telegram">tgram://</a></code> bots can't send messages to other bots, so you should specify chat ID of non-bot user.</li>
|
||||
<li><code><a target="newwindow" href="https://github.com/caronc/apprise/wiki/Notify_telegram">tgram://</a></code> only supports very limited HTML and can fail when extra tags are sent, <a href="https://core.telegram.org/bots/api#html-style">read more here</a> (or use plaintext/markdown format)</li>
|
||||
<li><code>gets://</code>, <code>posts://</code>, <code>puts://</code>, <code>deletes://</code> for direct API calls (or omit the "<code>s</code>" for non-SSL ie <code>get://</code>) <a href="https://github.com/dgtlmoon/changedetection.io/wiki/Notification-configuration-notes#postposts">more help here</a></li>
|
||||
<li>Accepts the <code>{{ '{{token}}' }}</code> placeholders listed below</li>
|
||||
</ul>
|
||||
</div>
|
||||
<div class="notifications-wrapper">
|
||||
<a id="send-test-notification" class="pure-button button-secondary button-xsmall" >Send test notification</a> <div class="spinner" style="display: none;"></div>
|
||||
{% if emailprefix %}
|
||||
<a id="add-email-helper" class="pure-button button-secondary button-xsmall" >Add email <img style="height: 1em; display: inline-block" src="{{url_for('static_content', group='images', filename='email.svg')}}" alt="Add an email address"> </a>
|
||||
{% endif %}
|
||||
<a href="{{url_for('settings.notification_logs')}}" class="pure-button button-secondary button-xsmall" >Notification debug logs</a>
|
||||
<br>
|
||||
<div id="notification-test-log" style="display: none;"><span class="pure-form-message-inline">Processing..</span></div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="pure-control-group grey-form-border">
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.notification_title, class="m-d notification-title", placeholder=settings_application['notification_title']) }}
|
||||
<span class="pure-form-message-inline">Title for all notifications</span>
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.notification_body , rows=5, class="notification-body", placeholder=settings_application['notification_body']) }}
|
||||
<span class="pure-form-message-inline">Body for all notifications ‐ You can use <a target="newwindow" href="https://jinja.palletsprojects.com/en/3.0.x/templates/">Jinja2</a> templating in the notification title, body and URL, and tokens from below.
|
||||
</span>
|
||||
{% macro show_token_placeholders(extra_notification_token_placeholder_info, suffix="") %}
|
||||
|
||||
|
||||
</div>
|
||||
<div class="pure-controls">
|
||||
<div data-target="#notification-tokens-info" class="toggle-show pure-button button-tag button-xsmall">Show token/placeholders</div>
|
||||
<span class="pure-form-message-inline">
|
||||
Body for all notifications ‐ You can use <a target="newwindow" href="https://jinja.palletsprojects.com/en/3.0.x/templates/">Jinja2</a> templating in the notification title, body and URL, and tokens from below.
|
||||
</span><br>
|
||||
<div data-target="#notification-tokens-info{{ suffix }}" class="toggle-show pure-button button-tag button-xsmall">Show
|
||||
token/placeholders
|
||||
</div>
|
||||
<div class="pure-controls" style="display: none;" id="notification-tokens-info">
|
||||
</div>
|
||||
<div class="pure-controls" style="display: none;" id="notification-tokens-info{{ suffix }}">
|
||||
<table class="pure-table" id="token-table">
|
||||
<thead>
|
||||
<tr>
|
||||
@@ -90,7 +55,8 @@
|
||||
</tr>
|
||||
<tr>
|
||||
<td><code>{{ '{{diff_clean}}' }}</code></td>
|
||||
<td>The diff output - only changes, additions, and removals ‐ <i>Without (added) prefix or colors</i></td>
|
||||
<td>The diff output - only changes, additions, and removals ‐ <i>Without (added) prefix or colors</i>
|
||||
</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td><code>{{ '{{diff_added}}' }}</code></td>
|
||||
@@ -139,26 +105,72 @@
|
||||
{% endif %}
|
||||
</tbody>
|
||||
</table>
|
||||
<div class="pure-form-message-inline">
|
||||
<p>
|
||||
|
||||
<span class="pure-form-message-inline">
|
||||
Warning: Contents of <code>{{ '{{diff}}' }}</code>, <code>{{ '{{diff_removed}}' }}</code>, and <code>{{ '{{diff_added}}' }}</code> depend on how the difference algorithm perceives the change. <br>
|
||||
For example, an addition or removal could be perceived as a change in some cases. <a target="newwindow" href="https://github.com/dgtlmoon/changedetection.io/wiki/Using-the-%7B%7Bdiff%7D%7D,-%7B%7Bdiff_added%7D%7D,-and-%7B%7Bdiff_removed%7D%7D-notification-tokens">More Here</a> <br>
|
||||
</p>
|
||||
<p>
|
||||
For JSON payloads, use <strong>|tojson</strong> without quotes for automatic escaping, for example - <code>{ "name": {{ '{{ watch_title|tojson }}' }} }</code>
|
||||
</p>
|
||||
<p>
|
||||
URL encoding, use <strong>|urlencode</strong>, for example - <code>gets://hook-website.com/test.php?title={{ '{{ watch_title|urlencode }}' }}</code>
|
||||
</p>
|
||||
<p>
|
||||
Regular-expression replace, use <strong>|regex_replace</strong>, for example - <code>{{ "{{ \"hello world 123\" | regex_replace('[0-9]+', 'no-more-numbers') }}" }}</code>
|
||||
</p>
|
||||
<p>
|
||||
For a complete reference of all Jinja2 built-in filters, users can refer to the <a href="https://jinja.palletsprojects.com/en/3.1.x/templates/#builtin-filters">https://jinja.palletsprojects.com/en/3.1.x/templates/#builtin-filters</a>
|
||||
</p>
|
||||
</span>
|
||||
</div>
|
||||
{% endmacro %}
|
||||
|
||||
{% macro render_common_settings_form(form, emailprefix, settings_application, extra_notification_token_placeholder_info) %}
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.notification_urls, rows=5, placeholder="Examples:
|
||||
Gitter - gitter://token/room
|
||||
Office365 - o365://TenantID:AccountEmail/ClientID/ClientSecret/TargetEmail
|
||||
AWS SNS - sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo
|
||||
SMTPS - mailtos://user:pass@mail.domain.com?to=receivingAddress@example.com",
|
||||
class="notification-urls" )
|
||||
}}
|
||||
<div class="pure-form-message-inline">
|
||||
<p>
|
||||
<strong>Tip:</strong> Use <a target="newwindow" href="https://github.com/caronc/apprise">AppRise Notification URLs</a> for notification to just about any service! <i><a target="newwindow" href="https://github.com/dgtlmoon/changedetection.io/wiki/Notification-configuration-notes">Please read the notification services wiki here for important configuration notes</a></i>.<br>
|
||||
</p>
|
||||
<div data-target="#advanced-help-notifications" class="toggle-show pure-button button-tag button-xsmall">Show advanced help and tips</div>
|
||||
<ul style="display: none" id="advanced-help-notifications">
|
||||
<li><code><a target="newwindow" href="https://github.com/caronc/apprise/wiki/Notify_discord">discord://</a></code> (or <code>https://discord.com/api/webhooks...</code>)) only supports a maximum <strong>2,000 characters</strong> of notification text, including the title.</li>
|
||||
<li><code><a target="newwindow" href="https://github.com/caronc/apprise/wiki/Notify_telegram">tgram://</a></code> bots can't send messages to other bots, so you should specify chat ID of non-bot user.</li>
|
||||
<li><code><a target="newwindow" href="https://github.com/caronc/apprise/wiki/Notify_telegram">tgram://</a></code> only supports very limited HTML and can fail when extra tags are sent, <a href="https://core.telegram.org/bots/api#html-style">read more here</a> (or use plaintext/markdown format)</li>
|
||||
<li><code>gets://</code>, <code>posts://</code>, <code>puts://</code>, <code>deletes://</code> for direct API calls (or omit the "<code>s</code>" for non-SSL ie <code>get://</code>) <a href="https://github.com/dgtlmoon/changedetection.io/wiki/Notification-configuration-notes#postposts">more help here</a></li>
|
||||
<li>Accepts the <code>{{ '{{token}}' }}</code> placeholders listed below</li>
|
||||
</ul>
|
||||
</div>
|
||||
<div class="notifications-wrapper">
|
||||
<a id="send-test-notification" class="pure-button button-secondary button-xsmall" >Send test notification</a> <div class="spinner" style="display: none;"></div>
|
||||
{% if emailprefix %}
|
||||
<a id="add-email-helper" class="pure-button button-secondary button-xsmall" >Add email <img style="height: 1em; display: inline-block" src="{{url_for('static_content', group='images', filename='email.svg')}}" alt="Add an email address"> </a>
|
||||
{% endif %}
|
||||
<a href="{{url_for('settings.notification_logs')}}" class="pure-button button-secondary button-xsmall" >Notification debug logs</a>
|
||||
<br>
|
||||
<div id="notification-test-log" style="display: none;"><span class="pure-form-message-inline">Processing..</span></div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="pure-control-group grey-form-border">
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.notification_title, class="m-d notification-title", placeholder=settings_application['notification_title']) }}
|
||||
<span class="pure-form-message-inline">Title for all notifications</span>
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.notification_body , rows=5, class="notification-body", placeholder=settings_application['notification_body']) }}
|
||||
{{ show_token_placeholders(extra_notification_token_placeholder_info=extra_notification_token_placeholder_info) }}
|
||||
<div class="pure-form-message-inline">
|
||||
<ul>
|
||||
<li><span class="pure-form-message-inline">
|
||||
For JSON payloads, use <strong>|tojson</strong> without quotes for automatic escaping, for example - <code>{ "name": {{ '{{ watch_title|tojson }}' }} }</code>
|
||||
</span></li>
|
||||
<li><span class="pure-form-message-inline">
|
||||
URL encoding, use <strong>|urlencode</strong>, for example - <code>gets://hook-website.com/test.php?title={{ '{{ watch_title|urlencode }}' }}</code>
|
||||
</span></li>
|
||||
<li><span class="pure-form-message-inline">
|
||||
Regular-expression replace, use <strong>|regex_replace</strong>, for example - <code>{{ "{{ \"hello world 123\" | regex_replace('[0-9]+', 'no-more-numbers') }}" }}</code>
|
||||
</span></li>
|
||||
<li><span class="pure-form-message-inline">
|
||||
For a complete reference of all Jinja2 built-in filters, users can refer to the <a href="https://jinja.palletsprojects.com/en/3.1.x/templates/#builtin-filters">https://jinja.palletsprojects.com/en/3.1.x/templates/#builtin-filters</a>
|
||||
</span></li>
|
||||
</ul>
|
||||
<br>
|
||||
</div>
|
||||
<div class="">
|
||||
{{ render_field(form.notification_format , class="notification-format") }}
|
||||
<span class="pure-form-message-inline">Format for all notifications</span>
|
||||
</div>
|
||||
|
||||
@@ -253,43 +253,6 @@ def test_check_notification(client, live_server, measure_memory_usage, datastore
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
def test_notification_validation(client, live_server, measure_memory_usage, datastore_path):
|
||||
|
||||
time.sleep(1)
|
||||
|
||||
# re #242 - when you edited an existing new entry, it would not correctly show the notification settings
|
||||
# Add our URL to the import page
|
||||
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
|
||||
|
||||
# Re #360 some validation
|
||||
# res = client.post(
|
||||
# url_for("ui.ui_edit.edit_page", uuid="first"),
|
||||
# data={"notification_urls": 'json://localhost/foobar',
|
||||
# "notification_title": "",
|
||||
# "notification_body": "",
|
||||
# "notification_format": 'text',
|
||||
# "url": test_url,
|
||||
# "tag": "my tag",
|
||||
# "title": "my title",
|
||||
# "headers": "",
|
||||
# "fetch_backend": "html_requests"},
|
||||
# follow_redirects=True
|
||||
# )
|
||||
# assert b"Notification Body and Title is required when a Notification URL is used" in res.data
|
||||
|
||||
# cleanup for the next
|
||||
client.get(
|
||||
url_for("ui.form_delete", uuid="all"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
|
||||
def test_notification_urls_jinja2_apprise_integration(client, live_server, measure_memory_usage, datastore_path):
|
||||
|
||||
@@ -532,6 +495,7 @@ def test_single_send_test_notification_on_watch(client, live_server, measure_mem
|
||||
test_url = url_for('test_endpoint', _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)
|
||||
|
||||
test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}&+custom-header=123"
|
||||
# 1995 UTF-8 content should be encoded
|
||||
|
||||
@@ -329,6 +329,7 @@ def test_change_with_notification_values(client, live_server, measure_memory_usa
|
||||
|
||||
# Should see new tokens register
|
||||
res = client.get(url_for("settings.settings_page"))
|
||||
|
||||
assert b'{{restock.original_price}}' in res.data
|
||||
assert b'Original price at first check' in res.data
|
||||
|
||||
|
||||
@@ -67,8 +67,17 @@ def set_html_content(datastore_path, content):
|
||||
with open(os.path.join(datastore_path, "endpoint-content.txt"), "wb") as f:
|
||||
f.write(test_return_data.encode('utf-8'))
|
||||
|
||||
# def test_setup(client, live_server, measure_memory_usage, datastore_path):
|
||||
# live_server_setup(live_server) # Setup on conftest per function
|
||||
|
||||
def test_rss_feed_empty(client, live_server, measure_memory_usage, datastore_path):
|
||||
|
||||
rss_token = extract_rss_token_from_UI(client)
|
||||
|
||||
res = client.get(
|
||||
url_for("rss.feed", token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
assert res.status_code == 200
|
||||
assert b'xml' in res.data
|
||||
|
||||
def test_rss_and_token(client, live_server, measure_memory_usage, datastore_path):
|
||||
# # live_server_setup(live_server) # Setup on conftest per function
|
||||
@@ -76,18 +85,10 @@ def test_rss_and_token(client, live_server, measure_memory_usage, datastore_path
|
||||
set_original_response(datastore_path=datastore_path)
|
||||
rss_token = extract_rss_token_from_UI(client)
|
||||
|
||||
# Add our URL to the import page
|
||||
res = client.post(
|
||||
url_for("imports.import_page"),
|
||||
data={"urls": url_for('test_random_content_endpoint', _external=True)},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=url_for('test_random_content_endpoint', _external=True))
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
set_modified_response(datastore_path=datastore_path)
|
||||
time.sleep(1)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
@@ -314,3 +315,40 @@ def test_rss_single_watch_feed(client, live_server, measure_memory_usage, datast
|
||||
item = root.findall('.//item')[0].findtext('description')
|
||||
check_formatting(expected_type=k, content=item, url=test_url)
|
||||
|
||||
# Test RSS entry order: Create multiple versions and verify newest appears first
|
||||
for version in range(3, 6): # Create versions 3, 4, 5
|
||||
set_html_content(datastore_path, f"Version {version} content")
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
time.sleep(0.5) # Small delay to ensure different timestamps
|
||||
|
||||
# Fetch RSS feed again to verify order
|
||||
res = client.get(
|
||||
url_for('rss.rss_single_watch', uuid=uuid, token=app_rss_token),
|
||||
follow_redirects=False
|
||||
)
|
||||
assert res.status_code == 200
|
||||
|
||||
# Parse RSS and check order (newest first)
|
||||
root = ET.fromstring(res.data)
|
||||
items = root.findall('.//item')
|
||||
assert len(items) >= 3, f"Expected at least 3 items, got {len(items)}"
|
||||
|
||||
# Get descriptions from first 3 items
|
||||
descriptions = []
|
||||
for item in items[:3]:
|
||||
desc = item.findtext('description')
|
||||
descriptions.append(desc if desc else "")
|
||||
|
||||
# First item should contain newest change (Version 5)
|
||||
assert b"Version 5" in descriptions[0].encode() or "Version 5" in descriptions[0], \
|
||||
f"First item should show newest change (Version 5), but got: {descriptions[0][:200]}"
|
||||
|
||||
# Second item should contain Version 4
|
||||
assert b"Version 4" in descriptions[1].encode() or "Version 4" in descriptions[1], \
|
||||
f"Second item should show Version 4, but got: {descriptions[1][:200]}"
|
||||
|
||||
# Third item should contain Version 3
|
||||
assert b"Version 3" in descriptions[2].encode() or "Version 3" in descriptions[2], \
|
||||
f"Third item should show Version 3, but got: {descriptions[2][:200]}"
|
||||
|
||||
|
||||
@@ -7,6 +7,61 @@ from flask import url_for
|
||||
from .util import set_original_response, set_modified_response, live_server_setup, wait_for_all_checks, extract_rss_token_from_UI, \
|
||||
extract_UUID_from_client, delete_all_watches
|
||||
|
||||
def set_xmlns_purl_content(datastore_path, extra=""):
|
||||
data=f"""<rss xmlns:content="http://purl.org/rss/1.0/modules/content/" xmlns:dc="https://purl.org/dc/elements/1.1/" xmlns:media="http://search.yahoo.com/mrss/" xmlns:atom="http://www.w3.org/2005/Atom" version="2.0">
|
||||
<channel>
|
||||
<atom:link href="https://www.xxxxxxxtechxxxxx.com/feeds.xml" rel="self" type="application/rss+xml"/>
|
||||
<title>
|
||||
<![CDATA[ Latest from xxxxxxxtechxxxxx ]]>
|
||||
</title>
|
||||
<link>https://www.xxxxx.com</link>
|
||||
<description>
|
||||
<![CDATA[ All the latest content from the xxxxxxxtechxxxxx team ]]>
|
||||
</description>
|
||||
<lastBuildDate>Wed, 19 Nov 2025 15:00:00 +0000</lastBuildDate>
|
||||
<language>en</language>
|
||||
<item>
|
||||
<title>
|
||||
<![CDATA[ Sony Xperia 1 VII review: has Sony’s long-standing Xperia family lost what it takes to compete? ]]>
|
||||
</title>
|
||||
<dc:content>
|
||||
<![CDATA[ {{extra}} a little harder, dc-content. blue often quite tough and purple usually very difficult.</p><p>On the plus side, you don't technically need to solve the final one, as you'll be able to answer that one by a process of elimination. What's more, you can make up to four mistakes, which gives you a little bit of breathing room.</p><p>It's a little more involved than something like Wordle, however, and there are plenty of opportunities for the game to trip you up with tricks. For instance, watch out for homophones and other word games that could disguise the answers.</p><p>It's playable for free via the <a href="https://www.nytimes.com/games/strands" target="_blank">NYT Games site</a> on desktop or mobile.</p></article></section> ]]>
|
||||
</dc:content>
|
||||
<link>https://www.xxxxxxx.com/gaming/nyt-connections-today-answers-hints-20-november-2025</link>
|
||||
<description>
|
||||
<![CDATA[ Looking for NYT Connections answers and hints? Here's all you need to know to solve today's game, plus my commentary on the puzzles. ]]>
|
||||
</description>
|
||||
<guid isPermaLink="false">N2C2T6DztpWdxSdKpSUx89</guid>
|
||||
<enclosure url="https://cdn.mos.cms.futurecdn.net/RCGfdf3yhQ9W3MHbTRT6yk-1280-80.jpg" type="image/jpeg" length="0"/>
|
||||
<pubDate>Wed, 19 Nov 2025 15:00:00 +0000</pubDate>
|
||||
<category>
|
||||
<![CDATA[ Gaming ]]>
|
||||
</category>
|
||||
<dc:creator>
|
||||
<![CDATA[ Johnny Dee ]]>
|
||||
</dc:creator>
|
||||
<media:content type="image/jpeg" url="https://cdn.mos.cms.futurecdn.net/RCGfdf3yhQ9W3MHbTRT6yk-1280-80.jpg">
|
||||
<media:credit>
|
||||
<![CDATA[ New York Times ]]>
|
||||
</media:credit>
|
||||
<media:text>
|
||||
<![CDATA[ NYT Connections homescreen on a phone, on a purple background ]]>
|
||||
</media:text>
|
||||
<media:title type="plain">
|
||||
<![CDATA[ NYT Connections homescreen on a phone, on a purple background ]]>
|
||||
</media:title>
|
||||
</media:content>
|
||||
<media:thumbnail url="https://cdn.mos.cms.futurecdn.net/RCGfdf3yhQ9W3MHbTRT6yk-1280-80.jpg"/>
|
||||
</item>
|
||||
</channel>
|
||||
</rss>
|
||||
"""
|
||||
|
||||
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
|
||||
f.write(data)
|
||||
|
||||
|
||||
|
||||
|
||||
def set_original_cdata_xml(datastore_path):
|
||||
test_return_data = """<rss xmlns:atom="http://www.w3.org/2005/Atom" version="2.0">
|
||||
@@ -98,3 +153,26 @@ def test_rss_reader_mode_with_css_filters(client, live_server, measure_memory_us
|
||||
assert 'The days of Terminator and The Matrix' in snapshot_contents
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
def test_xmlns_purl_content(client, live_server, measure_memory_usage, datastore_path):
|
||||
set_xmlns_purl_content(datastore_path=datastore_path)
|
||||
|
||||
# Rarely do endpoints give the right header, usually just text/xml, so we check also for <rss
|
||||
# This also triggers the automatic CDATA text parser so the RSS goes back a nice content list
|
||||
#test_url = url_for('test_endpoint', content_type="text/xml; charset=UTF-8", _external=True)
|
||||
|
||||
# Because NO utf-8 was specified here, we should be able to recover it in requests or other somehow.
|
||||
test_url = url_for('test_endpoint', content_type="text/xml;", _external=True)
|
||||
live_server.app.config['DATASTORE'].data['settings']['application']['rss_reader_mode'] = True
|
||||
|
||||
# Add our URL to the import page
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, extras={'include_filters': [".last"]})
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
|
||||
wait_for_all_checks(client)
|
||||
|
||||
watch = live_server.app.config['DATASTORE'].data['watching'][uuid]
|
||||
dates = list(watch.history.keys())
|
||||
snapshot_contents = watch.get_history_snapshot(timestamp=dates[0])
|
||||
assert "Title: Sony Xperia 1 VII review: has Sony’s long-standing Xperia family lost what it takes to compete?" in snapshot_contents
|
||||
assert "dc-content" in snapshot_contents
|
||||
|
||||
347
changedetectionio/tests/test_rss_single_watch.py
Normal file
347
changedetectionio/tests/test_rss_single_watch.py
Normal file
@@ -0,0 +1,347 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import time
|
||||
import os
|
||||
import xml.etree.ElementTree as ET
|
||||
from flask import url_for
|
||||
|
||||
from .restock.test_restock import set_original_response
|
||||
from .util import live_server_setup, wait_for_all_checks, extract_rss_token_from_UI, extract_UUID_from_client, delete_all_watches, set_modified_response
|
||||
from ..notification import default_notification_format
|
||||
|
||||
|
||||
# Watch with no change should not break the output
|
||||
def test_rss_feed_empty(client, live_server, measure_memory_usage, datastore_path):
|
||||
set_original_response(datastore_path=datastore_path)
|
||||
rss_token = extract_rss_token_from_UI(client)
|
||||
test_url = url_for('test_endpoint', _external=True)
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
|
||||
# Request RSS feed for the single watch
|
||||
res = client.get(
|
||||
url_for("rss.rss_single_watch", uuid=uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
assert res.status_code == 400
|
||||
assert b'does not have enough history snapshots to show' in res.data
|
||||
|
||||
def test_rss_single_watch_order(client, live_server, measure_memory_usage, datastore_path):
|
||||
"""
|
||||
Test that single watch RSS feed shows changes in correct order (newest first).
|
||||
"""
|
||||
|
||||
# Create initial content
|
||||
def set_response(datastore_path, version):
|
||||
test_return_data = f"""<html>
|
||||
<body>
|
||||
<p>Version {version} content</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
|
||||
f.write(test_return_data)
|
||||
|
||||
# Start with version 1
|
||||
set_response(datastore_path, 1)
|
||||
|
||||
# Add a watch
|
||||
test_url = url_for('test_endpoint', _external=True) + "?order_test=1"
|
||||
res = client.post(
|
||||
url_for("ui.ui_views.form_quick_watch_add"),
|
||||
data={"url": test_url, "tags": 'test-tag'},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Watch added" in res.data
|
||||
|
||||
# Get the watch UUID
|
||||
watch_uuid = extract_UUID_from_client(client)
|
||||
|
||||
# Wait for initial check
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Create multiple versions by triggering changes
|
||||
for version in range(2, 6): # Create versions 2, 3, 4, 5
|
||||
set_response(datastore_path, version)
|
||||
res = client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
time.sleep(0.5) # Small delay to ensure different timestamps
|
||||
|
||||
# Get RSS token
|
||||
rss_token = extract_rss_token_from_UI(client)
|
||||
|
||||
# Request RSS feed for the single watch
|
||||
res = client.get(
|
||||
url_for("rss.rss_single_watch", uuid=watch_uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Should return valid RSS
|
||||
assert res.status_code == 200
|
||||
assert b"<?xml" in res.data or b"<rss" in res.data
|
||||
|
||||
# Parse the RSS/XML
|
||||
root = ET.fromstring(res.data)
|
||||
|
||||
# Find all items (RSS 2.0) or entries (Atom)
|
||||
items = root.findall('.//item')
|
||||
if not items:
|
||||
items = root.findall('.//{http://www.w3.org/2005/Atom}entry')
|
||||
|
||||
# Should have multiple items
|
||||
assert len(items) >= 3, f"Expected at least 3 items, got {len(items)}"
|
||||
|
||||
# Get the descriptions/content from first 3 items
|
||||
descriptions = []
|
||||
for item in items[:3]:
|
||||
# Try RSS format first
|
||||
desc = item.findtext('description')
|
||||
if not desc:
|
||||
# Try Atom format
|
||||
content_elem = item.find('{http://www.w3.org/2005/Atom}content')
|
||||
if content_elem is not None:
|
||||
desc = content_elem.text
|
||||
descriptions.append(desc if desc else "")
|
||||
|
||||
print(f"First item content: {descriptions[0][:100] if descriptions[0] else 'None'}")
|
||||
print(f"Second item content: {descriptions[1][:100] if descriptions[1] else 'None'}")
|
||||
print(f"Third item content: {descriptions[2][:100] if descriptions[2] else 'None'}")
|
||||
|
||||
# The FIRST item should contain the NEWEST change (Version 5)
|
||||
# The SECOND item should contain Version 4
|
||||
# The THIRD item should contain Version 3
|
||||
assert b"Version 5" in descriptions[0].encode() or "Version 5" in descriptions[0], \
|
||||
f"First item should show newest change (Version 5), but got: {descriptions[0][:200]}"
|
||||
|
||||
# Verify the order is correct
|
||||
assert b"Version 4" in descriptions[1].encode() or "Version 4" in descriptions[1], \
|
||||
f"Second item should show Version 4, but got: {descriptions[1][:200]}"
|
||||
|
||||
assert b"Version 3" in descriptions[2].encode() or "Version 3" in descriptions[2], \
|
||||
f"Third item should show Version 3, but got: {descriptions[2][:200]}"
|
||||
|
||||
# Clean up
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
def test_rss_categories_from_tags(client, live_server, measure_memory_usage, datastore_path):
|
||||
"""
|
||||
Test that RSS feeds include category tags from watch tags.
|
||||
"""
|
||||
|
||||
# Create initial content
|
||||
test_return_data = """<html>
|
||||
<body>
|
||||
<p>Test content for RSS categories</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
|
||||
f.write(test_return_data)
|
||||
|
||||
# Create some tags first
|
||||
res = client.post(
|
||||
url_for("tags.form_tag_add"),
|
||||
data={"name": "Security"},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
res = client.post(
|
||||
url_for("tags.form_tag_add"),
|
||||
data={"name": "Python"},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
res = client.post(
|
||||
url_for("tags.form_tag_add"),
|
||||
data={"name": "Tech News"},
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Add a watch with tags
|
||||
test_url = url_for('test_endpoint', _external=True) + "?category_test=1"
|
||||
res = client.post(
|
||||
url_for("ui.ui_views.form_quick_watch_add"),
|
||||
data={"url": test_url, "tags": "Security, Python, Tech News"},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Watch added" in res.data
|
||||
|
||||
# Get the watch UUID
|
||||
watch_uuid = extract_UUID_from_client(client)
|
||||
|
||||
# Wait for initial check
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Trigger one change
|
||||
test_return_data_v2 = """<html>
|
||||
<body>
|
||||
<p>Updated content for RSS categories</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
|
||||
f.write(test_return_data_v2)
|
||||
|
||||
res = client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Get RSS token
|
||||
rss_token = extract_rss_token_from_UI(client)
|
||||
|
||||
# Test 1: Check single watch RSS feed
|
||||
res = client.get(
|
||||
url_for("rss.rss_single_watch", uuid=watch_uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
assert res.status_code == 200
|
||||
assert b"<?xml" in res.data or b"<rss" in res.data
|
||||
|
||||
# Parse the RSS/XML
|
||||
root = ET.fromstring(res.data)
|
||||
|
||||
# Find all items
|
||||
items = root.findall('.//item')
|
||||
assert len(items) >= 1, "Expected at least 1 item in RSS feed"
|
||||
|
||||
# Get categories from first item
|
||||
categories = [cat.text for cat in items[0].findall('category')]
|
||||
|
||||
print(f"Found categories in single watch RSS: {categories}")
|
||||
|
||||
# Should have all three categories
|
||||
assert "Security" in categories, f"Expected 'Security' category, got: {categories}"
|
||||
assert "Python" in categories, f"Expected 'Python' category, got: {categories}"
|
||||
assert "Tech News" in categories, f"Expected 'Tech News' category, got: {categories}"
|
||||
assert len(categories) == 3, f"Expected 3 categories, got {len(categories)}: {categories}"
|
||||
|
||||
# Test 2: Check main RSS feed
|
||||
res = client.get(
|
||||
url_for("rss.feed", token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
assert res.status_code == 200
|
||||
|
||||
root = ET.fromstring(res.data)
|
||||
items = root.findall('.//item')
|
||||
assert len(items) >= 1, "Expected at least 1 item in main RSS feed"
|
||||
|
||||
# Get categories from first item in main feed
|
||||
categories = [cat.text for cat in items[0].findall('category')]
|
||||
|
||||
print(f"Found categories in main RSS feed: {categories}")
|
||||
|
||||
# Should have all three categories
|
||||
assert "Security" in categories, f"Expected 'Security' category in main feed, got: {categories}"
|
||||
assert "Python" in categories, f"Expected 'Python' category in main feed, got: {categories}"
|
||||
assert "Tech News" in categories, f"Expected 'Tech News' category in main feed, got: {categories}"
|
||||
|
||||
# Test 3: Check tag-specific RSS feed (should also have categories)
|
||||
# Get the tag UUID for "Security" and verify the tag feed also has categories
|
||||
from .util import get_UUID_for_tag_name
|
||||
security_tag_uuid = get_UUID_for_tag_name(client, name="Security")
|
||||
|
||||
if security_tag_uuid:
|
||||
res = client.get(
|
||||
url_for("rss.rss_tag_feed", tag_uuid=security_tag_uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
assert res.status_code == 200
|
||||
|
||||
root = ET.fromstring(res.data)
|
||||
items = root.findall('.//item')
|
||||
|
||||
if len(items) >= 1:
|
||||
categories = [cat.text for cat in items[0].findall('category')]
|
||||
print(f"Found categories in tag RSS feed: {categories}")
|
||||
|
||||
# Should still have all three categories
|
||||
assert "Security" in categories, f"Expected 'Security' category in tag feed, got: {categories}"
|
||||
assert "Python" in categories, f"Expected 'Python' category in tag feed, got: {categories}"
|
||||
assert "Tech News" in categories, f"Expected 'Tech News' category in tag feed, got: {categories}"
|
||||
|
||||
# Clean up
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
# RSS <description> should follow Main Settings -> Tag/Group -> Watch in that order of priority if set.
|
||||
def test_rss_single_watch_follow_notification_body(client, live_server, measure_memory_usage, datastore_path):
|
||||
rss_token = extract_rss_token_from_UI(client)
|
||||
|
||||
|
||||
res = client.post(
|
||||
url_for("settings.settings_page"),
|
||||
data={
|
||||
"application-fetch_backend": "html_requests",
|
||||
"application-minutes_between_check": 180,
|
||||
"application-notification_body": 'Boo yeah hello from main settings notification body<br>\nTitle: {{ watch_title }} changed',
|
||||
"application-notification_format": default_notification_format,
|
||||
"application-rss_template_type" : 'notification_body',
|
||||
"application-notification_urls": "",
|
||||
|
||||
},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b'Settings updated' in res.data
|
||||
|
||||
|
||||
set_original_response(datastore_path=datastore_path)
|
||||
|
||||
# Add our URL to the import page
|
||||
test_url = url_for('test_endpoint', _external=True)
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, tag="RSS-Custom")
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
set_modified_response(datastore_path=datastore_path)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
|
||||
# Request RSS feed for the single watch
|
||||
res = client.get(
|
||||
url_for("rss.rss_single_watch", uuid=uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Should return valid RSS
|
||||
assert res.status_code == 200
|
||||
assert b"<?xml" in res.data or b"<rss" in res.data
|
||||
|
||||
# Check it took the notification body from main settings ####
|
||||
item_description = ET.fromstring(res.data).findall('.//item')[0].findtext('description')
|
||||
assert "Boo yeah hello from main settings notification body" in item_description
|
||||
assert "Title: modified head" in item_description
|
||||
|
||||
|
||||
## Edit the tag notification_body, it should cascade up and become the RSS output
|
||||
res = client.post(
|
||||
url_for("tags.form_tag_edit_submit", uuid="first"),
|
||||
data={"name": "rss-custom",
|
||||
"notification_body": 'Hello from the group/tag level'},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated" in res.data
|
||||
res = client.get(
|
||||
url_for("rss.rss_single_watch", uuid=uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
item_description = ET.fromstring(res.data).findall('.//item')[0].findtext('description')
|
||||
assert 'Hello from the group/tag level' in item_description
|
||||
|
||||
# Override notification body at watch level and check ####
|
||||
res = client.post(
|
||||
url_for("ui.ui_edit.edit_page", uuid=uuid),
|
||||
data={"notification_body": "RSS body description set from watch level at notification body - {{ watch_title }}",
|
||||
"url": test_url,
|
||||
'fetch_backend': "html_requests",
|
||||
"time_between_check_use_default": "y"
|
||||
},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated watch." in res.data
|
||||
res = client.get(
|
||||
url_for("rss.rss_single_watch", uuid=uuid, token=rss_token, _external=True),
|
||||
follow_redirects=True
|
||||
)
|
||||
item_description = ET.fromstring(res.data).findall('.//item')[0].findtext('description')
|
||||
assert 'RSS body description set from watch level at notification body - modified head title' in item_description
|
||||
delete_all_watches(client)
|
||||
@@ -110,13 +110,14 @@ def get_UUID_for_tag_name(client, name):
|
||||
|
||||
# kinda funky, but works for now
|
||||
def extract_rss_token_from_UI(client):
|
||||
import re
|
||||
res = client.get(
|
||||
url_for("watchlist.index"),
|
||||
)
|
||||
m = re.search('token=(.+?)"', str(res.data))
|
||||
token_key = m.group(1)
|
||||
return token_key.strip()
|
||||
return client.application.config.get('DATASTORE').data['settings']['application'].get('rss_access_token')
|
||||
# import re
|
||||
# res = client.get(
|
||||
# url_for("watchlist.index"),
|
||||
# )
|
||||
# m = re.search('token=(.+?)"', str(res.data))
|
||||
# token_key = m.group(1)
|
||||
# return token_key.strip()
|
||||
|
||||
# kinda funky, but works for now
|
||||
def extract_UUID_from_client(client):
|
||||
|
||||
Reference in New Issue
Block a user