notify-email/scripts/send.py

#!/usr/bin/env python3
"""Build and send a fritzlab-themed CI notification email.

Reads inputs from environment variables (set by action.yaml). Delivers a
multipart/alternative message (plain text + inline-styled HTML) over plain SMTP
to mail.fritzlab.net, which relays via trusted-CIDR rules (see mail.md).
"""
from __future__ import annotations

import os
import re
import smtplib
import sys
import urllib.error
import urllib.request
from email.message import EmailMessage
from html import escape
from typing import Iterable

# fritzlab tokens (mirror websites/fritzlab.net/theme.css). Inlined here because
# email clients strip <style>, drop @import, and cannot fetch web fonts.
CANVAS = "#0a0a0a"
SURFACE = "#111113"
BORDER = "#1f1f23"
BORDER_STRONG = "#2a2a2e"
TEXT = "#ededed"
MUTED = "#888888"
MUTED_STRONG = "#a8a8a8"

ACCENT_VIOLET = "#7c5cff"
ERROR_FG = "#fca5a5"
ERROR_BG = "#2a0f0f"
ERROR_BORDER = "#4a1818"
OK_FG = "#86efac"  # not in theme.css; added for HTML email status pills
OK_BORDER = "#1f3a2a"
OK_BG = "#0f1f15"

FONT_SANS = (
    "'Geist', -apple-system, BlinkMacSystemFont, 'Inter', 'Segoe UI', "
    "Helvetica, Arial, sans-serif"
)
FONT_MONO = "'Geist Mono', ui-monospace, 'JetBrains Mono', Menlo, monospace"


def status_palette(status: str) -> dict[str, str]:
    """Return colour + label tokens for a given status string."""
    s = (status or "info").strip().lower()
    if s == "failure":
        return {
            "key": "failure",
            "label": "FAILED",
            "fg": ERROR_FG,
            "bg": ERROR_BG,
            "border": ERROR_BORDER,
            "subject_prefix": "[FAILED]",
        }
    if s == "success":
        return {
            "key": "success",
            "label": "OK",
            "fg": OK_FG,
            "bg": OK_BG,
            "border": OK_BORDER,
            "subject_prefix": "[OK]",
        }
    return {
        "key": "info",
        "label": "INFO",
        "fg": ACCENT_VIOLET,
        "bg": "rgba(124, 92, 255, 0.08)",
        "border": "#2a2050",
        "subject_prefix": "[INFO]",
    }


def required(name: str) -> str:
    v = os.environ.get(name, "").strip()
    if not v:
        print(f"FATAL: required env var {name} is empty", file=sys.stderr)
        sys.exit(1)
    return v


# Strip the leading `2026-05-28T17:09:35.337087Z ` timestamp that act_runner
# prepends to every job-log line. Pattern: ISO-8601 + variable-precision micros
# + Z + single space.
_TS_PREFIX = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z\s")


def _api_get(url: str, token: str) -> bytes:
    req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.read()


def fetch_failing_log(
    *,
    server_url: str,
    repo: str,
    run_id: str,
    token: str,
    log_lines: int,
) -> str:
    """Return a digestible tail of the current run's failing-job log.

    Resolution: enumerate jobs in this run, pick the one whose conclusion is
    'failure' (or — for an in-flight notify-on-failure step — whose status is
    'in_progress' with at least one step in failure). Fetch that job's full
    text log. If any line carries an `::error::` annotation, return those
    lines plus a small context window above each. Otherwise, return the last
    `log_lines` lines. Timestamp prefixes are stripped.
    """
    import json as _json

    base = server_url.rstrip("/")
    jobs_url = f"{base}/api/v1/repos/{repo}/actions/runs/{run_id}/jobs"
    try:
        jobs = _json.loads(_api_get(jobs_url, token)).get("jobs", [])
    except (urllib.error.HTTPError, urllib.error.URLError, ValueError) as e:
        print(f"notify-email: auto-log: jobs lookup failed: {e}", file=sys.stderr)
        return ""

    failed = None
    for j in jobs:
        if j.get("conclusion") == "failure":
            failed = j
            break
        # Notify step itself is still in_progress at this point — pick the job
        # if any of its prior steps failed.
        if j.get("status") == "in_progress" and any(
            s.get("status") == "failure" or s.get("conclusion") == "failure"
            for s in j.get("steps", [])
        ):
            failed = j
            break
    if not failed:
        # Heuristic last resort: most recent job in the run.
        failed = jobs[0] if jobs else None
    if not failed:
        return ""

    job_id = failed.get("id")
    log_url = f"{base}/api/v1/repos/{repo}/actions/jobs/{job_id}/logs"
    try:
        raw = _api_get(log_url, token).decode("utf-8", errors="replace")
    except (urllib.error.HTTPError, urllib.error.URLError) as e:
        print(f"notify-email: auto-log: log fetch failed: {e}", file=sys.stderr)
        return ""

    cleaned = [_TS_PREFIX.sub("", ln) for ln in raw.splitlines()]
    error_idxs = [i for i, ln in enumerate(cleaned) if "::error::" in ln]
    if error_idxs:
        # Window: 12 lines of context above each ::error:: annotation,
        # plus the annotation itself; merge overlapping windows.
        ranges: list[tuple[int, int]] = []
        for i in error_idxs:
            lo = max(0, i - 12)
            hi = i + 1
            if ranges and lo <= ranges[-1][1] + 1:
                ranges[-1] = (ranges[-1][0], max(ranges[-1][1], hi))
            else:
                ranges.append((lo, hi))
        chunks: list[str] = []
        for lo, hi in ranges:
            chunks.append("\n".join(cleaned[lo:hi]))
        return "\n…\n".join(chunks).strip()

    return "\n".join(cleaned[-log_lines:]).strip()


def fact_rows(items: Iterable[tuple[str, str, str]]) -> tuple[str, str]:
    """Build the HTML <table> rows and plain-text lines for the fact list.

    Each item is (label, text_value, html_value). Items whose text_value is
    empty are skipped — the HTML/text outputs stay in lockstep.
    """
    html_rows: list[str] = []
    text_lines: list[str] = []
    for label, text_value, html_value in items:
        if not text_value:
            continue
        html_rows.append(
            f'<tr>'
            f'<td style="color:{MUTED};font-family:{FONT_MONO};font-size:12px;'
            f'padding:4px 24px 4px 0;vertical-align:top;white-space:nowrap;'
            f'text-transform:uppercase;letter-spacing:0.12em;">{escape(label)}</td>'
            f'<td style="color:{TEXT};font-family:{FONT_SANS};font-size:14px;'
            f'padding:4px 0;line-height:1.5;">{html_value}</td>'
            f'</tr>'
        )
        text_lines.append(f"{label:<11} {text_value}")
    return "\n".join(html_rows), "\n".join(text_lines)


def render(
    *,
    palette: dict[str, str],
    summary: str,
    details: str,
    facts: list[tuple[str, str]],
    run_url: str,
) -> tuple[str, str]:
    """Return (html, text) bodies."""
    rows_html, rows_text = fact_rows(facts)

    pill = (
        f'<span style="display:inline-block;font-family:{FONT_MONO};'
        f'font-size:11px;font-weight:600;letter-spacing:0.14em;'
        f'padding:4px 10px;border-radius:6px;'
        f'color:{palette["fg"]};background:{palette["bg"]};'
        f'border:1px solid {palette["border"]};text-transform:uppercase;">'
        f'{palette["label"]}</span>'
    )

    details_block_html = ""
    if details.strip():
        details_block_html = f"""
        <tr><td style="padding:8px 32px 0 32px;">
          <div style="font-family:{FONT_MONO};font-size:11px;color:{MUTED};
                      text-transform:uppercase;letter-spacing:0.14em;
                      margin:18px 0 8px 0;">details</div>
          <pre style="margin:0;padding:14px 16px;background:{CANVAS};
                      border:1px solid {BORDER};border-radius:8px;
                      color:{MUTED_STRONG};font-family:{FONT_MONO};
                      font-size:12px;line-height:1.55;white-space:pre-wrap;
                      word-break:break-word;overflow-x:auto;">{escape(details)}</pre>
        </td></tr>
        """

    html = f"""<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="color-scheme" content="dark">
  <meta name="supported-color-schemes" content="dark">
  <title>{escape(summary)}</title>
</head>
<body style="margin:0;padding:0;background:{CANVAS};color:{TEXT};
             font-family:{FONT_SANS};-webkit-font-smoothing:antialiased;">
  <table role="presentation" width="100%" cellpadding="0" cellspacing="0"
         border="0" style="background:{CANVAS};padding:32px 16px;">
    <tr><td align="center">
      <table role="presentation" width="600" cellpadding="0" cellspacing="0"
             border="0" style="max-width:600px;width:100%;background:{SURFACE};
                               border:1px solid {BORDER};border-radius:12px;
                               overflow:hidden;">
        <tr><td style="padding:28px 32px 8px 32px;">
          {pill}
          <h1 style="margin:14px 0 0 0;font-family:{FONT_SANS};font-size:20px;
                     font-weight:600;letter-spacing:-0.015em;color:{TEXT};
                     line-height:1.35;">{escape(summary)}</h1>
        </td></tr>
        <tr><td style="padding:20px 32px 8px 32px;">
          <hr style="border:0;border-top:1px solid {BORDER};margin:0 0 16px 0;">
          <table role="presentation" cellpadding="0" cellspacing="0" border="0"
                 style="width:100%;">
            {rows_html}
          </table>
        </td></tr>
        {details_block_html}
        <tr><td style="padding:24px 32px 28px 32px;">
          <a href="{escape(run_url)}"
             style="display:inline-block;font-family:{FONT_SANS};font-size:14px;
                    font-weight:500;color:{TEXT};text-decoration:none;
                    padding:11px 22px;border:1px solid {BORDER_STRONG};
                    border-radius:8px;background:transparent;">
            view run <span style="color:{palette['fg']};">&rarr;</span>
          </a>
        </td></tr>
        <tr><td style="padding:0 32px 24px 32px;">
          <div style="font-family:{FONT_MONO};font-size:11px;color:{MUTED};
                      border-top:1px solid {BORDER};padding-top:14px;
                      letter-spacing:0.02em;">
            sent by <span style="color:{MUTED_STRONG};">action/notify-email</span>
            &middot; mail.fritzlab.net
          </div>
        </td></tr>
      </table>
    </td></tr>
  </table>
</body>
</html>
"""

    text = (
        f"[{palette['label']}] {summary}\n"
        f"{'-' * 64}\n"
        f"{rows_text}\n"
    )
    if details.strip():
        text += f"\nDetails:\n{details.rstrip()}\n"
    text += f"\nRun: {run_url}\n"
    return html, text


def main() -> int:
    summary = required("NOTIFY_SUMMARY")
    to_addr = required("NOTIFY_TO")
    from_addr = required("NOTIFY_FROM")
    smtp_host = required("NOTIFY_SMTP_HOST")
    smtp_port = int(required("NOTIFY_SMTP_PORT"))
    subject_in = os.environ.get("NOTIFY_SUBJECT", "").strip()
    details = os.environ.get("NOTIFY_DETAILS", "")

    palette = status_palette(os.environ.get("NOTIFY_STATUS", "info"))

    server_url = os.environ.get("GH_SERVER_URL", "").rstrip("/")
    repo = os.environ.get("GH_REPOSITORY", "")
    sha = os.environ.get("GH_SHA", "")
    short_sha = sha[:8] if sha else ""
    run_id = os.environ.get("GH_RUN_ID", "")
    run_url = f"{server_url}/{repo}/actions/runs/{run_id}" if server_url and repo and run_id else ""

    auto_log = os.environ.get("NOTIFY_AUTO_LOG", "").strip().lower() in {"1", "true", "yes"}
    if auto_log and not details.strip() and server_url and repo and run_id:
        token = os.environ.get("NOTIFY_TOKEN", "").strip()
        if token:
            try:
                lines = int(os.environ.get("NOTIFY_LOG_LINES", "40") or "40")
            except ValueError:
                lines = 40
            fetched = fetch_failing_log(
                server_url=server_url,
                repo=repo,
                run_id=run_id,
                token=token,
                log_lines=lines,
            )
            if fetched:
                details = fetched

    commit_html = ""
    if sha:
        if server_url and repo:
            commit_html = (
                f'<a href="{escape(server_url)}/{escape(repo)}/commit/{escape(sha)}" '
                f'style="color:{TEXT};text-decoration:none;'
                f'border-bottom:1px dotted {MUTED};font-family:{FONT_MONO};'
                f'font-size:12px;">{escape(short_sha)}</a>'
            )
        else:
            commit_html = f'<span style="font-family:{FONT_MONO};font-size:12px;">{escape(short_sha)}</span>'

    repo_html = ""
    if repo:
        if server_url:
            repo_html = (
                f'<a href="{escape(server_url)}/{escape(repo)}" '
                f'style="color:{TEXT};text-decoration:none;'
                f'border-bottom:1px dotted {MUTED};">{escape(repo)}</a>'
            )
        else:
            repo_html = escape(repo)

    ref_name = os.environ.get("GH_REF_NAME", "")
    workflow = os.environ.get("GH_WORKFLOW", "")
    job = os.environ.get("GH_JOB", "")
    run_number = os.environ.get("GH_RUN_NUMBER", "")
    run_attempt = os.environ.get("GH_RUN_ATTEMPT", "")
    actor = os.environ.get("GH_ACTOR", "")
    event_name = os.environ.get("GH_EVENT_NAME", "")

    # Workflow line: include job only if it differs from the workflow name
    # (composite-action runs often have workflow == job, e.g. "deploy / deploy").
    if workflow and job and job != workflow:
        workflow_text = f"{workflow} / {job}"
        workflow_html = f"{escape(workflow)} <span style=\"color:{MUTED};\">/</span> {escape(job)}"
    elif workflow:
        workflow_text = workflow
        workflow_html = escape(workflow)
    else:
        workflow_text = workflow_html = ""

    run_text = ""
    run_html = ""
    if run_number:
        run_text = f"#{run_number}"
        run_html = f"#{escape(run_number)}"
        if run_attempt and run_attempt != "1":
            run_text += f" (attempt {run_attempt})"
            run_html += f" (attempt {escape(run_attempt)})"

    trigger_text = ""
    trigger_html = ""
    if event_name:
        trigger_text = event_name + (f" by {actor}" if actor else "")
        trigger_html = escape(event_name) + (f" by {escape(actor)}" if actor else "")

    facts: list[tuple[str, str, str]] = [
        ("Repository", repo, repo_html),
        ("Branch", ref_name, escape(ref_name) if ref_name else ""),
        ("Commit", short_sha, commit_html),
        ("Workflow", workflow_text, workflow_html),
        ("Run", run_text, run_html),
        ("Trigger", trigger_text, trigger_html),
    ]

    html, text = render(
        palette=palette,
        summary=summary,
        details=details,
        facts=facts,
        run_url=run_url,
    )

    msg = EmailMessage()
    subject_body = subject_in or summary
    msg["Subject"] = f"{palette['subject_prefix']} {subject_body}"
    msg["From"] = from_addr
    msg["To"] = to_addr
    msg["X-Fritzlab-Status"] = palette["key"]
    msg["X-Fritzlab-Source"] = "action/notify-email"
    if repo:
        msg["X-Fritzlab-Repo"] = repo
    msg.set_content(text)
    msg.add_alternative(html, subtype="html")

    # Force a FQDN-shaped EHLO. Runner pods don't set setHostnameAsFQDN, so
    # socket.getfqdn() returns the single-label pod name and Stalwart's
    # reject-non-fqdn rule trips before MAIL FROM. See k8s-manager skill
    # mail.md <smtp-from-k8s-pod>.
    print(f"notify-email: sending {palette['key']} to {to_addr} via {smtp_host}:{smtp_port}", file=sys.stderr)
    with smtplib.SMTP(smtp_host, smtp_port, local_hostname="ci.fritzlab.net", timeout=15) as s:
        s.send_message(msg)
    print("notify-email: sent", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())