#!/usr/bin/env python3
# Counts AI crawler hits in an access log and checks each IP against the operator ranges. User agent alone can be spoofed; the ip_verified column is the check.
# Tested October 7, 2026 on 7 sample lines; a spoofed GPTBot from 1.2.3.4 was flagged NO.
"""ai_bot_report.py: count AI crawler hits in an access log and check each IP
against the operator's published IP ranges.

Usage:
  python3 ai_bot_report.py access.log [more.log ...] > ai-bots.csv
  zcat access.log.*.gz | python3 ai_bot_report.py - > ai-bots.csv

Expects the common "combined" log format (Nginx/Apache default):
  IP - - [10/Oct/2026:13:55:36 +0000] "GET /path HTTP/1.1" 200 512 "referer" "user-agent"
"""
import csv, ipaddress, json, re, sys, urllib.request
from collections import Counter

# User-agent token -> the operator's published IP list (checked Oct 2026)
BOTS = {
    "OAI-SearchBot":    "https://openai.com/searchbot.json",
    "ChatGPT-User":     "https://openai.com/chatgpt-user.json",
    "GPTBot":           "https://openai.com/gptbot.json",
    "Claude-SearchBot": "https://claude.com/crawling/bots.json",
    "Claude-User":      "https://claude.com/crawling/bots.json",
    "ClaudeBot":        "https://claude.com/crawling/bots.json",
    "PerplexityBot":    "https://www.perplexity.ai/perplexitybot.json",
    "Perplexity-User":  "https://www.perplexity.ai/perplexity-user.json",
    "Googlebot":        "https://developers.google.com/static/crawling/ipranges/common-crawlers.json",
    "bingbot":          "https://www.bing.com/toolbox/bingbot.json",
}
LINE = re.compile(r'^(\S+) \S+ \S+ \[([^:\]]+)[^\]]*\] "(?:\S+) (\S+)[^"]*" (\d{3}) \S+ "[^"]*" "([^"]*)"')

def load_ranges(url, cache={}):
    if url not in cache:
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (ai-bot-report)"})
            with urllib.request.urlopen(req, timeout=20) as r:
                data = json.load(r)
            cache[url] = [ipaddress.ip_network(c, strict=False)
                          for c in (p.get("ipv4Prefix") or p.get("ipv6Prefix") for p in data.get("prefixes", []))
                          if c]
        except Exception as e:  # list unreachable: mark "unknown" rather than guess
            print(f"warning: could not load {url}: {e}", file=sys.stderr)
            cache[url] = None
    return cache[url]

def verify(ip, url):
    nets = load_ranges(url)
    if nets is None:
        return "unknown"
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return "unknown"
    return "yes" if any(addr in n for n in nets) else "NO"

def main(paths):
    counts = Counter()
    for path in paths or ["-"]:
        fh = sys.stdin if path == "-" else open(path, encoding="utf-8", errors="replace")
        for line in fh:
            m = LINE.match(line)
            if not m:
                continue
            ip, day, url_path, status, ua = m.groups()
            bot = next((b for b in BOTS if b.lower() in ua.lower()), None)
            if bot:
                counts[(day, bot, verify(ip, BOTS[bot]), status, url_path)] += 1
    w = csv.writer(sys.stdout)
    w.writerow(["date", "bot", "ip_verified", "status", "path", "hits"])
    for (day, bot, ok, status, url_path), n in sorted(counts.items()):
        w.writerow([day, bot, ok, status, url_path, n])

if __name__ == "__main__":
    main(sys.argv[1:])