Refactor + tests for Apprise custom integration

changedetectionio/apprise_plugin/assets.py

@@ -0,0 +1,16 @@
+from apprise import AppriseAsset
+
+# Refer to:
+# https://github.com/caronc/apprise/wiki/Development_API#the-apprise-asset-object
+
+APPRISE_APP_ID = "changedetection.io"
+APPRISE_APP_DESC = "ChangeDetection.io best and simplest website monitoring and change detection"
+APPRISE_APP_URL = "https://changedetection.io"
+APPRISE_AVATAR_URL = "https://raw.githubusercontent.com/dgtlmoon/changedetection.io/master/changedetectionio/static/images/avatar-256x256.png"
+
+apprise_asset = AppriseAsset(
+    app_id=APPRISE_APP_ID,
+    app_desc=APPRISE_APP_DESC,
+    app_url=APPRISE_APP_URL,
+    image_url_logo=APPRISE_AVATAR_URL,
+)

changedetectionio/apprise_plugin/custom_handlers.py

@@ -0,0 +1,112 @@
+import json
+import re
+from urllib.parse import unquote_plus
+
+import requests
+from apprise.decorators import notify
+from apprise.utils.parse import parse_url as apprise_parse_url
+from loguru import logger
+from requests.structures import CaseInsensitiveDict
+
+SUPPORTED_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head"}
+
+
+def notify_supported_methods(func):
+    for method in SUPPORTED_HTTP_METHODS:
+        func = notify(on=method)(func)
+        # Add support for https, for each supported http method
+        func = notify(on=f"{method}s")(func)
+    return func
+
+
+def _get_auth(parsed_url: dict) -> str | tuple[str, str]:
+    user: str | None = parsed_url.get("user")
+    password: str | None = parsed_url.get("password")
+
+    if user is not None and password is not None:
+        return (unquote_plus(user), unquote_plus(password))
+
+    if user is not None:
+        return unquote_plus(user)
+
+    return ""
+
+
+def _get_headers(parsed_url: dict, body: str) -> CaseInsensitiveDict:
+    headers = CaseInsensitiveDict(
+        {unquote_plus(k).title(): unquote_plus(v) for k, v in parsed_url["qsd+"].items()}
+    )
+
+    # If Content-Type is not specified, guess if the body is a valid JSON
+    if headers.get("Content-Type") is None:
+        try:
+            json.loads(body)
+            headers["Content-Type"] = "application/json; charset=utf-8"
+        except Exception:
+            pass
+
+    return headers
+
+
+def _get_params(parsed_url: dict) -> CaseInsensitiveDict:
+    # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
+    # In Apprise, it relies on prefixing each request arg with "-", because it uses say &method=update as a flag for apprise
+    # but here we are making straight requests, so we need todo convert this against apprise's logic
+    params = CaseInsensitiveDict(
+        {
+            unquote_plus(k): unquote_plus(v)
+            for k, v in parsed_url["qsd"].items()
+            if k.strip("-") not in parsed_url["qsd-"]
+            and k.strip("+") not in parsed_url["qsd+"]
+        }
+    )
+
+    return params
+
+
+@notify_supported_methods
+def apprise_http_custom_handler(
+    body: str,
+    title: str,
+    notify_type: str,
+    meta: dict,
+    *args,
+    **kwargs,
+) -> bool:
+    url: str = meta.get("url")
+    schema: str = meta.get("schema")
+    method: str = re.sub(r"s$", "", schema).upper()
+
+    # Convert /foobar?+some-header=hello to proper header dictionary
+    parsed_url: dict[str, str | dict | None] | None = apprise_parse_url(url)
+    if parsed_url is None:
+        return False
+
+    auth = _get_auth(parsed_url=parsed_url)
+    headers = _get_headers(parsed_url=parsed_url, body=body)
+    params = _get_params(parsed_url=parsed_url)
+
+    url = re.sub(rf"^{schema}", "https" if schema.endswith("s") else "http", parsed_url.get("url"))
+
+    try:
+        response = requests.request(
+            method=method,
+            url=url,
+            auth=auth,
+            headers=headers,
+            params=params,
+            data=body.encode("utf-8") if isinstance(body, str) else body,
+        )
+
+        response.raise_for_status()
+
+        logger.info(f"Successfully sent custom notification to {url}")
+        return True
+
+    except requests.RequestException as e:
+        logger.error(f"Remote host error while sending custom notification to {url}: {e}")
+        return False
+
+    except Exception as e:
+        logger.error(f"Unexpected error occurred while sending custom notification to {url}: {e}")
+        return False

changedetectionio/blueprint/ui/notification.py

@@ -4,6 +4,7 @@
 
 from changedetectionio.store import ChangeDetectionStore
 from changedetectionio.auth_decorator import login_optionally_required
+from changedetectionio.notification import process_notification
 
 def construct_blueprint(datastore: ChangeDetectionStore):
     notification_blueprint = Blueprint('ui_notification', __name__, template_folder="../ui/templates")
@@ -17,11 +18,10 @@ def ajax_callback_send_notification_test(watch_uuid=None):
 
         # Watch_uuid could be unset in the case it`s used in tag editor, global settings
         import apprise
-        from changedetectionio.apprise_asset import asset
-        apobj = apprise.Apprise(asset=asset)
+        from ...apprise_plugin.assets import apprise_asset
+        from ...apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
+        apobj = apprise.Apprise(asset=apprise_asset)
 
-        # so that the custom endpoints are registered
-        from changedetectionio.apprise_plugin import apprise_custom_api_call_wrapper
         is_global_settings_form = request.args.get('mode', '') == 'global-settings'
         is_group_settings_form = request.args.get('mode', '') == 'group-settings'
 
@@ -90,7 +90,6 @@ def ajax_callback_send_notification_test(watch_uuid=None):
 
             n_object['as_async'] = False
             n_object.update(watch.extra_notification_token_values())
-            from changedetectionio.notification import process_notification
             sent_obj = process_notification(n_object, datastore)
 
         except Exception as e:

changedetectionio/forms.py

@@ -305,10 +305,10 @@ def __init__(self, message=None):
 
     def __call__(self, form, field):
         import apprise
-        apobj = apprise.Apprise()
+        from .apprise_plugin.assets import apprise_asset
+        from .apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
 
-        # so that the custom endpoints are registered
-        from .apprise_asset import asset
+        apobj = apprise.Apprise(asset=apprise_asset)
 
         for server_url in field.data:
             url = server_url.strip()

changedetectionio/notification.py

@@ -4,6 +4,9 @@
 import apprise
 from loguru import logger
 
+from .apprise_plugin.assets import APPRISE_AVATAR_URL
+from .apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
+from .safe_jinja import render as jinja_render
 
 valid_tokens = {
     'base_url': '',
@@ -39,10 +42,6 @@
 
 
 def process_notification(n_object, datastore):
-    # so that the custom endpoints are registered
-    from changedetectionio.apprise_plugin import apprise_custom_api_call_wrapper
-
-    from .safe_jinja import render as jinja_render
     now = time.time()
     if n_object.get('notification_timestamp'):
         logger.trace(f"Time since queued {now-n_object['notification_timestamp']:.3f}s")
@@ -66,12 +65,12 @@ def process_notification(n_object, datastore):
     # raise it as an exception
 
     sent_objs = []
-    from .apprise_asset import asset
+    from .apprise_plugin.assets import apprise_asset
 
     if 'as_async' in n_object:
-        asset.async_mode = n_object.get('as_async')
+        apprise_asset.async_mode = n_object.get('as_async')
 
-    apobj = apprise.Apprise(debug=True, asset=asset)
+    apobj = apprise.Apprise(debug=True, asset=apprise_asset)
 
     if not n_object.get('notification_urls'):
         return None
@@ -112,7 +111,7 @@ def process_notification(n_object, datastore):
                     and not url.startswith('get') \
                     and not url.startswith('delete') \
                     and not url.startswith('put'):
-                url += k + 'avatar_url=https://raw.githubusercontent.com/dgtlmoon/changedetection.io/master/changedetectionio/static/images/avatar-256x256.png'
+                url += k + f"avatar_url={APPRISE_AVATAR_URL}"
 
             if url.startswith('tgram://'):
                 # Telegram only supports a limit subset of HTML, remove the '<br>' we place in.

changedetectionio/tests/apprise/test_apprise_asset.py

@@ -0,0 +1,24 @@
+import pytest
+from apprise import AppriseAsset
+
+from changedetectionio.apprise_asset import (
+    APPRISE_APP_DESC,
+    APPRISE_APP_ID,
+    APPRISE_APP_URL,
+    APPRISE_AVATAR_URL,
+)
+
+
+@pytest.fixture(scope="function")
+def apprise_asset() -> AppriseAsset:
+    from changedetectionio.apprise_asset import apprise_asset
+
+    return apprise_asset
+
+
+def test_apprise_asset_init(apprise_asset: AppriseAsset):
+    assert isinstance(apprise_asset, AppriseAsset)
+    assert apprise_asset.app_id == APPRISE_APP_ID
+    assert apprise_asset.app_desc == APPRISE_APP_DESC
+    assert apprise_asset.app_url == APPRISE_APP_URL
+    assert apprise_asset.image_url_logo == APPRISE_AVATAR_URL