adds pushover library

For use in other scripts. Untested.
2024-12-18 19:52:44 +01:00
parent f826c4f008
commit 819901741d
1 changed files with 104 additions and 0 deletions
@@ -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