105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
import os
|
|
import requests
|
|
import logging
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
|
|
class PushoverError(Exception):
|
|
"""Custom exception for Pushover-related errors"""
|
|
pass
|
|
|
|
def notify(
|
|
message: str,
|
|
title: Optional[str] = None,
|
|
attachment: Optional[Path] = None,
|
|
priority: int = 0
|
|
) -> bool:
|
|
"""
|
|
Send a push notification via Pushover API.
|
|
|
|
Args:
|
|
message: The message to send
|
|
title: Optional title (defaults to hostname)
|
|
attachment: Optional file path to attach (supports images, PDFs etc)
|
|
priority: Message priority (-2 to 2, default 0)
|
|
-2: Lowest, -1: Low, 0: Normal, 1: High, 2: Emergency
|
|
|
|
Returns:
|
|
bool: True if successful, False if failed
|
|
|
|
Raises:
|
|
PushoverError: If environment variables are missing or API call fails
|
|
"""
|
|
# Check for required environment variables
|
|
app_token = os.environ.get('PUSHOVER_APP_TOKEN')
|
|
user_token = os.environ.get('PUSHOVER_USER_TOKEN')
|
|
|
|
if not app_token or not user_token:
|
|
raise PushoverError("Missing required environment variables: PUSHOVER_APP_TOKEN, PUSHOVER_USER_TOKEN")
|
|
|
|
# Default title to hostname if not specified
|
|
if title is None:
|
|
title = os.uname().nodename
|
|
|
|
# Prepare the POST data
|
|
data = {
|
|
'token': app_token,
|
|
'user': user_token,
|
|
'message': message,
|
|
'title': title,
|
|
'priority': priority
|
|
}
|
|
|
|
files = {}
|
|
|
|
# Handle attachment if provided
|
|
if attachment:
|
|
if not attachment.exists():
|
|
raise PushoverError(f"Attachment file does not exist: {attachment}")
|
|
|
|
# Check file size (Pushover limit is 5MB)
|
|
if attachment.stat().st_size > 5 * 1024 * 1024:
|
|
raise PushoverError("Attachment exceeds Pushover's 5MB limit")
|
|
|
|
files = {'attachment': open(attachment, 'rb')}
|
|
|
|
try:
|
|
response = requests.post(
|
|
'https://api.pushover.net/1/messages.json',
|
|
data=data,
|
|
files=files,
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logging.error(f"Failed to send Pushover notification: {str(e)}")
|
|
return False
|
|
|
|
finally:
|
|
# Make sure we close the file if it was opened
|
|
if files and 'attachment' in files:
|
|
files['attachment'].close()
|
|
|
|
# Optional: Add rate limiting functionality
|
|
class RateLimiter:
|
|
"""Simple rate limiter for notifications"""
|
|
def __init__(self, cooldown_seconds: int = 300):
|
|
self.last_notification = {}
|
|
self.cooldown = cooldown_seconds
|
|
|
|
def can_notify(self, category: str) -> bool:
|
|
"""Check if enough time has passed since last notification in this category"""
|
|
import time
|
|
now = time.time()
|
|
if category not in self.last_notification:
|
|
self.last_notification[category] = now
|
|
return True
|
|
|
|
if now - self.last_notification[category] >= self.cooldown:
|
|
self.last_notification[category] = now
|
|
return True
|
|
|
|
return False
|