Spaces:
Running
Running
File size: 1,991 Bytes
8e7f687 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
# Pushover notification service extracted from app.py
import time
import requests
from collections import deque
from config.settings import PUSH_WINDOW_SECONDS, PUSH_MAX_IN_WINDOW, PUSH_DEDUPE_SECONDS
class PushoverService:
"""Rate-limited, de-duplicated Pushover notification service"""
def __init__(self, token: str, user: str):
self.token = token
self.user = user
# Rate limiting and deduplication state
self._recent_pushes = deque() # (timestamp, message)
self._last_seen = {} # message -> last_ts
def _should_push(self, message: str) -> bool:
"""Check if message should be sent based on rate limits and deduplication"""
now = time.time()
# De-dupe identical messages
last = self._last_seen.get(message)
if last and now - last < PUSH_DEDUPE_SECONDS:
return False
# Windowed rate limit
while self._recent_pushes and now - self._recent_pushes[0][0] > PUSH_WINDOW_SECONDS:
self._recent_pushes.popleft()
if len(self._recent_pushes) >= PUSH_MAX_IN_WINDOW:
return False
self._recent_pushes.append((now, message))
self._last_seen[message] = now
return True
def send(self, message: str) -> bool:
"""Send notification if rate limits allow. Returns True if sent, False if skipped."""
if not self._should_push(message):
return False
try:
response = requests.post(
"https://api.pushover.net/1/messages.json",
data={
"token": self.token,
"user": self.user,
"message": message[:1024], # Pushover message limit
},
timeout=10,
)
return response.status_code == 200
except Exception:
# Never crash chat due to notification errors
return False |