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.
Duplicate alerts (identical reasons since last send) are suppressed.
"""
import hashlib
import json
import logging
import os
import smtplib
from email.message import EmailMessage
from pathlib import Path

import requests

from .plain import ampel
from .report import render_markdown

logger = logging.getLogger(__name__)

_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2}
_ALERT_STATE_FILE = "last_alert.json"


def send_alert(report: dict, cfg: dict) -> None:
    """Send alert if the report's level is at or above min_level.

    Duplicate alerts (identical assessment reasons since last send) are suppressed.
    The dedup state is persisted in data_dir/last_alert.json per site.
    """
    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

    # --- Deduplizierung: nur senden wenn sich die Alarm-Gründe geändert haben ---
    alert_key = _compute_alert_key(report)
    data_dir = cfg.get("data_dir")
    if data_dir:
        last = _load_last_alert(Path(data_dir))
        if last and last.get("key") == alert_key:
            logger.info("Alert suppressed (duplicate, reasons unchanged since %s)",
                        last.get("sent_at", "unknown"))
            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)

    # Zustand speichern — für nächste Dedup-Runde
    if data_dir:
        _save_last_alert(Path(data_dir), alert_key)


# ---------------------------------------------------------------------------
# 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)


# ---------------------------------------------------------------------------
# Deduplication state (last_alert.json in data_dir)
# ---------------------------------------------------------------------------

def _compute_alert_key(report: dict) -> str:
    """Stable hash of the assessment reasons + level for deduplication."""
    payload = {
        "level": report.get("assessment", {}).get("level", "green"),
        "score": report.get("assessment", {}).get("score", 0),
        "reasons": sorted(report.get("assessment", {}).get("reasons", [])),
    }
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()


def _load_last_alert(data_dir: Path) -> dict:
    """Load previous alert state. Returns {} on missing/corrupt file."""
    path = data_dir / _ALERT_STATE_FILE
    try:
        return json.loads(path.read_text())
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        return {}


def _save_last_alert(data_dir: Path, alert_key: str) -> None:
    """Persist current alert state for next dedup round."""
    from datetime import datetime, timezone
    path = data_dir / _ALERT_STATE_FILE
    state = {
        "key": alert_key,
        "sent_at": datetime.now(timezone.utc).isoformat(),
    }
    try:
        path.write_text(json.dumps(state))
    except OSError as exc:
        logger.debug("Alert state write failed: %s", exc)