"""alerts.py - Alert rule engine for Squid Web Manager (P1-2).

Provides:
- AlertRule dataclass - configurable metric threshold rule
- evaluate_rules(entries, rules, now=None) - run all enabled rules against
  the current access.log stats and return a list of triggered alerts.
- save_rules / load_rules - persist rule list as JSON
- default_rules() - opinionated starting set that covers the common SRE cases

All metric computation uses only Python stdlib + existing log_parser statistics
- no new dependencies. Audit writes use a lazy import of `app` so this module
can be imported without forcing a circular dependency on app.py at startup.
"""
from __future__ import annotations

import json
import os
import shutil
import time
from dataclasses import asdict, dataclass, field
from typing import Any, Iterable

# --- Supported metrics ----------------------------------------------------
# Each entry documents the operator semantics. Values are floats unless noted.
METRICS: dict[str, str] = {
    "5xx_rate": "HTTP 5xx 错误占比 (0.0 - 1.0)",
    "4xx_rate": "HTTP 4xx 错误占比 (0.0 - 1.0)",
    "hit_ratio": "缓存命中率 (0.0 - 1.0)",
    "denied_rate": "ACL 拒绝请求占比 (0.0 - 1.0)",
    "disk_usage_pct": "缓存目录磁盘使用率 (0.0 - 100.0)",
    "client_bytes_per_min": "Top 客户端每分钟流量 (字节/分钟)",
}

OPERATORS: dict[str, str] = {
    "gt": ">",
    "lt": "<",
    "gte": ">=",
    "lte": "<=",
}

SEVERITIES = ("info", "warning", "critical")


# --- Rule dataclass -------------------------------------------------------

@dataclass
class AlertRule:
    """One alert rule. Persisted as JSON via save_rules / load_rules."""

    name: str
    metric: str
    operator: str
    threshold: float
    window_minutes: int = 5
    enabled: bool = True
    cooldown_minutes: int = 30
    last_triggered: float = 0.0
    severity: str = "warning"
    notify_webhook: str = ""

    def to_dict(self) -> dict:
        d = asdict(self)
        # JSON doesn't love bare floats like 0.0 - keep it simple.
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "AlertRule":
        """Build from a dict (e.g. one loaded from JSON). Missing keys fall
        back to dataclass defaults so old config files keep working."""
        if not isinstance(d, dict):
            raise ValueError("AlertRule.from_dict expects a dict")
        # Only forward the keys we know about; ignore extras.
        allowed = {f for f in cls.__dataclass_fields__.keys()}  # type: ignore[attr-defined]
        kwargs = {k: d[k] for k in d.keys() & allowed}
        # Coerce numerics - if the file is hand-edited we don't want to crash.
        for key in ("threshold", "last_triggered"):
            if key in kwargs:
                try:
                    kwargs[key] = float(kwargs[key])
                except (TypeError, ValueError):
                    kwargs[key] = 0.0
        for key in ("window_minutes", "cooldown_minutes"):
            if key in kwargs:
                try:
                    kwargs[key] = int(kwargs[key])
                except (TypeError, ValueError):
                    kwargs[key] = 5 if key == "window_minutes" else 30
        kwargs["enabled"] = bool(kwargs.get("enabled", True))
        # Severity / operator / metric - fall back if hand-edited to junk.
        sev = kwargs.get("severity", "warning")
        if sev not in SEVERITIES:
            sev = "warning"
        kwargs["severity"] = sev
        op = kwargs.get("operator", "gt")
        if op not in OPERATORS:
            op = "gt"
        kwargs["operator"] = op
        return cls(**kwargs)


# --- Metric evaluation ---------------------------------------------------

def _apply_operator(value: float, op: str, threshold: float) -> bool:
    if op == "gt":
        return value > threshold
    if op == "lt":
        return value < threshold
    if op == "gte":
        return value >= threshold
    if op == "lte":
        return value <= threshold
    return False


def _compute_metric(
    metric: str,
    stats: dict,
    cache_dir: str | None,
    window_minutes: int,
) -> float:
    """Compute the current value of a metric from pre-aggregated stats.

    For most metrics this is just `stats[metric]`. For client_bytes_per_min we
    fold in the time window (which is configurable on the rule, so we can't
    pre-bake it into log_parser). disk_usage_pct needs shutil.disk_usage.
    """
    if metric == "5xx_rate":
        return float(stats.get("errors_5xx_rate") or 0.0)
    if metric == "4xx_rate":
        return float(stats.get("errors_4xx_rate") or 0.0)
    if metric == "hit_ratio":
        return float(stats.get("hit_ratio") or 0.0)
    if metric == "denied_rate":
        return float(stats.get("denied_rate") or 0.0)
    if metric == "disk_usage_pct":
        target = cache_dir or "/var/spool/squid"
        try:
            total, used, free = shutil.disk_usage(target)
        except (FileNotFoundError, PermissionError, OSError):
            return 0.0
        if not total:
            return 0.0
        return round(used * 100.0 / total, 2)
    if metric == "client_bytes_per_min":
        tops = stats.get("top_clients") or []
        if not tops:
            return 0.0
        max_bytes = max((c.get("bytes") or 0) for c in tops)
        # Approximate bytes-per-minute for the top client over the rule window.
        # If the time span is < window_minutes we still divide by window so
        # the operator gets a comparable value (long window = low rate).
        w = max(1, int(window_minutes))
        return float(max_bytes) * 60.0 / w
    return 0.0


def _humanise_metric(metric: str, value: float) -> str:
    if metric.endswith("_rate"):
        return f"{value * 100:.2f}%"
    if metric == "hit_ratio":
        return f"{value * 100:.1f}%"
    if metric == "disk_usage_pct":
        return f"{value:.1f}%"
    if metric == "client_bytes_per_min":
        # Display in KB/MB/GB - keep parity with log_parser.format_bytes shape.
        n = float(value)
        for unit in ("B/s", "KB/s", "MB/s", "GB/s"):
            if abs(n) < 1024.0:
                if unit == "B/s":
                    return f"{n:.0f} {unit}"
                return f"{n:.1f} {unit}"
            n /= 1024.0
        return f"{n:.2f} TB/s"
    return f"{value:.4f}"


def _format_message(rule: AlertRule, value: float, window_minutes: int) -> str:
    op = OPERATORS.get(rule.operator, rule.operator)
    human = _humanise_metric(rule.metric, value)
    metric_desc = METRICS.get(rule.metric, rule.metric)
    return (
        f"{rule.name}: {metric_desc} 当前值 {human} {op} 阈值 "
        f"{_humanise_metric(rule.metric, rule.threshold)} "
        f"(窗口 {window_minutes}m, 严重度 {rule.severity})"
    )


# --- Audit (lazy import to avoid circular reference) ---------------------

def _audit_trigger(rule: AlertRule, value: float, message: str):
    """Try to write a row into the audit_log. Silently swallow on failure
    so a missing app context never breaks evaluation."""
    try:
        from app import audit, db  # type: ignore
        audit(
            "alert_triggered",
            f"name={rule.name} metric={rule.metric} "
            f"value={value:.6f} threshold={rule.threshold} "
            f"severity={rule.severity} msg={message[:300]}",
        )
        db.session.commit()
    except Exception:
        # Never let an audit failure mask a real alert.
        try:
            from app import db  # type: ignore
            db.session.rollback()
        except Exception:
            pass


def _post_webhook(rule: AlertRule, payload: dict):
    """Fire-and-forget webhook POST. We import urllib lazily; if the URL is
    bad or the import fails we just log silently - alerts mustn't crash
    the calling request."""
    url = (rule.notify_webhook or "").strip()
    if not url:
        return
    try:
        import urllib.request
        import urllib.error
        body = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            url,
            data=body,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        urllib.request.urlopen(req, timeout=5).read()
    except Exception:
        # We deliberately don't surface webhook errors - delivery is
        # best-effort and the audit row is the source of truth.
        pass


# --- Main entry point ----------------------------------------------------

def evaluate_rules(
    entries: list | None,
    rules: Iterable[AlertRule],
    *,
    cache_dir: str | None = None,
    now: float | None = None,
) -> list[dict]:
    """Evaluate all enabled rules against the current access.log parsing.

    Parameters
    ----------
    entries : list of log_parser.LogEntry (may be empty / None).
    rules   : iterable of AlertRule to check.
    cache_dir : path to the Squid cache dir. Used for disk_usage_pct. If
                None, defaults to /var/spool/squid.
    now     : epoch seconds to treat as "now". Defaults to time.time().

    Returns
    -------
    list of dict:
        {rule, value, severity, message, ts, triggered}
    Each dict contains:
      - rule       : the AlertRule instance that matched
      - value      : the float value that triggered the rule
      - severity   : copy of rule.severity
      - message    : human-readable summary
      - ts         : epoch seconds when this was evaluated
      - triggered  : bool - True if outside cooldown, False if suppressed
    The list always contains one entry per *evaluated* rule; rows with
    triggered=False are useful for the UI to show "currently OK / cooldown".
    """
    ts = float(now if now is not None else time.time())
    stats = _safe_stats(entries)
    out: list[dict] = []
    rules_list = list(rules)
    for rule in rules_list:
        if not getattr(rule, "enabled", True):
            continue
        metric = getattr(rule, "metric", "")
        op = getattr(rule, "operator", "gt")
        threshold = float(getattr(rule, "threshold", 0.0) or 0.0)
        window = max(1, int(getattr(rule, "window_minutes", 5) or 5))
        try:
            value = _compute_metric(metric, stats, cache_dir, window)
        except Exception:
            value = 0.0
        triggered = _apply_operator(value, op, threshold)
        in_cooldown = False
        if triggered:
            last = float(getattr(rule, "last_triggered", 0.0) or 0.0)
            cooldown = max(0, int(getattr(rule, "cooldown_minutes", 30) or 30))
            if last and (ts - last) < cooldown * 60:
                # Suppress but still report so the UI can show "cooling down".
                triggered = False
                in_cooldown = True
            else:
                rule.last_triggered = ts
        msg = _format_message(rule, value, window)
        if triggered:
            _audit_trigger(rule, value, msg)
            _post_webhook(
                rule,
                {
                    "name": rule.name,
                    "metric": rule.metric,
                    "value": value,
                    "threshold": threshold,
                    "operator": op,
                    "severity": rule.severity,
                    "message": msg,
                    "ts": ts,
                },
            )
        out.append({
            "rule": rule,
            "value": value,
            "threshold": threshold,
            "operator": op,
            "severity": rule.severity,
            "message": msg,
            "ts": ts,
            "triggered": triggered,
            "in_cooldown": in_cooldown,
        })
    return out


def _safe_stats(entries: list | None) -> dict:
    """Run log_parser.aggregate_stats if available, otherwise return {}.
    Import log_parser lazily so unit tests / minimal contexts don't crash."""
    if not entries:
        return {}
    try:
        import log_parser  # type: ignore
        return log_parser.aggregate_stats(entries) or {}
    except Exception:
        return {}


# --- Persistence --------------------------------------------------------

def save_rules(rules: list[AlertRule], path: str) -> None:
    """Write the rule list to `path` as pretty-printed JSON."""
    payload = [r.to_dict() for r in (rules or [])]
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)


def load_rules(path: str) -> list[AlertRule]:
    """Load rules from a JSON file. Missing / corrupt files yield the
    `default_rules()` set; that way the UI always has something to show."""
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        return default_rules()
    except (json.JSONDecodeError, OSError, ValueError):
        return default_rules()
    if not isinstance(data, list):
        return default_rules()
    out: list[AlertRule] = []
    for item in data:
        try:
            out.append(AlertRule.from_dict(item))
        except Exception:
            # Skip individual bad rows instead of nuking the whole file.
            continue
    return out or default_rules()


def default_rules() -> list[AlertRule]:
    """A balanced starter set: catches the common SRE cases without being
    noisy. Operators can tune thresholds or disable rules from the UI."""
    return [
        AlertRule(
            name="5xx 错误率突增",
            metric="5xx_rate",
            operator="gt",
            threshold=0.05,           # > 5% 5xx
            window_minutes=5,
            cooldown_minutes=30,
            severity="critical",
        ),
        AlertRule(
            name="缓存命中率下跌",
            metric="hit_ratio",
            operator="lt",
            threshold=0.20,           # < 20%
            window_minutes=5,
            cooldown_minutes=60,
            severity="warning",
        ),
        AlertRule(
            name="磁盘空间不足",
            metric="disk_usage_pct",
            operator="gte",
            threshold=90.0,           # >= 90%
            window_minutes=5,
            cooldown_minutes=60,
            severity="critical",
        ),
        AlertRule(
            name="客户端流量异常",
            metric="client_bytes_per_min",
            operator="gt",
            threshold=200 * 1024 * 1024,  # > 200 MB/min
            window_minutes=5,
            cooldown_minutes=15,
            severity="warning",
        ),
        AlertRule(
            name="拒绝率过高",
            metric="denied_rate",
            operator="gt",
            threshold=0.50,           # > 50%
            window_minutes=5,
            cooldown_minutes=30,
            severity="warning",
        ),
    ]


# --- CLI smoke test ------------------------------------------------------

if __name__ == "__main__":  # pragma: no cover - manual sanity check
    import log_parser as _lp
    sample = os.path.join(os.path.dirname(__file__), "sample_access.log")
    text = open(sample, "r", encoding="utf-8").read() if os.path.exists(sample) else ""
    entries = _lp.parse_lines(text) if text else []
    triggered = evaluate_rules(entries, default_rules())
    print(f"triggered: {len(triggered)}")
    for t in triggered:
        marker = "*" if t["triggered"] else " "
        print(f"  {marker} {t['rule'].name}: value={t['value']:.4f} threshold={t['threshold']} ({t['message']})")