From 819901741d39aa907c25761d509cf377d5256606 Mon Sep 17 00:00:00 2001 From: Mark Hoekveen Date: Wed, 18 Dec 2024 19:52:44 +0100 Subject: [PATCH] adds pushover library For use in other scripts. Untested. --- bin/lib/pushover.py | 104 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 bin/lib/pushover.py diff --git a/bin/lib/pushover.py b/bin/lib/pushover.py new file mode 100644 index 0000000..641581b --- /dev/null +++ b/bin/lib/pushover.py @@ -0,0 +1,104 @@ +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