#!/usr/bin/env python3
"""Read-only Whetstone usage audit for rotated nginx logs and aggregate stats."""

from __future__ import annotations

import collections
import datetime as dt
import gzip
import glob
import json
import re
import sys
from pathlib import Path
from typing import Iterable
from urllib.parse import urlsplit


LOG_RE = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[(?P<timestamp>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<target>\S+)(?: (?P<protocol>[^"]+))?" '
    r'(?P<status>\d{3}) (?P<bytes>\S+) "(?P<referer>[^"]*)" "(?P<agent>[^"]*)"$'
)
SESSION_GAP = dt.timedelta(minutes=30)
WEB_PATHS = {"/", "/for-agents", "/skill.md", "/llms.txt"}
DOC_PATHS = {"/for-agents", "/skill.md", "/llms.txt", "/api/catalog", "/api/examples"}


def open_text(path: str) -> Iterable[str]:
    if path.endswith(".gz"):
        with gzip.open(path, "rt", encoding="utf-8", errors="replace") as handle:
            yield from handle
    else:
        with open(path, "rt", encoding="utf-8", errors="replace") as handle:
            yield from handle


def parse_logs(prefix: str) -> tuple[list[dict], list[str]]:
    paths = sorted(
        glob.glob(prefix + "*"),
        key=lambda value: (Path(value).name.count("."), Path(value).name),
    )
    entries: list[dict] = []
    rejected = 0
    rejected_samples: list[str] = []
    for path in paths:
        for raw in open_text(path):
            match = LOG_RE.match(raw.rstrip("\n"))
            if not match:
                rejected += 1
                if len(rejected_samples) < 5:
                    rejected_samples.append(raw.rstrip("\n")[:300])
                continue
            row = match.groupdict()
            try:
                stamp = dt.datetime.strptime(
                    row["timestamp"], "%d/%b/%Y:%H:%M:%S %z"
                )
            except ValueError:
                rejected += 1
                continue
            target = row["target"]
            row.update(
                timestamp=stamp,
                status=int(row["status"]),
                bytes=0 if row["bytes"] == "-" else int(row["bytes"]),
                path=urlsplit(target).path,
                source=path,
            )
            entries.append(row)
    entries.sort(key=lambda row: row["timestamp"])
    if rejected:
        print(f"warning: rejected {rejected} unparsed log lines", file=sys.stderr)
        for sample in rejected_samples:
            print(f"warning: sample: {sample!r}", file=sys.stderr)
    return entries, paths


def cluster_sessions(rows: list[dict]) -> list[list[dict]]:
    sessions: list[list[dict]] = []
    for row in rows:
        if not sessions or row["timestamp"] - sessions[-1][-1]["timestamp"] > SESSION_GAP:
            sessions.append([row])
        else:
            sessions[-1].append(row)
    return sessions


def classify_client(agent_counts: collections.Counter, rows: list[dict], sessions: list[list[dict]]) -> str:
    agent_blob = " ".join(agent_counts).lower()
    active_days = len({row["timestamp"].date() for row in rows})
    mcp_exact = sum(row["path"] == "/mcp" for row in rows)
    if "sentineloracle" in agent_blob or "liveness-only" in agent_blob:
        return "declared_liveness_monitor"
    if any(token in agent_blob for token in ("aisec-registry", "rugpull", "scanner", "security-research")):
        return "registry_or_security_scanner"
    if any(token in agent_blob for token in ("bot", "crawler", "spider", "search")):
        return "crawler_or_bot"
    if active_days >= 3 and len(sessions) >= active_days and mcp_exact <= active_days * 8:
        return "periodic_probe_pattern"
    if len(sessions) == 1 and mcp_exact >= 2:
        return "one_off_programmatic_session"
    if active_days > 1 and len(sessions) > 1:
        return "recurring_client_or_monitor"
    return "sparse_or_unclassified"


def summarize_client(ip: str, rows: list[dict]) -> dict:
    sessions = cluster_sessions(rows)
    agents = collections.Counter(row["agent"] or "(empty)" for row in rows)
    paths = collections.Counter(row["path"] for row in rows)
    statuses = collections.Counter(str(row["status"]) for row in rows)
    daily: dict[str, dict[str, int]] = {}
    for row in rows:
        day = row["timestamp"].date().isoformat()
        daily.setdefault(day, {"requests": 0, "mcp": 0})
        daily[day]["requests"] += 1
        daily[day]["mcp"] += row["path"] == "/mcp"
    return {
        "ip": ip,
        "classification": classify_client(agents, rows, sessions),
        "first": rows[0]["timestamp"].isoformat(),
        "last": rows[-1]["timestamp"].isoformat(),
        "active_days": len(daily),
        "requests": len(rows),
        "mcp_exact": sum(row["path"] == "/mcp" for row in rows),
        "mcp_post": sum(row["path"] == "/mcp" and row["method"] == "POST" for row in rows),
        "mcp_get": sum(row["path"] == "/mcp" and row["method"] == "GET" for row in rows),
        "mcp_well_known": sum(row["path"].startswith("/mcp/.well-known/") for row in rows),
        "api": sum(row["path"].startswith("/api/") for row in rows),
        "web_pages": sum(row["path"] in WEB_PATHS for row in rows),
        "docs": sum(row["path"] in DOC_PATHS for row in rows),
        "sessions": [
            {
                "start": session[0]["timestamp"].isoformat(),
                "end": session[-1]["timestamp"].isoformat(),
                "duration_seconds": int(
                    (session[-1]["timestamp"] - session[0]["timestamp"]).total_seconds()
                ),
                "requests": len(session),
                "mcp": sum(row["path"] == "/mcp" for row in session),
            }
            for session in sessions
        ],
        "daily": daily,
        "top_paths": paths.most_common(10),
        "statuses": dict(statuses),
        "user_agents": agents.most_common(5),
    }


def counter_summary(stats: dict) -> dict:
    counters = stats.get("counters", {})
    core_tool_calls = {
        key: value
        for key, value in counters.items()
        if key.startswith("tool.") and not key.startswith("tool.report_card_")
    }
    return {
        "launched_at": dt.datetime.fromtimestamp(
            stats.get("launched_at", 0), tz=dt.timezone.utc
        ).isoformat(),
        "unique_client_hashes": len(stats.get("clients", {})),
        "requests_total": counters.get("requests_total", 0),
        "core_tool_calls_total": sum(core_tool_calls.values()),
        "core_mcp_tool_calls": sum(
            value for key, value in core_tool_calls.items() if key.endswith(".mcp")
        ),
        "core_rest_tool_calls": sum(
            value for key, value in core_tool_calls.items() if key.endswith(".rest")
        ),
        "core_tool_calls": core_tool_calls,
        "report_card": {
            key.removeprefix("report_card."): value
            for key, value in counters.items()
            if key.startswith("report_card.")
        },
        "mcp": {
            key.removeprefix("mcp."): value
            for key, value in counters.items()
            if key.startswith("mcp.")
        },
        "outcomes": {
            key: value
            for key, value in counters.items()
            if key.startswith("outcome.") or key.startswith("refused.")
        },
    }


def main() -> int:
    if len(sys.argv) < 3:
        print(
            "usage: audit_whetstone_usage.py LOG_PREFIX STATS_JSON [RECENT_SINCE_ISO]",
            file=sys.stderr,
        )
        return 2
    prefix, stats_path = sys.argv[1:3]
    recent_since = (
        dt.datetime.fromisoformat(sys.argv[3])
        if len(sys.argv) > 3
        else dt.datetime(2026, 7, 22, 8, 30, tzinfo=dt.timezone.utc)
    )
    entries, paths = parse_logs(prefix)
    by_ip: dict[str, list[dict]] = collections.defaultdict(list)
    for row in entries:
        by_ip[row["ip"]].append(row)
    clients = [summarize_client(ip, rows) for ip, rows in by_ip.items()]
    clients.sort(key=lambda item: (item["mcp_exact"], item["requests"]), reverse=True)

    recent_rows = [row for row in entries if row["timestamp"] >= recent_since]
    recent_ips = {row["ip"] for row in recent_rows}
    first_seen = {ip: rows[0]["timestamp"] for ip, rows in by_ip.items()}
    stats = json.loads(Path(stats_path).read_text(encoding="utf-8"))
    output = {
        "generated_at": dt.datetime.now(tz=dt.timezone.utc).isoformat(),
        "log_files": paths,
        "log_range": {
            "first": entries[0]["timestamp"].isoformat() if entries else None,
            "last": entries[-1]["timestamp"].isoformat() if entries else None,
            "requests": len(entries),
            "unique_ips": len(by_ip),
            "mcp_exact": sum(row["path"] == "/mcp" for row in entries),
            "mcp_post": sum(
                row["path"] == "/mcp" and row["method"] == "POST" for row in entries
            ),
            "mcp_get": sum(
                row["path"] == "/mcp" and row["method"] == "GET" for row in entries
            ),
            "well_known": sum(
                row["path"].startswith("/mcp/.well-known/") for row in entries
            ),
        },
        "recent": {
            "since": recent_since.isoformat(),
            "requests": len(recent_rows),
            "unique_ips": len(recent_ips),
            "new_ips": sorted(
                ip for ip in recent_ips if first_seen[ip] >= recent_since
            ),
            "mcp_exact": sum(row["path"] == "/mcp" for row in recent_rows),
            "mcp_post": sum(
                row["path"] == "/mcp" and row["method"] == "POST"
                for row in recent_rows
            ),
            "well_known": sum(
                row["path"].startswith("/mcp/.well-known/")
                for row in recent_rows
            ),
            "top_paths": collections.Counter(
                row["path"] for row in recent_rows
            ).most_common(20),
            "top_agents": collections.Counter(
                row["agent"] or "(empty)" for row in recent_rows
            ).most_common(20),
        },
        "top_paths": collections.Counter(row["path"] for row in entries).most_common(30),
        "top_agents": collections.Counter(
            row["agent"] or "(empty)" for row in entries
        ).most_common(30),
        "stats": counter_summary(stats),
        "clients": [
            client
            for client in clients
            if client["mcp_exact"] or client["mcp_well_known"]
        ],
    }
    print(json.dumps(output, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
