neptune/core/lib/notify.py

import requests

from core.util import logs

NTFY_URL = "https://ntfy.sh"

log = logs.get_logger(__name__)


# Actual function to send a message to a topic
def ntfy_sendmsg(msg, **kwargs):
    title = kwargs.get("title", None)
    priority = kwargs.get("priority", None)
    tags = kwargs.get("tags", None)
    url = kwargs.get("url", NTFY_URL)
    topic = kwargs.get("topic", None)

    headers = {"Title": "Fisk"}
    if title:
        headers["Title"] = title
    if priority:
        headers["Priority"] = priority
    if tags:
        headers["Tags"] = tags
    try:
        requests.post(
            f"{url}/{topic}",
            data=msg,
            headers=headers,
        )
    except requests.exceptions.ConnectionError as e:
        log.error(f"Error sending notification: {e}")


def webhook_sendmsg(msg, url):
    try:
        requests.post(
            f"{url}",
            data=msg,
        )
    except requests.exceptions.ConnectionError as e:
        log.error(f"Error sending webhook: {e}")


# Sendmsg helper to send a message to a user's notification settings
def sendmsg(user, msg, **kwargs):
    service = kwargs.get("service", "ntfy")
    notification_settings = user.get_notification_settings()

    # No custom topic specified
    if "topic" not in kwargs:
        # No user topic set either
        if notification_settings.topic is None:
            # No topic set, so don't send
            return
        else:
            kwargs["topic"] = notification_settings.topic

    if "url" not in kwargs:
        if notification_settings.url is not None:
            kwargs["url"] = notification_settings.url

    if service == "ntfy":
        ntfy_sendmsg(msg, **kwargs)
    elif service == "webhook":
        webhook_sendmsg(msg, kwargs["url"])