66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""NTFY push notification plugin."""
|
|
|
|
import subprocess
|
|
import threading
|
|
|
|
EVENT_TAGS = {
|
|
"start": "red_circle",
|
|
"stop": "white_check_mark",
|
|
"error": "rotating_light",
|
|
"upload": "outbox_tray",
|
|
}
|
|
|
|
|
|
def send_notification(
|
|
url: str,
|
|
title: str,
|
|
message: str,
|
|
token: str = "",
|
|
priority: str = "default",
|
|
tags: str = "",
|
|
):
|
|
if not url:
|
|
return
|
|
|
|
cmd = [
|
|
"curl", "-s", "-o", "/dev/null", "--max-time", "10",
|
|
"-H", f"Title: {title}",
|
|
"-H", f"Priority: {priority}",
|
|
]
|
|
if tags:
|
|
cmd += ["-H", f"Tags: {tags}"]
|
|
cmd += ["-d", message]
|
|
if token:
|
|
cmd += ["-H", f"Authorization: Bearer {token}"]
|
|
cmd.append(url)
|
|
|
|
def _send():
|
|
try:
|
|
subprocess.run(cmd, capture_output=True, timeout=15, stdin=subprocess.DEVNULL)
|
|
except Exception:
|
|
pass
|
|
|
|
threading.Thread(target=_send, daemon=True).start()
|
|
|
|
|
|
class NtfyNotifier:
|
|
def __init__(self, config):
|
|
self.config = config
|
|
|
|
def _should_notify(self, event: str, job_ntfy_enabled: bool) -> bool:
|
|
if not job_ntfy_enabled:
|
|
return False
|
|
url = self.config.get("ntfy", "url", default="")
|
|
if not url:
|
|
return False
|
|
events = self.config.get("ntfy", "events", default="error")
|
|
return event in [e.strip() for e in events.split(",")]
|
|
|
|
def notify(self, event: str, title: str, message: str, job_ntfy_enabled: bool = False, priority: str = "default"):
|
|
if not self._should_notify(event, job_ntfy_enabled):
|
|
return
|
|
url = self.config.get("ntfy", "url", default="")
|
|
token = self.config.get("ntfy", "token", default="")
|
|
tag = EVENT_TAGS.get(event, "")
|
|
send_notification(url, title, message, token=token, priority=priority, tags=tag)
|