| """ |
| 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] = {} |
| 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": |
| |
| self._last_alert.pop(source_name, None) |
| continue |
|
|
| |
| now = time.time() |
| last = self._last_alert.get(source_name, 0) |
| if (now - last) < ALERT_COOLDOWN_MIN * 60: |
| continue |
|
|
| |
| 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 |
|
|
