"""P1-5: 流量异常检测 (traffic anomaly detection).

Reads ``list[LogEntry]`` from :mod:`log_parser` and flags:

1. **Sudden traffic spikes** per client
   Compare a baseline window (first ``BASELINE_RATIO`` of entries) to the
   current window (last ``CURRENT_RATIO`` of entries) and flag clients whose
   request-per-minute rate grew more than ``SPIKE_WARN_RATIO`` (default 3x).
2. **Abnormally large responses**
   Single responses larger than ``size_threshold_mb`` (default 100 MB).
3. **High-frequency, small payload clients**
   Clients that emit lots of tiny responses (< ``size_threshold_kb``).
4. **First-time-seen clients**
   Clients whose first request happens in the current window.

All helpers defensively swallow exceptions and return safe empty values so a
bad log line never breaks the dashboard.
"""
from __future__ import annotations

import statistics
from collections import Counter, defaultdict
from datetime import datetime
from typing import Iterable

# --- Default thresholds (overridable via kwargs) ----------------------------

#: Fraction of entries used to compute the baseline (the "past")
BASELINE_RATIO = 0.8
#: Fraction of entries used to compute the current state (the "now")
CURRENT_RATIO = 0.2
#: Spike ratio above which a client is flagged as anomaly (warning)
SPIKE_WARN_RATIO = 3.0
#: Spike ratio above which a client is flagged as critical
SPIKE_CRIT_RATIO = 10.0
#: Minimum samples required for baseline; below this the result is unreliable
MIN_SAMPLES = 5

# Worst case fallback so we don't return monstrous lists for giant logs
DEFAULT_LARGE_LIMIT = 100
DEFAULT_SMALL_LIMIT = 50
DEFAULT_NEW_LIMIT = 50


# --- Internal helpers --------------------------------------------------------

def _safe_float(value, default: float = 0.0) -> float:
    """Return ``value`` as float, or ``default`` on conversion failure."""
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _entry_ts(entry) -> float | None:
    """Pull a numeric timestamp from a LogEntry, falling back to float(time)."""
    try:
        ts = entry.get("timestamp") if isinstance(entry, dict) else None
        if isinstance(ts, datetime):
            return ts.timestamp()
        if ts is not None:
            return _safe_float(ts)
        return _safe_float(entry.get("time"), 0.0) if isinstance(entry, dict) else None
    except Exception:
        return None


def _entry_field(entry, key: str, default=""):
    """Dict-or-attribute getter that never raises."""
    try:
        if isinstance(entry, dict):
            return entry.get(key, default)
        return getattr(entry, key, default)
    except Exception:
        return default


def _window_split(entries: list) -> tuple[list, list]:
    """Split ``entries`` into (baseline, current).

    First BASELINE_RATIO of entries go to the baseline, the remainder into the
    current window. The current window is dropped entirely for tiny logs.
    """
    if not entries:
        return [], []
    try:
        n = len(entries)
        cut = max(1, int(n * BASELINE_RATIO))
        baseline = entries[:cut]
        current = entries[cut:]
        # ensure both windows have a minimum sample size to be meaningful
        if len(current) < max(2, MIN_SAMPLES // 2):
            return baseline, []
        return baseline, current
    except Exception:
        return [], []


def _window_duration(window: list) -> float:
    """Compute the spanned seconds in ``window`` using timestamp field.

    Returns 0.0 when timestamps are unavailable so callers can avoid div-by-zero.
    """
    if not window:
        return 0.0
    try:
        timestamps: list[float] = []
        for e in window:
            ts = _entry_ts(e)
            if ts is not None and ts > 0:
                timestamps.append(ts)
        if len(timestamps) < 2:
            return 0.0
        span = max(timestamps) - min(timestamps)
        return span if span > 0 else 0.0
    except Exception:
        return 0.0


def _rate_per_minute(count: int, duration_sec: float) -> float:
    """Convert ``count over duration_sec`` to requests-per-minute."""
    try:
        if duration_sec <= 0:
            return float(count)
        return count / (duration_sec / 60.0)
    except Exception:
        return 0.0


def _client_counts(window: Iterable) -> Counter:
    """Build a Counter {client: count} for ``window``."""
    out: Counter = Counter()
    if not window:
        return out
    try:
        for e in window:
            client = _entry_field(e, "client", "")
            if not client:
                continue
            out[client] += 1
    except Exception:
        return out
    return out


# --- Public API --------------------------------------------------------------

def detect_client_anomalies(
    entries: list,
    baseline_window: list | None = None,
    current_window: list | None = None,
) -> list[dict]:
    """Detect clients whose recent request rate spikes vs. their own past.

    If ``baseline_window`` / ``current_window`` are omitted we split the given
    ``entries`` using :data:`BASELINE_RATIO` / :data:`CURRENT_RATIO`.

    Returns a list of dicts sorted by severity (critical first), then by ratio::

        {
            "client":          "172.16.x.y",
            "baseline_rpm":    12.3,     # requests/min in baseline window
            "current_rpm":     180.0,    # requests/min in current window
            "baseline_count":  50,
            "current_count":   200,
            "ratio":           14.6,     # current / baseline
            "is_anomaly":      True,
            "severity":        "critical" | "warning" | None,
            "baseline_window_sec": 244.0,
            "current_window_sec":  66.5,
        }
    """
    try:
        if baseline_window is None or current_window is None:
            baseline_window, current_window = _window_split(entries)

        if not baseline_window or not current_window:
            return []

        base_counts = _client_counts(baseline_window)
        cur_counts = _client_counts(current_window)

        if not base_counts or not cur_counts:
            return []

        base_dur = _window_duration(baseline_window)
        cur_dur = _window_duration(current_window)

        anomalies: list[dict] = []
        # only consider clients that appear in BOTH windows for ratio math
        for client, cur_n in cur_counts.items():
            try:
                base_n = base_counts.get(client, 0)
                if base_n < MIN_SAMPLES:
                    # New client; show as anomaly but skip ratio math
                    anomalies.append({
                        "client": client,
                        "baseline_rpm": 0.0,
                        "current_rpm": _rate_per_minute(cur_n, cur_dur),
                        "baseline_count": 0,
                        "current_count": cur_n,
                        "ratio": float("inf"),
                        "is_anomaly": True,
                        "severity": "warning",
                        "baseline_window_sec": round(base_dur, 2),
                        "current_window_sec": round(cur_dur, 2),
                    })
                    continue

                base_rpm = _rate_per_minute(base_n, base_dur)
                cur_rpm = _rate_per_minute(cur_n, cur_dur)
                # avoid div-by-zero: if base_rpm is zero, treat as infinite ratio
                if base_rpm <= 0:
                    ratio = float("inf") if cur_rpm > 0 else 0.0
                else:
                    ratio = cur_rpm / base_rpm

                if ratio > SPIKE_WARN_RATIO:
                    severity = "critical" if ratio > SPIKE_CRIT_RATIO else "warning"
                    anomalies.append({
                        "client": client,
                        "baseline_rpm": round(base_rpm, 2),
                        "current_rpm": round(cur_rpm, 2),
                        "baseline_count": base_n,
                        "current_count": cur_n,
                        "ratio": round(ratio, 2) if ratio != float("inf") else ratio,
                        "is_anomaly": True,
                        "severity": severity,
                        "baseline_window_sec": round(base_dur, 2),
                        "current_window_sec": round(cur_dur, 2),
                    })
            except Exception:
                # single client failure doesn't kill the whole result
                continue

        # sort: critical first, then highest ratio
        def _sort_key(a: dict):
            sev_rank = 0 if a.get("severity") == "critical" else 1
            ratio = a.get("ratio", 0)
            ratio_key = -ratio if isinstance(ratio, (int, float)) and ratio != float("inf") else -1e18
            return (sev_rank, ratio_key)

        try:
            anomalies.sort(key=_sort_key)
        except Exception:
            pass
        return anomalies
    except Exception:
        return []


def detect_large_requests(
    entries: list,
    size_threshold_mb: float = 100,
    limit: int = DEFAULT_LARGE_LIMIT,
) -> list[dict]:
    """Find individual responses larger than ``size_threshold_mb`` MB.

    Returned dicts are sorted by size descending and capped at ``limit``::

        {
            "time":       "2024-01-01 12:34:56",
            "timestamp":  datetime(...),
            "client":     "172.16.x.y",
            "url":        "https://...",
            "host":       "example.com",
            "method":     "GET",
            "result":     "TCP_MISS/200",
            "size_bytes": 157286400,
            "size_mb":    150.0,
        }
    """
    try:
        if not entries:
            return []
        threshold_bytes = float(size_threshold_mb) * 1024.0 * 1024.0
        out: list[dict] = []
        for e in entries:
            try:
                size = _safe_float(_entry_field(e, "size", 0), 0.0)
                if size <= threshold_bytes:
                    continue
                ts = _entry_ts(e)
                dt = None
                try:
                    dt = _entry_field(e, "timestamp", None)
                    if not isinstance(dt, datetime):
                        dt = datetime.fromtimestamp(ts) if ts else None
                except Exception:
                    dt = None
                out.append({
                    "time": dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "",
                    "timestamp": dt.isoformat() if dt else None,
                    "client": _entry_field(e, "client", ""),
                    "url": _entry_field(e, "url", ""),
                    "host": _entry_field(e, "host", ""),
                    "method": _entry_field(e, "method", ""),
                    "result": _entry_field(e, "result", ""),
                    "result_code": _entry_field(e, "result_code", ""),
                    "http_code": _entry_field(e, "http_code", ""),
                    "size_bytes": int(size),
                    "size_mb": round(size / (1024.0 * 1024.0), 2),
                })
            except Exception:
                continue

        try:
            out.sort(key=lambda x: x.get("size_bytes", 0), reverse=True)
        except Exception:
            pass
        return out[: max(1, int(limit))]
    except Exception:
        return []


def detect_high_freq_small(
    entries: list,
    size_threshold_kb: float = 1,
    min_requests: int = 100,
    limit: int = DEFAULT_SMALL_LIMIT,
) -> list[dict]:
    """Find clients that emit lots of tiny responses (< ``size_threshold_kb``).

    Tiny-response spam is often a fingerprint of polling bots, beaconing
    implants, or chatty telemetry agents.

    Returns dicts sorted by request count desc::

        {
            "client":        "172.16.x.y",
            "request_count": 540,
            "total_bytes":   420000,
            "avg_size":      778,
            "sample_url":    "https://...",
            "sample_host":   "example.com",
        }
    """
    try:
        if not entries:
            return []
        threshold_bytes = float(size_threshold_kb) * 1024.0
        counts: Counter = Counter()
        total_bytes: dict[str, int] = defaultdict(int)
        sample_url: dict[str, str] = {}
        sample_host: dict[str, str] = {}
        for e in entries:
            try:
                size = _safe_float(_entry_field(e, "size", 0), 0.0)
                if size >= threshold_bytes:
                    continue
                client = _entry_field(e, "client", "")
                if not client:
                    continue
                counts[client] += 1
                total_bytes[client] += int(size)
                if client not in sample_url:
                    sample_url[client] = _entry_field(e, "url", "")
                    sample_host[client] = _entry_field(e, "host", "")
            except Exception:
                continue

        out: list[dict] = []
        for client, n in counts.most_common():
            if n < max(1, int(min_requests)):
                continue
            tb = total_bytes.get(client, 0)
            avg = tb / n if n else 0
            out.append({
                "client": client,
                "request_count": n,
                "total_bytes": tb,
                "avg_size": round(avg, 1),
                "sample_url": sample_url.get(client, ""),
                "sample_host": sample_host.get(client, ""),
            })

        try:
            out.sort(key=lambda x: x.get("request_count", 0), reverse=True)
        except Exception:
            pass
        return out[: max(1, int(limit))]
    except Exception:
        return []


def detect_new_clients(
    entries: list,
    known_clients: set[str] | list[str] | None = None,
    limit: int = DEFAULT_NEW_LIMIT,
) -> list[dict]:
    """Find clients that appear for the first time within ``entries``.

    "First time" means: their first LogEntry timestamp is among the entries
    we are inspecting AND they aren't in ``known_clients`` (when supplied).

    Returns::

        {
            "client":        "172.16.x.y",
            "first_seen":    "2024-...",
            "first_ts":      float,
            "request_count": int,
            "sample_url":    "...",
        }
    """
    try:
        if not entries:
            return []
        if known_clients is None:
            known = set()
        elif isinstance(known_clients, set):
            known = known_clients
        else:
            known = set(known_clients or [])

        # find earliest timestamp per client
        first_ts: dict[str, float] = {}
        first_dt: dict[str, datetime] = {}
        counts: Counter = Counter()
        sample_url: dict[str, str] = {}
        for e in entries:
            try:
                client = _entry_field(e, "client", "")
                if not client:
                    continue
                ts = _entry_ts(e) or 0.0
                dt = _entry_field(e, "timestamp", None)
                if ts and (client not in first_ts or ts < first_ts[client]):
                    first_ts[client] = ts
                if isinstance(dt, datetime) and (client not in first_dt or dt < first_dt[client]):
                    first_dt[client] = dt
                counts[client] += 1
                if client not in sample_url:
                    sample_url[client] = _entry_field(e, "url", "")
            except Exception:
                continue

        # min timestamp across all entries = baseline cutoff
        all_ts = [t for t in first_ts.values() if t > 0]
        if not all_ts:
            return []
        global_min_ts = min(all_ts)
        out: list[dict] = []
        for client, ts in first_ts.items():
            try:
                # first-seen in entries exactly equal to global min means
                # they've never been seen before — anything <= min qualifies
                # as a "new" arrival within this log slice
                if ts > global_min_ts:
                    continue
                if client in known:
                    continue
                dt = first_dt.get(client)
                out.append({
                    "client": client,
                    "first_seen": dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "",
                    "first_ts": ts,
                    "request_count": counts.get(client, 1),
                    "sample_url": sample_url.get(client, ""),
                })
            except Exception:
                continue

        try:
            out.sort(key=lambda x: x.get("first_ts", 0))
        except Exception:
            pass
        return out[: max(1, int(limit))]
    except Exception:
        return []


def get_anomaly_summary(entries: list) -> dict:
    """Run all four detectors and bundle the results.

    Layout::

        {
            "anomalous_clients": [...],   # detect_client_anomalies
            "large_requests":    [...],   # detect_large_requests
            "high_freq_small":   [...],   # detect_high_freq_small
            "new_clients":       [...],   # detect_new_clients
            "total_anomalies":   int,
            "thresholds":        {...},
        }
    """
    try:
        anomalous = detect_client_anomalies(entries)
    except Exception:
        anomalous = []
    try:
        large = detect_large_requests(entries)
    except Exception:
        large = []
    try:
        small = detect_high_freq_small(entries)
    except Exception:
        small = []
    try:
        new = detect_new_clients(entries)
    except Exception:
        new = []

    total = len(anomalous) + len(large) + len(small) + len(new)

    # quick stats over the (current) window for the page header
    header_stats: dict = {}
    try:
        if entries:
            n = len(entries)
            cut = max(1, int(n * CURRENT_RATIO))
            current = entries[cut:]
            header_stats = {
                "current_window_count": len(current),
                "current_window_clients": len({_entry_field(e, "client", "") for e in current}),
            }
    except Exception:
        pass

    return {
        "anomalous_clients": anomalous,
        "large_requests": large,
        "high_freq_small": small,
        "new_clients": new,
        "total_anomalies": total,
        "thresholds": {
            "spike_warn_ratio": SPIKE_WARN_RATIO,
            "spike_crit_ratio": SPIKE_CRIT_RATIO,
            "large_size_mb": 100,
            "small_size_kb": 1,
            "min_small_requests": 100,
        },
        "header_stats": header_stats,
    }


# --- Tiny self-test when run directly --------------------------------------

if __name__ == "__main__":  # pragma: no cover
    import json
    import log_parser

    sample_path = "/root/squid-manager/sample_access.log"
    try:
        with open(sample_path, encoding="utf-8") as f:
            text = f.read()
        entries = log_parser.parse_lines(text)
    except OSError:
        entries = []

    summary = get_anomaly_summary(entries)
    print(json.dumps(summary, default=str, ensure_ascii=False, indent=2)[:2000])