FabledSteward/steward/core/mail.py

"""Emails Steward sends (#5335): an alert and the weekly report.

Each is multipart/alternative, a plain-text part and an HTML part rendered
from the same facts by templates/email/. The HTML keeps to what mail
clients draw: a table layout, inline styles and no remote assets, so
opening the mail fetches nothing (no fonts, images or stylesheets). Mail
clients do not read CSS variables, so the colours are the design system's
light theme resolved to plain values from theme.css (palette()).
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone, tzinfo
from email.message import EmailMessage
from functools import lru_cache
from pathlib import Path

from jinja2 import Environment, FileSystemLoader, select_autoescape

_STEWARD = Path(__file__).resolve().parent.parent

# The part each colour plays in an email, and the theme token it is.
_ROLES = {
    "page": "--fs-surface-page", "card": "--fs-surface-raised", "border": "--fs-border-color",
    "text": "--fs-text-primary", "muted": "--fs-text-secondary", "faint": "--fs-text-tertiary",
    "action": "--fs-action-primary", "on_action": "--fs-text-on-action",
    "ok": "--fs-success", "warn": "--fs-warning", "crit": "--fs-error", "info": "--fs-info",
}
# A "dim" state is drawn in secondary text: the faintest is under 4.5:1 on the card.
_TONE_DIM = "muted"

_env = Environment(loader=FileSystemLoader(_STEWARD / "templates" / "email"),
                   autoescape=select_autoescape(["html"]), trim_blocks=True, lstrip_blocks=True,
                   keep_trailing_newline=True)


def _block(css: str, selector: str) -> dict[str, str]:
    i = css.index(selector + " {")
    body = re.sub(r"/\*.*?\*/", "", css[i:css.index("\n}", i)], flags=re.S)
    return {k: v.strip() for k, v in re.findall(r"(--[\w-]+)\s*:\s*([^;]+);", body)}


@lru_cache(maxsize=1)
def palette() -> dict[str, str]:
    """{role: colour} from theme.css, the light theme over :root with var()
    followed to its value, and the body font as "font"."""
    css = (_STEWARD / "static" / "theme.css").read_text()
    tokens = {**_block(css, ":root"), **_block(css, '[data-theme="light"]')}

    def resolve(name: str) -> str:
        m = re.fullmatch(r"var\((--[\w-]+)\)", tokens[name])
        return resolve(m.group(1)) if m else tokens[name]
    colours = {role: resolve(token) for role, token in _ROLES.items()}
    colours["dim"] = colours[_TONE_DIM]
    colours["font"] = tokens["--fs-font-body"]   # the stack's fallbacks; nothing is loaded
    return colours


# ── Who gets an email (milestone 495) ─────────────────────────────────────────

KINDS = ("alerts", "report")
_REAL_ADDRESS = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+")


@dataclass(frozen=True)
class Recipient:
    """One address an email goes to, and the zone its times are written in.
    `user` is the account's username, "" for an "Also send to" address."""
    address: str
    tz: tzinfo
    show_utc: bool
    user: str = ""


def is_real_address(address: str | None) -> bool:
    """Whether mail can reach it: an SSO account its provider gave no email
    holds a placeholder like "name@oidc", which has no domain to deliver to."""
    return bool(address and _REAL_ADDRESS.fullmatch(address.strip()))


def wants(user, kind: str, settings: dict) -> bool:
    """Whether an account is emailed `kind` ("alerts" or "report"): its own
    choice, or else its role's default."""
    choice = getattr(user, f"notify_{kind}")
    if choice is None:
        return bool(settings.get(f"notify.{kind}.{user.role.value}", True))
    return choice


def pick_recipients(users, kind: str, settings: dict, extra: list[str], extra_tz: str | None) -> list[Recipient]:
    """Each active account that wants `kind` and has a real address, in its own
    zone (UTC when it has none, and never UTC beside a zone it chose), then the
    "Also send to" addresses in `extra_tz` with UTC beside. An address is sent
    to once."""
    from steward.core.formatting import zone
    out: dict[str, Recipient] = {}
    for u in sorted(users, key=lambda u: u.username.lower()):
        if u.is_active and wants(u, kind, settings) and is_real_address(u.email):
            out.setdefault(u.email.strip().lower(), Recipient(u.email.strip(), zone(u.time_zone), False, u.username))
    for address in extra or []:
        if is_real_address(address):
            out.setdefault(address.strip().lower(), Recipient(address.strip(), zone(extra_tz), True))
    return list(out.values())


# An account's choice for each kind of email, as a form sends it.
CHOICES = {"": None, "on": True, "off": False}


def role_defaults(user, settings: dict) -> dict[str, bool]:
    """{kind: whether this account's role gets it by default}."""
    return {kind: bool(settings.get(f"notify.{kind}.{user.role.value}", True)) for kind in KINDS}


def apply_email_form(user, form) -> tuple[dict, str | None]:
    """Apply an email form to `user`: "email" and "notify_alerts" /
    "notify_report" ("", "on" or "off"), each only when the form has it. Returns
    ({field: [before, after]} for what changed, None), or ({}, why) having
    changed nothing."""
    wanted = {}
    if "email" in form:
        address = (form.get("email") or "").strip()
        if not is_real_address(address) or len(address) > 255:
            return {}, "That isn't an email address mail can reach."
        wanted["email"] = address
    for kind in KINDS:
        if f"notify_{kind}" not in form:
            continue
        value = form.get(f"notify_{kind}")
        if value not in CHOICES:
            return {}, "Pick On, Off or the role's default."
        wanted[f"notify_{kind}"] = CHOICES[value]
    changes = {k: [getattr(user, k), v] for k, v in wanted.items() if getattr(user, k) != v}
    for k, (_, v) in changes.items():
        setattr(user, k, v)
    return changes, None


async def recipients(db, kind: str, smtp_cfg: dict) -> list[Recipient]:
    """pick_recipients from the accounts and settings as they are now."""
    from sqlalchemy import select
    from steward.core.settings import get_all_settings
    from steward.models.users import User
    users = (await db.execute(select(User).where(User.is_active.is_(True)))).scalars().all()
    return pick_recipients(users, kind, await get_all_settings(db),
                           smtp_cfg.get("recipients") or [], smtp_cfg.get("tz"))


async def audience(db) -> dict:
    """Who the emails reach, for the Notifications page and the Alerts page:
    {"alerts": [User], "report": [User]} the active accounts that get each,
    and "unreachable": the active accounts mail can't reach because their
    address isn't real."""
    from sqlalchemy import select
    from steward.core.settings import get_all_settings
    from steward.models.users import User
    users = sorted((await db.execute(select(User).where(User.is_active.is_(True)))).scalars().all(),
                   key=lambda u: u.username.lower())
    settings = await get_all_settings(db)
    out = {kind: [u for u in users if wants(u, kind, settings) and is_real_address(u.email)]
           for kind in KINDS}
    out["unreachable"] = [u for u in users if not is_real_address(u.email)]
    return out


def _render(name: str, **ctx) -> tuple[str, str]:
    ctx = {"c": palette(), **ctx}
    return (_env.get_template(f"{name}.txt").render(ctx),
            _env.get_template(f"{name}.html").render(ctx))


def sender(cfg: dict) -> str:
    return cfg.get("from") or cfg.get("username") or "steward@localhost"


def message(cfg: dict, to: str, subject: str, text: str, html: str | None = None) -> EmailMessage:
    msg = EmailMessage()
    # A header may not hold a line break; a rule name can (MCP takes any string).
    msg["Subject"] = " ".join(subject.split())
    msg["From"] = sender(cfg)
    msg["To"] = to
    msg.set_content(text)
    if html:
        msg.add_alternative(html, subtype="html")
    return msg


def moment(at: datetime, tz: tzinfo) -> str:
    """'Thu 8 Oct 2026 13:58 EDT': a time in an email, always with its date
    and zone, since a mail is read long after it is sent."""
    local = at.astimezone(tz)
    return f"{local:%a} {local.day} {local:%b %Y %H:%M} {local.tzname()}"


def _lower_first(label: str) -> str:
    """A label mid-sentence: "Response time" -> "response time", "CPU" stays."""
    return label if label[:2].isupper() else label[:1].lower() + label[1:]


def alert_facts(alert: dict, tz: tzinfo, show_utc: bool = True) -> dict:
    """What an alert email says, in words, from the dict alerts dispatch, its
    time in `tz` (and in UTC beside it when `show_utc` and they differ)."""
    from steward.alerts.service import _BINARY, condition_text
    from steward.core.formatting import metric_label, metric_value, status_info
    tone, label = status_info(alert["state"])
    metric, resource, value = alert["metric"], alert["resource"], alert.get("value")
    reading = metric_value(metric, value) if value is not None else None
    what = _lower_first(metric_label(metric))
    state = alert["state"].lower()
    if state == "no_data":
        headline = f"Steward has had no {what} reading from {resource}."
    elif state == "data_resumed":
        headline = f"{resource} is reporting {what} again."
    elif metric in _BINARY:
        headline = f"{resource} is {reading or 'back to normal'}."
    elif state == "resolved":
        headline = f"{metric_label(metric)} on {resource} is back within its limit" + (
            f", at {reading}." if reading else ".")
    elif reading is None:
        headline = f"{metric_label(metric)} on {resource} has no reading yet."
    else:
        headline = f"{metric_label(metric)} on {resource} is {reading}."
    at = datetime.fromisoformat(alert["timestamp"])
    local, utc = moment(at, tz), moment(at, timezone.utc)
    return {
        "state": label, "tone": tone, "headline": headline,
        "rule": alert["rule_name"], "resource": resource,
        "reading": reading or "no data",
        "condition": condition_text(metric, alert["operator"], alert["threshold"]),
        "when": local, "when_utc": utc if show_utc and utc != local else "",
        "note": alert.get("note") or "",
        "subject": f"[Steward] {label}: {alert['rule_name']} — {resource}"
                   + (f" {reading}" if reading and state in ("firing", "test") else ""),
    }


def alert_email(cfg: dict, alert: dict, to: Recipient) -> tuple[str, str, str]:
    """(subject, text, html) for an alert, written for one recipient in their
    zone, linking to Steward when its address is set."""
    facts = alert_facts(alert, to.tz, to.show_utc)
    base = (cfg.get("base_url") or "").rstrip("/")
    text, html = _render("alert", a=facts, link=f"{base}/alerts/" if base else "")
    return facts["subject"], text, html


def burst_email(cfg: dict, alerts: list[dict], to: Recipient) -> tuple[str, str, str]:
    """(subject, text, html) for several alerts that arrived together
    (core.outbox.alert_words, #5465): how many and what they came to in the subject,
    then each alert's headline, rule and time, oldest first, in the
    recipient's zone."""
    from collections import Counter
    facts = [alert_facts(a, to.tz, to.show_utc) for a in alerts]
    counts = Counter(f["state"] for f in facts)
    said = ", ".join(f"{n} {state.lower()}" for state, n in counts.most_common())
    base = (cfg.get("base_url") or "").rstrip("/")
    text, html = _render("burst", alerts=facts, said=said,
                         link=f"{base}/alerts/" if base else "")
    return f"[Steward] {len(facts)} alerts: {said}", text, html


def report_email(cfg: dict, data: dict, to: Recipient) -> tuple[str, str, str]:
    """(subject, text, html) for the weekly report built by
    reports.build_report_data, its dates in the recipient's zone."""
    from steward.core.formatting import status_info
    tz = to.tz
    end = data["generated_at"].astimezone(tz)
    start = end - timedelta(days=7)
    period = (f"{start.day} {start:%b} – {end.day} {end:%b %Y}" if start.year == end.year
              else f"{start.day} {start:%b %Y} – {end.day} {end:%b %Y}")
    alerts = [{**a, "tone": status_info(a["state"])[0], "state": status_info(a["state"])[1]}
              for a in data["active_alerts"]]
    base = (cfg.get("base_url") or "").rstrip("/")
    text, html = _render("report", d=data, alerts=alerts, period=period,
                         sent=moment(data["generated_at"], tz), base=base)
    return f"[Steward] Weekly report, {period}", text, html


ROLE_WORDS = {
    "viewer": "a viewer: you can read every page",
    "operator": "an operator: you can also run playbooks and change monitors, hosts and alerts",
    "admin": "an admin: you can also change Settings and users",
}


def reset_email(username: str, expires_at: datetime, url: str, tz: tzinfo) -> tuple[str, str, str]:
    """(subject, text, html) for a password-reset link (milestone 497): whose
    account, when the link stops working (in the account's own zone) and the
    link, with what to do if they didn't ask."""
    facts = {"username": username, "link": url, "expires": moment(expires_at, tz)}
    text, html = _render("password_reset", r=facts)
    return "[Steward] Reset your password", text, html


def invite_email(inviter: str, role: str, expires_at: datetime, url: str,
                 tz: tzinfo) -> tuple[str, str, str]:
    """(subject, text, html) for an invite (milestone 496): who sent it, the
    role it gives, when the link stops working (in the sender's zone, with
    UTC beside it, since the invitee's zone isn't known) and the link."""
    local, utc = moment(expires_at, tz), moment(expires_at, timezone.utc)
    facts = {"inviter": inviter, "role": ROLE_WORDS.get(role, role), "link": url,
             "expires": local, "expires_utc": utc if utc != local else ""}
    text, html = _render("invite", i=facts)
    return f"[Steward] {inviter} invited you to Steward", text, html