Merge branch 'master' into history-preview-ignore-text-highlighting

changedetectionio/__init__.py

@@ -2,7 +2,7 @@
 
 # Read more https://github.com/dgtlmoon/changedetection.io/wiki
 
-__version__ = '0.50.24'
+__version__ = '0.50.27'
 
 from changedetectionio.strtobool import strtobool
 from json.decoder import JSONDecodeError

changedetectionio/api/__init__.py

@@ -37,6 +37,10 @@ def get_openapi_spec():
     from openapi_core import OpenAPI  # Lazy import - saves ~10.7 MB on startup
 
     spec_path = os.path.join(os.path.dirname(__file__), '../../docs/api-spec.yaml')
+    if not os.path.exists(spec_path):
+        # Possibly for pip3 packages
+        spec_path = os.path.join(os.path.dirname(__file__), '../docs/api-spec.yaml')
+
     with open(spec_path, 'r') as f:
         spec_dict = yaml.safe_load(f)
     _openapi_spec = OpenAPI.from_dict(spec_dict)

changedetectionio/html_tools.py

@@ -411,6 +411,9 @@ def strip_ignore_text(content, wordlist, mode="content"):
         return ''
 
     for k in wordlist:
+        # Skip empty strings to avoid matching everything
+        if not k or not k.strip():
+            continue
         # Is it a regex?
         res = re.search(PERL_STYLE_REGEX, k, re.IGNORECASE)
         if res:

changedetectionio/notification/handler.py

@@ -1,11 +1,75 @@
 
 import time
 import apprise
+from apprise import NotifyFormat
 from loguru import logger
 from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
 from ..notification_service import NotificationContextData
 
 
+def markup_text_links_to_html(body):
+    """
+    Convert plaintext to HTML with clickable links.
+    Uses Jinja2's escape and Markup for XSS safety.
+    """
+    from linkify_it import LinkifyIt
+    from markupsafe import Markup, escape
+
+    linkify = LinkifyIt()
+
+    # Match URLs in the ORIGINAL text (before escaping)
+    matches = linkify.match(body)
+
+    if not matches:
+        # No URLs, just escape everything
+        return Markup(escape(body))
+
+    result = []
+    last_index = 0
+
+    # Process each URL match
+    for match in matches:
+        # Add escaped text before the URL
+        if match.index > last_index:
+            text_part = body[last_index:match.index]
+            result.append(escape(text_part))
+
+        # Add the link with escaped URL (both in href and display)
+        url = match.url
+        result.append(Markup(f'<a href="{escape(url)}">{escape(url)}</a>'))
+
+        last_index = match.last_index
+
+    # Add remaining escaped text
+    if last_index < len(body):
+        result.append(escape(body[last_index:]))
+
+    # Join all parts
+    return str(Markup(''.join(str(part) for part in result)))
+
+def notification_format_align_with_apprise(n_format : str):
+    """
+    Correctly align changedetection's formats with apprise's formats
+    Probably these are the same - but good to be sure.
+    :param n_format:
+    :return:
+    """
+
+    if n_format.lower().startswith('html'):
+        # Apprise only knows 'html' not 'htmlcolor' etc, which shouldnt matter here
+        n_format = NotifyFormat.HTML
+    elif n_format.lower().startswith('markdown'):
+        # probably the same but just to be safe
+        n_format = NotifyFormat.MARKDOWN
+    elif n_format.lower().startswith('text'):
+        # probably the same but just to be safe
+        n_format = NotifyFormat.TEXT
+    else:
+        n_format = NotifyFormat.TEXT
+
+    # Must be str for apprise notify body_format
+    return str(n_format)
+
 def process_notification(n_object: NotificationContextData, datastore):
     from changedetectionio.jinja2_custom import render as jinja_render
     from . import default_notification_format_for_watch, default_notification_format, valid_notification_formats
@@ -30,7 +94,9 @@ def process_notification(n_object: NotificationContextData, datastore):
     # If we arrived with 'System default' then look it up
     if n_format == default_notification_format_for_watch and datastore.data['settings']['application'].get('notification_format') != default_notification_format_for_watch:
         # Initially text or whatever
-        n_format = datastore.data['settings']['application'].get('notification_format', valid_notification_formats[default_notification_format])
+        n_format = datastore.data['settings']['application'].get('notification_format', valid_notification_formats[default_notification_format]).lower()
+
+    n_format = notification_format_align_with_apprise(n_format=n_format)
 
     logger.trace(f"Complete notification body including Jinja and placeholders calculated in  {time.time() - now:.2f}s")
 
@@ -53,7 +119,11 @@ def process_notification(n_object: NotificationContextData, datastore):
 
             # Get the notification body from datastore
             n_body = jinja_render(template_str=n_object.get('notification_body', ''), **notification_parameters)
-            if n_object.get('notification_format', '').startswith('HTML'):
+
+            if n_object.get('markup_text_to_html'):
+                n_body = markup_text_links_to_html(body=n_body)
+
+            if n_format == str(NotifyFormat.HTML):
                 n_body = n_body.replace("\n", '<br>')
 
             n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)

changedetectionio/notification_service.py

@@ -9,6 +9,8 @@ for both sync and async workers
 from loguru import logger
 import time
 
+from changedetectionio.notification import default_notification_format
+
 # What is passed around as notification context, also used as the complete list of valid {{ tokens }}
 class NotificationContextData(dict):
     def __init__(self, initial_data=None, **kwargs):
@@ -28,7 +30,8 @@ class NotificationContextData(dict):
             'diff_url': None,
             'preview_url': None,
             'watch_tag': None,
-            'watch_title': None
+            'watch_title': None,
+            'markup_text_to_html': False, # If automatic conversion of plaintext to HTML should happen
         })
 
         # Apply any initial data passed in
@@ -123,10 +126,10 @@ class NotificationService:
         n_object.update({
             'current_snapshot': snapshot_contents,
             'diff': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep, html_colour=html_colour_enable, ignore_junk=ignore_junk),
-            'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False, line_feed_sep=line_feed_sep, ignore_junk=ignore_junk),
+            'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False, line_feed_sep=line_feed_sep, html_colour=html_colour_enable, ignore_junk=ignore_junk),
             'diff_full': diff.render_diff(prev_snapshot, current_snapshot, include_equal=True, line_feed_sep=line_feed_sep, html_colour=html_colour_enable, ignore_junk=ignore_junk),
             'diff_patch': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep, patch_format=True, ignore_junk=ignore_junk),
-            'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False, line_feed_sep=line_feed_sep, ignore_junk=ignore_junk),
+            'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False, line_feed_sep=line_feed_sep, html_colour=html_colour_enable, ignore_junk=ignore_junk),
             'screenshot': watch.get_screenshot() if watch and watch.get('notification_screenshot') else None,
             'triggered_text': triggered_text,
             'uuid': watch.get('uuid') if watch else None,
@@ -227,12 +230,25 @@ class NotificationService:
         if not watch:
             return
 
+        n_format = self.datastore.data['settings']['application'].get('notification_format', default_notification_format)
+        filter_list = ", ".join(watch['include_filters'])
+        # @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_to_html' is not needed
+        body = f"""Hello,
+
+Your configured CSS/xPath filters of '{filter_list}' for {{{{watch_url}}}} did not appear on the page after {threshold} attempts.
+
+It's possible the page changed layout and the filter needs updating ( Try the 'Visual Selector' tab )
+
+Edit link: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}
+
+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': "Your configured CSS/xPath filters of '{}' for {{{{watch_url}}}} did not appear on the page after {} attempts, did the page change layout?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\nThanks - Your omniscient changedetection.io installation :)\n".format(
-                ", ".join(watch['include_filters']),
-                threshold),
-            'notification_format': 'text'
+            'notification_body': body,
+            'notification_format': n_format,
+            'markup_text_to_html': n_format.lower().startswith('html')
         })
 
         if len(watch['notification_urls']):
@@ -261,13 +277,27 @@ class NotificationService:
         if not watch:
             return
         threshold = self.datastore.data['settings']['application'].get('filter_failure_notification_threshold_attempts')
+        n_format = self.datastore.data['settings']['application'].get('notification_format', default_notification_format).lower()
+        step = step_n + 1
+        # @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_to_html' is not needed
+
+        # {{{{ }}}} because this will be Jinja2 {{ }} tokens
+        body = f"""Hello,
+        
+Your configured browser step at position {step} for the web page watch {{{{watch_url}}}} did not appear on the page after {threshold} attempts, did the page change layout?
+
+The element may have moved and needs editing, or does it need a delay added?
+
+Edit link: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}
+
+Thanks - Your omniscient changedetection.io installation.
+"""
+
         n_object = NotificationContextData({
-            'notification_title': "Changedetection.io - Alert - Browser step at position {} could not be run".format(step_n+1),
-            'notification_body': "Your configured browser step at position {} for {{{{watch_url}}}} "
-                                 "did not appear on the page after {} attempts, did the page change layout? "
-                                 "Does it need a delay added?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\n"
-                                 "Thanks - Your omniscient changedetection.io installation :)\n".format(step_n+1, threshold),
-            'notification_format': 'text'
+            'notification_title': f"Changedetection.io - Alert - Browser step at position {step} could not be run",
+            'notification_body': body,
+            'notification_format': n_format,
+            'markup_text_to_html': n_format.lower().startswith('html')
         })
 
         if len(watch['notification_urls']):

changedetectionio/processors/text_json_diff/processor.py

@@ -324,13 +324,13 @@ class ContentProcessor:
                     append_pretty_line_formatting=not self.watch.is_source_type_url
                 )
 
-            # Raise error if filter returned nothing
-            if not filtered_content.strip():
-                raise FilterNotFoundInResponse(
-                    msg=self.filter_config.include_filters,
-                    screenshot=self.fetcher.screenshot,
-                    xpath_data=self.fetcher.xpath_data
-                )
+        # Raise error if filter returned nothing
+        if not filtered_content.strip():
+            raise FilterNotFoundInResponse(
+                msg=self.filter_config.include_filters,
+                screenshot=self.fetcher.screenshot,
+                xpath_data=self.fetcher.xpath_data
+            )
 
         return filtered_content
 

changedetectionio/run_basic_tests.sh

@@ -15,7 +15,7 @@ find tests/test_*py -type f|while read test_name
 do
   echo "TEST RUNNING $test_name"
   # REMOVE_REQUESTS_OLD_SCREENSHOTS disabled so that we can write a screenshot and send it in test_notifications.py without a real browser
-  REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest $test_name
+  REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest -vv -s --maxfail=1 --tb=long $test_name
 done
 
 echo "RUNNING WITH BASE_URL SET"
@@ -23,20 +23,20 @@ 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 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
 export HIDE_REFERER=True
-pytest tests/test_access_control.py
+pytest -vv -s --maxfail=1 tests/test_access_control.py
 
 # Re-run a few tests that will trigger brotli based storage
 export SNAPSHOT_BROTLI_COMPRESSION_THRESHOLD=5
-pytest tests/test_access_control.py
+pytest -vv -s --maxfail=1 tests/test_access_control.py
 REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest tests/test_notification.py
-pytest tests/test_backend.py
-pytest tests/test_rss.py
-pytest tests/test_unique_lines.py
+pytest -vv -s --maxfail=1 tests/test_backend.py
+pytest -vv -s --maxfail=1 tests/test_rss.py
+pytest -vv -s --maxfail=1 tests/test_unique_lines.py
 
 # Try high concurrency
 FETCH_WORKERS=130 pytest  tests/test_history_consistency.py -v -l

changedetectionio/store.py

@@ -228,26 +228,36 @@ class ChangeDetectionStore:
         d['settings']['application']['active_base_url'] = active_base_url.strip('" ')
         return d
 
+    from pathlib import Path
+
+    def delete_path(self, path: Path):
+        import shutil
+        """Delete a file or directory tree, including the path itself."""
+        if not path.exists():
+            return
+        if path.is_file() or path.is_symlink():
+            path.unlink(missing_ok=True)  # deletes a file or symlink
+        else:
+            shutil.rmtree(path, ignore_errors=True)  # deletes dir *and* its contents
+
     # Delete a single watch by UUID
     def delete(self, uuid):
         import pathlib
-        import shutil
 
         with self.lock:
             if uuid == 'all':
                 self.__data['watching'] = {}
                 time.sleep(1) # Mainly used for testing to allow all items to flush before running next test
-
-                # GitHub #30 also delete history records
                 for uuid in self.data['watching']:
                     path = pathlib.Path(os.path.join(self.datastore_path, uuid))
                     if os.path.exists(path):
-                        shutil.rmtree(path)
+                        self.delete(uuid)
 
             else:
                 path = pathlib.Path(os.path.join(self.datastore_path, uuid))
                 if os.path.exists(path):
-                    shutil.rmtree(path)
+                    self.delete_path(path)
+
                 del self.data['watching'][uuid]
 
         self.needs_write_urgent = True

changedetectionio/tests/test_filter_exist_changes.py

@@ -93,7 +93,9 @@ def test_filter_doesnt_exist_then_exists_should_get_notification(client, live_se
         "tags": "my tag",
         "title": "my title",
         "headers": "",
-        "include_filters": '.ticket-available',
+        # preprended with extra filter that intentionally doesn't match any entry,
+        # notification should still be sent even if first filter does not match (PR#3516)
+        "include_filters": ".non-matching-selector\n.ticket-available",
         "fetch_backend": "html_requests",
         "time_between_check_use_default": "y"})
 

changedetectionio/tests/test_filter_failure_notification.py

@@ -1,10 +1,8 @@
 import os
 import time
-from loguru import logger
 from flask import url_for
-from .util import set_original_response, live_server_setup, extract_UUID_from_client, wait_for_all_checks, \
-    wait_for_notification_endpoint_output
-from changedetectionio.model import App
+from .util import set_original_response,  wait_for_all_checks, wait_for_notification_endpoint_output
+from ..notification import valid_notification_formats
 
 
 def set_response_with_filter():
@@ -23,13 +21,14 @@ def set_response_with_filter():
         f.write(test_return_data)
     return None
 
-def run_filter_test(client, live_server, content_filter):
+def run_filter_test(client, live_server, content_filter, app_notification_format):
 
     # Response WITHOUT the filter ID element
     set_original_response()
+    live_server.app.config['DATASTORE'].data['settings']['application']['notification_format'] = app_notification_format
 
     # Goto the edit page, add our ignore text
-    notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'json')
+    notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'post')
 
     # Add our URL to the import page
     test_url = url_for('test_endpoint', _external=True)
@@ -127,8 +126,23 @@ def run_filter_test(client, live_server, content_filter):
     with open("test-datastore/notification.txt", 'r') as f:
         notification = f.read()
 
-    assert 'CSS/xPath filter was not present in the page' in notification
-    assert content_filter.replace('"', '\\"') in notification
+    assert 'Your configured CSS/xPath filters' in notification
+
+
+    # Text (or HTML conversion) markup to make the notifications a little nicer should have worked
+    if app_notification_format.startswith('html'):
+        # apprise should have used sax-escape (' instead of ", " etc), lets check it worked
+
+        from apprise.conversion import convert_between
+        from apprise.common import NotifyFormat
+        escaped_filter = convert_between(NotifyFormat.TEXT, NotifyFormat.HTML, content_filter)
+
+        assert escaped_filter in notification or escaped_filter.replace('"', '"') in notification
+        assert 'a href="' in notification # Quotes should still be there so the link works
+
+    else:
+        assert 'a href' not in notification
+        assert content_filter in notification
 
     # Remove it and prove that it doesn't trigger when not expected
     # It should register a change, but no 'filter not found'
@@ -159,14 +173,20 @@ def run_filter_test(client, live_server, content_filter):
     os.unlink("test-datastore/notification.txt")
 
 
-
-
 def test_check_include_filters_failure_notification(client, live_server, measure_memory_usage):
-#   #  live_server_setup(live_server) # Setup on conftest per function
-    run_filter_test(client, live_server,'#nope-doesnt-exist')
+    #   #  live_server_setup(live_server) # Setup on conftest per function
+    run_filter_test(client=client, live_server=live_server, content_filter='#nope-doesnt-exist', app_notification_format=valid_notification_formats.get('HTML Color'))
+    # Check markup send conversion didnt affect plaintext preference
+    run_filter_test(client=client, live_server=live_server, content_filter='#nope-doesnt-exist', app_notification_format=valid_notification_formats.get('Text'))
 
 def test_check_xpath_filter_failure_notification(client, live_server, measure_memory_usage):
-#   #  live_server_setup(live_server) # Setup on conftest per function
-    run_filter_test(client, live_server, '//*[@id="nope-doesnt-exist"]')
+    #   #  live_server_setup(live_server) # Setup on conftest per function
+    run_filter_test(client=client, live_server=live_server, content_filter='//*[@id="nope-doesnt-exist"]', app_notification_format=valid_notification_formats.get('HTML Color'))
 
 # Test that notification is never sent
+
+def test_basic_markup_from_text(client, live_server, measure_memory_usage):
+    # Test the notification error templates convert to HTML if needed (link activate)
+    from ..notification.handler import markup_text_links_to_html
+    x = markup_text_links_to_html("hello https://google.com")
+    assert 'a href' in x