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(**kwargs):
    msg = kwargs.get("msg", None)
    notification_settings = kwargs.get("notification_settings")

    title = kwargs.get("title", None)
    priority = notification_settings.get("priority", None)
    tags = kwargs.get("tags", None)
    url = notification_settings.get("url") or NTFY_URL
    topic = notification_settings.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(**kwargs):
    msg = kwargs.get("msg", None)
    notification_settings = kwargs.get("notification_settings")
    url = notification_settings.get("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(**kwargs):
    user = kwargs.get("user", None)
    notification_settings = kwargs.get(
        "notification_settings", user.get_notification_settings().__dict__
    )
    if not notification_settings:
        return

    service = notification_settings.get("service")

    if service == "ntfy":
        ntfy_sendmsg(**kwargs)
    elif service == "webhook":
        webhook_sendmsg(**kwargs)