migrate from gotify to ntfy with image attachment support
- Replace GotifyNotifier with NtfyNotifier for push notifications - Add support for sending email image attachments to ntfy - Add NTFY_URL, NTFY_TOKEN, NTFY_TOPIC environment variables - Add get_mailpit_image() and get_mailpit_image_thumb() helpers - Sanitize message content for HTTP headers (remove newlines) - Remove all gotify references 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
81
app/utils/ntfy_api.py
Normal file
81
app/utils/ntfy_api.py
Normal file
@@ -0,0 +1,81 @@
|
||||
import requests
|
||||
|
||||
|
||||
class NtfyNotifier:
|
||||
|
||||
def __init__(self, server_url: str, api_token: str, topic: str):
|
||||
self.base_url = server_url.rstrip('/')
|
||||
self.token = api_token
|
||||
self.topic = topic
|
||||
|
||||
def send(self, title: str = "Test", message: str = "testing msg", priority: int = 5) -> bool:
|
||||
"""
|
||||
Send a text message to Ntfy.
|
||||
|
||||
Args:
|
||||
title (str): Title of the notification.
|
||||
message (str): Body of the notification.
|
||||
priority (int, optional): Message priority (1-5). Defaults to 5.
|
||||
|
||||
Returns:
|
||||
bool: True if the message was sent successfully, False otherwise.
|
||||
"""
|
||||
if not self.base_url or not self.token or not self.topic:
|
||||
print("[!] Missing NTFY_URL, NTFY_TOKEN, or NTFY_TOPIC")
|
||||
return False
|
||||
|
||||
url = f"{self.base_url}/{self.topic}"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.token}",
|
||||
"Title": title,
|
||||
"Priority": str(priority),
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(url, headers=headers, data=message, timeout=10)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except requests.RequestException as e:
|
||||
print(f"[!] Failed to send Ntfy message: {e}")
|
||||
return False
|
||||
|
||||
def send_with_image(self, title: str, message: str, image_data: bytes,
|
||||
filename: str = "image.jpg", priority: int = 5) -> bool:
|
||||
"""
|
||||
Send a notification with an attached image to Ntfy.
|
||||
|
||||
Args:
|
||||
title (str): Title of the notification.
|
||||
message (str): Body of the notification.
|
||||
image_data (bytes): Binary image data to attach.
|
||||
filename (str): Filename for the attachment. Defaults to "image.jpg".
|
||||
priority (int, optional): Message priority (1-5). Defaults to 5.
|
||||
|
||||
Returns:
|
||||
bool: True if the message was sent successfully, False otherwise.
|
||||
"""
|
||||
if not self.base_url or not self.token or not self.topic:
|
||||
print("[!] Missing NTFY_URL, NTFY_TOKEN, or NTFY_TOPIC")
|
||||
return False
|
||||
|
||||
url = f"{self.base_url}/{self.topic}"
|
||||
|
||||
# Sanitize message for HTTP header (no newlines/carriage returns allowed)
|
||||
safe_message = message.replace('\r', ' ').replace('\n', ' ').strip()
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.token}",
|
||||
"Title": title,
|
||||
"Priority": str(priority),
|
||||
"Filename": filename,
|
||||
"Message": safe_message,
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.put(url, headers=headers, data=image_data, timeout=10)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except requests.RequestException as e:
|
||||
print(f"[!] Failed to send Ntfy message with image: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user