integrity_scanner_fuer_statische_Webseiten/scanner/alerter.py

"""
Alert dispatch: e-mail (smtplib) and HTTP webhook.

Alerts are only sent when the assessment level meets or exceeds min_level.
"""
import logging
import os
import smtplib
from email.message import EmailMessage

import requests

from .plain import ampel
from .report import render_markdown

logger = logging.getLogger(__name__)

_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2}


def send_alert(report: dict, cfg: dict) -> None:
    """Send alert if the report's level is at or above min_level."""
    alerting = cfg.get("alerting", {})
    level = report.get("assessment", {}).get("level", "green")
    min_level = alerting.get("min_level", "yellow")

    if _LEVEL_ORDER.get(level, 0) < _LEVEL_ORDER.get(min_level, 1):
        logger.debug("Alert suppressed: level=%s < min_level=%s", level, min_level)
        return

    subject = _make_subject(report)
    body = _format_body(report)

    email_cfg = alerting.get("email", {})
    if email_cfg.get("enabled"):
        _send_email(subject, body, email_cfg)

    webhook_cfg = alerting.get("webhook", {})
    if webhook_cfg.get("enabled") and webhook_cfg.get("url"):
        _send_webhook(webhook_cfg["url"], report, subject)


# ---------------------------------------------------------------------------
# Formatting
# ---------------------------------------------------------------------------

def _make_subject(report: dict) -> str:
    level = report.get("assessment", {}).get("level", "green")
    target = report.get("target", "?")
    symbol, _ = ampel(level)
    if level == "red":
        return f"{symbol} Ihre Website {target} braucht Aufmerksamkeit"
    if level == "yellow":
        return f"{symbol} Bitte einmal auf Ihrer Website {target} nachschauen"
    return f"{symbol} Ihre Website {target}: alles in Ordnung"


def _format_body(report: dict) -> str:
    """E-Mail-Body = derselbe Renderer wie der Dateibericht (eine Quelle der
    Wahrheit). So enthalten Report und Mail garantiert denselben Inhalt."""
    return render_markdown(report)


# ---------------------------------------------------------------------------
# Transport
# ---------------------------------------------------------------------------

def _send_email(subject: str, body: str, cfg: dict) -> None:
    password = os.environ.get(cfg.get("smtp_password_env", "SCANNER_SMTP_PASSWORD"), "")
    recipients = cfg.get("to", [])
    if not recipients:
        logger.warning("E-Mail alert enabled but no recipients configured.")
        return

    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = cfg.get("from", "scanner@localhost")
    msg["To"] = ", ".join(recipients)
    msg.set_content(body)

    host = cfg.get("smtp_host", "localhost")
    port = cfg.get("smtp_port", 587)
    use_tls = cfg.get("smtp_tls", True)

    try:
        with smtplib.SMTP(host, port) as server:
            if use_tls:
                server.starttls()
            user = cfg.get("smtp_user", "")
            if user and password:
                server.login(user, password)
            server.send_message(msg)
        logger.info("Alert e-mail sent to %s", recipients)
    except Exception as exc:
        logger.error("E-mail alert failed: %s", exc)


def _send_webhook(url: str, report: dict, subject: str) -> None:
    payload = {
        "text": subject,
        "target": report.get("target"),
        "level": report.get("assessment", {}).get("level"),
        "score": report.get("assessment", {}).get("score"),
        "reasons": report.get("assessment", {}).get("reasons", []),
        "new_external_domains": report.get("diff", {}).get("new_external_domains", []),
        "new_internal_urls": report.get("diff", {}).get("new_internal_urls", []),
        "missing_internal_urls": report.get("diff", {}).get("missing_internal_urls", []),
        "changed_pages": report.get("diff", {}).get("changed_pages", 0),
    }
    try:
        resp = requests.post(url, json=payload, timeout=10)
        logger.info("Webhook delivered: HTTP %s", resp.status_code)
    except Exception as exc:
        logger.error("Webhook alert failed: %s", exc)