File size: 3,620 Bytes
bfbaecb 063a7cd bfbaecb 063a7cd bfbaecb | 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
Email alerter — sends notifications when data sources go stale.
Activated by setting env vars: SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASSWORD, ALERT_EMAIL_TO.
Respects a per-source cooldown to avoid spamming.
"""
from __future__ import annotations
import logging
import os
import smtplib
import time
from email.mime.text import MIMEText
from typing import Any
from config.settings import ALERT_COOLDOWN_MIN
log = logging.getLogger("solarwine.alerter")
class EmailAlerter:
"""Sends email alerts when data sources are in 'red' status."""
def __init__(self):
self._last_alert: dict[str, float] = {} # source_name -> epoch of last alert
self._smtp_host = os.environ.get("SMTP_HOST", "")
self._smtp_port = int(os.environ.get("SMTP_PORT", "587"))
self._smtp_user = os.environ.get("SMTP_USER", "")
self._smtp_password = os.environ.get("SMTP_PASSWORD", "")
self._alert_to = os.environ.get("ALERT_EMAIL_TO", "")
self._alert_from = os.environ.get("ALERT_EMAIL_FROM", self._smtp_user)
@property
def enabled(self) -> bool:
return bool(self._smtp_host and self._alert_to)
def check_and_alert(self, status: dict[str, Any]) -> list[str]:
"""Check status and send alerts for red sources. Returns list of alerted sources."""
if not self.enabled:
return []
alerted: list[str] = []
sources = status.get("sources", {})
for source_name, info in sources.items():
if info.get("status") != "red":
# Source recovered — clear cooldown so next outage triggers immediately
self._last_alert.pop(source_name, None)
continue
# Check cooldown
now = time.time()
last = self._last_alert.get(source_name, 0)
if (now - last) < ALERT_COOLDOWN_MIN * 60:
continue
# Send alert
message = info.get("message", f"{source_name} is down")
age = info.get("age_minutes")
subject = f"[SolarWine] Data flow alert: {source_name}"
age_line = f"Age: {age:.0f} min\n" if age is not None else ""
body = (
f"Data source: {source_name}\n"
f"Status: RED\n"
f"{age_line}"
f"Detail: {message}\n"
f"\nChecked at: {status.get('checked_at', 'unknown')}\n"
f"Overall system status: {status.get('overall', 'unknown')}\n"
f"\n---\nSolarWine Data Flow Monitor"
)
if self._send_email(subject, body):
self._last_alert[source_name] = now
alerted.append(source_name)
return alerted
def _send_email(self, subject: str, body: str) -> bool:
"""Send an email via SMTP. Returns True on success."""
try:
msg = MIMEText(body, "plain", "utf-8")
msg["Subject"] = subject
msg["From"] = self._alert_from
msg["To"] = self._alert_to
with smtplib.SMTP(self._smtp_host, self._smtp_port, timeout=10) as server:
server.starttls()
if self._smtp_user and self._smtp_password:
server.login(self._smtp_user, self._smtp_password)
server.sendmail(self._alert_from, self._alert_to.split(","), msg.as_string())
log.info("Alert email sent: %s → %s", subject, self._alert_to)
return True
except Exception as exc:
log.error("Failed to send alert email: %s", exc)
return False
|