Initial commit

src/notifications/ntfy.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import requests
+
+from ..config import NtfyConfig
+
+
+class NtfyNotifier:
+    def __init__(self, config: NtfyConfig, timeout: float = 5.0) -> None:
+        self.config = config
+        self.timeout = timeout
+
+    def send(self, message: str, title: str = "Stall Watcher") -> tuple[bool, str]:
+        if not self.config.is_complete:
+            return False, "ntfy is not configured."
+
+        url = f"{self.config.url}/{self.config.topic}"
+        headers = {
+            "Authorization": f"Bearer {self.config.token}",
+            "Title": title,
+            "Priority": "high",
+        }
+        try:
+            response = requests.post(url, data=message.encode("utf-8"), headers=headers, timeout=self.timeout)
+            response.raise_for_status()
+            return True, "ntfy notification sent."
+        except requests.RequestException as exc:
+            return False, f"ntfy notification failed: {exc}"
+