106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""Optional Discord-webhook and ntfy notification helpers."""
|
|
|
|
import logging
|
|
import os
|
|
from urllib.parse import quote, urlparse
|
|
import uuid
|
|
|
|
import requests
|
|
|
|
import core.postgres as postgres
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
REQUEST_TIMEOUT = float(os.getenv("NOTIFICATION_TIMEOUT", 10))
|
|
|
|
|
|
def _sendToEnabledChannels(notif_settings, message):
|
|
"""Send to enabled channels and return True when at least one succeeds."""
|
|
if not isinstance(notif_settings, dict):
|
|
return False
|
|
sent = False
|
|
if notif_settings.get("discord_enabled") and notif_settings.get(
|
|
"discord_webhook"
|
|
):
|
|
sent = discord.send(notif_settings["discord_webhook"], message) or sent
|
|
if notif_settings.get("ntfy_enabled") and notif_settings.get("ntfy_topic"):
|
|
sent = ntfy.send(notif_settings["ntfy_topic"], message) or sent
|
|
return sent
|
|
|
|
|
|
def getNotificationSettings(userUUID):
|
|
return postgres.select_one("notifications", {"user_uuid": userUUID}) or False
|
|
|
|
|
|
def setNotificationSettings(userUUID, data_dict):
|
|
if not isinstance(data_dict, dict):
|
|
return False
|
|
allowed = {
|
|
"discord_webhook",
|
|
"discord_enabled",
|
|
"ntfy_topic",
|
|
"ntfy_enabled",
|
|
}
|
|
updates = {key: value for key, value in data_dict.items() if key in allowed}
|
|
if not updates:
|
|
return False
|
|
if updates.get("discord_webhook"):
|
|
_validateDiscordWebhook(updates["discord_webhook"])
|
|
|
|
existing = postgres.select_one("notifications", {"user_uuid": userUUID})
|
|
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
|
|
|
|
|
|
def _validateDiscordWebhook(webhookURL):
|
|
parsed = urlparse(webhookURL)
|
|
allowedHosts = {"discord.com", "canary.discord.com", "ptb.discord.com"}
|
|
if parsed.scheme != "https" or parsed.hostname not in allowedHosts:
|
|
raise ValueError("Discord webhook must use an official HTTPS Discord host")
|
|
if not parsed.path.startswith("/api/webhooks/"):
|
|
raise ValueError("Invalid Discord webhook path")
|
|
return webhookURL
|
|
|
|
|
|
class discord:
|
|
@staticmethod
|
|
def send(webhook_url, message):
|
|
try:
|
|
_validateDiscordWebhook(webhook_url)
|
|
response = requests.post(
|
|
webhook_url,
|
|
json={"content": str(message)},
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
return response.status_code in {200, 204}
|
|
except (ValueError, requests.RequestException) as error:
|
|
logger.warning("Discord webhook delivery failed: %s", error)
|
|
return False
|
|
|
|
|
|
class ntfy:
|
|
@staticmethod
|
|
def send(topic, message):
|
|
if not isinstance(topic, str) or not topic.strip():
|
|
return False
|
|
baseURL = os.getenv("NTFY_BASE_URL", "https://ntfy.sh").rstrip("/")
|
|
headers = {}
|
|
if os.getenv("NTFY_TOKEN"):
|
|
headers["Authorization"] = f"Bearer {os.environ['NTFY_TOKEN']}"
|
|
try:
|
|
response = requests.post(
|
|
f"{baseURL}/{quote(topic.strip(), safe='')}",
|
|
data=str(message).encode("utf-8"),
|
|
headers=headers,
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
return 200 <= response.status_code < 300
|
|
except requests.RequestException as error:
|
|
logger.warning("ntfy delivery failed: %s", error)
|
|
return False
|