75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
"""
|
|
notifications.py - Multi-channel notification routing
|
|
|
|
Supported channels: Discord webhook, ntfy
|
|
"""
|
|
|
|
import core.postgres as postgres
|
|
import uuid
|
|
import requests
|
|
import time
|
|
|
|
|
|
def _sendToEnabledChannels(notif_settings, message):
|
|
"""Send message to all enabled channels. Returns True if at least one succeeded."""
|
|
sent = False
|
|
|
|
if notif_settings.get("discord_enabled") and notif_settings.get("discord_webhook"):
|
|
if discord.send(notif_settings["discord_webhook"], message):
|
|
sent = True
|
|
|
|
if notif_settings.get("ntfy_enabled") and notif_settings.get("ntfy_topic"):
|
|
if ntfy.send(notif_settings["ntfy_topic"], message):
|
|
sent = True
|
|
|
|
return sent
|
|
|
|
|
|
def getNotificationSettings(userUUID):
|
|
settings = postgres.select_one("notifications", {"user_uuid": userUUID})
|
|
if not settings:
|
|
return False
|
|
return settings
|
|
|
|
|
|
def setNotificationSettings(userUUID, data_dict):
|
|
existing = postgres.select_one("notifications", {"user_uuid": userUUID})
|
|
allowed = [
|
|
"discord_webhook",
|
|
"discord_enabled",
|
|
"ntfy_topic",
|
|
"ntfy_enabled",
|
|
]
|
|
updates = {k: v for k, v in data_dict.items() if k in allowed}
|
|
if not updates:
|
|
return False
|
|
if existing:
|
|
postgres.update("notifications", updates, {"user_uuid": userUUID})
|
|
else:
|
|
updates["id"] = str(uuid.uuid4())
|
|
updates["user_uuid"] = userUUID
|
|
postgres.insert("notifications", updates)
|
|
return True
|
|
|
|
|
|
class discord:
|
|
@staticmethod
|
|
def send(webhook_url, message):
|
|
try:
|
|
response = requests.post(webhook_url, json={"content": message})
|
|
return response.status_code == 204
|
|
except:
|
|
return False
|
|
|
|
|
|
class ntfy:
|
|
@staticmethod
|
|
def send(topic, message):
|
|
try:
|
|
response = requests.post(
|
|
f"https://ntfy.sh/{topic}", data=message.encode("utf-8")
|
|
)
|
|
return response.status_code == 200
|
|
except:
|
|
return False
|