Spaces:
Paused
Paused
| import time | |
| class SQLiteAlertManager: | |
| """Alert sender with cooldown and persistent alert history.""" | |
| def __init__(self, pusher, store, logger, stats, cooldowns=None, state_store=None): | |
| self.pusher = pusher | |
| self.store = store | |
| self.logger = logger | |
| self.stats = stats | |
| self.cooldowns = cooldowns or {} | |
| self.last_sent = {} | |
| self.state_store = state_store | |
| if self.state_store is not None: | |
| self.last_sent = dict(self.state_store.get("alert_sent_at") or {}) | |
| def emit(self, alerts): | |
| alerts = sorted(alerts, key=lambda a: (a.get("direction") != "跌",)) | |
| now = time.time() | |
| for alert in alerts: | |
| monitor_id = alert.get("monitor_id") or "crypto:unknown" | |
| key = f"{monitor_id}:{alert.get('key')}" | |
| cooldown_seconds = self.cooldowns.get(alert.get("category"), 5) * 60 | |
| if now - self.last_sent.get(key, 0) < cooldown_seconds: | |
| continue | |
| delivered = self.pusher.send_text(alert["text"]) | |
| self.last_sent[key] = now | |
| self.stats["alerts"] += 1 | |
| if not delivered: | |
| self.stats["notify_fails"] += 1 | |
| self.store.add_alert( | |
| monitor_id=monitor_id, | |
| text=alert["text"], | |
| category=alert.get("category"), | |
| title=alert.get("title"), | |
| direction=alert.get("direction"), | |
| delivered=delivered, | |
| error=None if delivered else "notifier returned false", | |
| ) | |
| if self.state_store is not None: | |
| self.state_store.set_kv("alert_sent_at", dict(self.last_sent)) | |
| self.logger( | |
| f"{alert.get('sev_emoji', '')} 推送[{alert.get('category')}] " | |
| f"{alert.get('title', key)}", | |
| monitor_id=monitor_id, | |
| ) | |