You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
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):
|
|
"""
|
|
Send a message to a topic using NTFY.
|
|
kwargs:
|
|
msg: Message to send, must be specified
|
|
notification_settings: Notification settings, must be specified
|
|
url: URL to NTFY server, can be None to use default
|
|
topic: Topic to send message to, must be specified
|
|
priority: Priority of message, optional
|
|
title: Title of message, optional
|
|
tags: Tags to add to message, optional
|
|
"""
|
|
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):
|
|
"""
|
|
Send a message to a webhook.
|
|
kwargs:
|
|
msg: Message to send, must be specified
|
|
notification_settings: Notification settings, must be specified
|
|
url: URL to webhook, must be specified"""
|
|
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):
|
|
"""
|
|
Send a message to a user's notification settings.
|
|
Fetches the user's default notification settings if not specified.
|
|
kwargs:
|
|
user: User to send message to, must be specified
|
|
notification_settings: Notification settings, optional
|
|
service: Notification service to use
|
|
|
|
kwargs for both services:
|
|
msg: Message to send, must be specified
|
|
notification_settings: Notification settings, must be specified
|
|
url: URL to NTFY server, can be None to use default
|
|
|
|
extra kwargs for ntfy:
|
|
title: Title of message, optional
|
|
tags: Tags to add to message, optional
|
|
notification_settings: Notification settings, must be specified
|
|
topic: Topic to send message to, must be specified
|
|
priority: Priority of message, optional
|
|
"""
|
|
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 == "none":
|
|
# Don't send anything
|
|
return
|
|
|
|
if service == "ntfy":
|
|
ntfy_sendmsg(**kwargs)
|
|
elif service == "webhook":
|
|
webhook_sendmsg(**kwargs)
|