Track privacy-safe website traffic (#3392)

scripts/collect_website_traffic.py

@@ -0,0 +1,330 @@
+#!/usr/bin/env python3
+"""Aggregate privacy-safe funasr.com traffic metrics from Nginx access logs."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import glob
+import gzip
+import hashlib
+import json
+import os
+import re
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Iterable, Iterator, NamedTuple, Optional, Sequence
+from urllib.parse import unquote, urlsplit
+
+
+LOG_PATTERN = re.compile(
+    r'^(?P<client_ip>\S+) \S+ \S+ \[(?P<timestamp>[^]]+)] '
+    r'"(?P<method>[A-Z]+) (?P<target>\S+) [^"]+" '
+    r'(?P<status>\d{3}) \S+ "(?P<referrer>[^"]*)" "(?P<user_agent>[^"]*)"$'
+)
+BOT_PATTERN = re.compile(
+    r"bot|spider|crawler|slurp|headless|lighthouse|pagespeed|curl/|wget/|"
+    r"python-requests|aiohttp|scrapy|go-http-client|uptime|monitoring|zgrab|masscan",
+    re.IGNORECASE,
+)
+STATIC_EXTENSIONS = {
+    ".avif",
+    ".css",
+    ".eot",
+    ".gif",
+    ".ico",
+    ".jpeg",
+    ".jpg",
+    ".js",
+    ".json",
+    ".map",
+    ".mp3",
+    ".mp4",
+    ".ogg",
+    ".pdf",
+    ".png",
+    ".svg",
+    ".txt",
+    ".wav",
+    ".webm",
+    ".webp",
+    ".woff",
+    ".woff2",
+    ".xml",
+    ".zip",
+}
+IGNORED_PREFIXES = ("/.well-known/", "/assets/", "/fonts/", "/images/", "/static/")
+DOC_PREFIXES = ("/docs/", "/en/docs/", "/guide/", "/en/guide/")
+DOC_FILENAMES = {
+    "agent.html",
+    "benchmark.html",
+    "cli-tutorial.html",
+    "download.html",
+    "ecosystem.html",
+    "llama-cpp.html",
+    "models.html",
+    "quickstart.html",
+    "vs-whisper.html",
+}
+SECTIONS = ("homepage", "docs", "blog", "donors", "other")
+
+
+class AccessRecord(NamedTuple):
+    client_ip: str
+    timestamp: dt.datetime
+    method: str
+    path: str
+    status: int
+    referrer_host: str
+    user_agent: str
+
+
+def normalize_path(target: str) -> str:
+    path = unquote(urlsplit(target).path or "/")
+    if not path.startswith("/"):
+        path = f"/{path}"
+    return path
+
+
+def referrer_host(referrer: str) -> str:
+    if not referrer or referrer == "-":
+        return ""
+    return (urlsplit(referrer).hostname or "").lower()
+
+
+def parse_access_log_line(line: str) -> Optional[AccessRecord]:
+    match = LOG_PATTERN.match(line.rstrip("\n"))
+    if match is None:
+        return None
+    try:
+        timestamp = dt.datetime.strptime(match.group("timestamp"), "%d/%b/%Y:%H:%M:%S %z")
+        status = int(match.group("status"))
+    except ValueError:
+        return None
+    return AccessRecord(
+        client_ip=match.group("client_ip"),
+        timestamp=timestamp,
+        method=match.group("method"),
+        path=normalize_path(match.group("target")),
+        status=status,
+        referrer_host=referrer_host(match.group("referrer")),
+        user_agent=match.group("user_agent"),
+    )
+
+
+def is_page_path(path: str) -> bool:
+    lowered = path.lower()
+    if lowered in {"/favicon.ico", "/robots.txt", "/sitemap.xml"}:
+        return False
+    if lowered.startswith(IGNORED_PREFIXES):
+        return False
+    return Path(lowered).suffix not in STATIC_EXTENSIONS
+
+
+def classify_page(path: str) -> str:
+    lowered = path.lower()
+    if lowered in {"/", "/index.html", "/en", "/en/", "/en/index.html"}:
+        return "homepage"
+    if lowered.startswith(("/blog/", "/en/blog/")):
+        return "blog"
+    if lowered in {"/donors.html", "/en/donors.html"}:
+        return "donors"
+    if lowered.startswith(DOC_PREFIXES) or Path(lowered).name in DOC_FILENAMES:
+        return "docs"
+    return "other"
+
+
+def anonymized_visitor(record: AccessRecord, salt: bytes) -> bytes:
+    value = f"{record.client_ip}\0{record.user_agent}".encode("utf-8", errors="replace")
+    return hashlib.blake2b(value, key=salt, digest_size=16).digest()
+
+
+def aggregate_records(
+    records: Iterable[Optional[AccessRecord]],
+    *,
+    now: dt.datetime,
+    days: int,
+    top_limit: int,
+    visitor_salt: Optional[bytes] = None,
+) -> dict:
+    if now.tzinfo is None:
+        raise ValueError("now must include a timezone")
+    if days <= 0:
+        raise ValueError("days must be positive")
+    if top_limit <= 0:
+        raise ValueError("top_limit must be positive")
+
+    cutoff = now - dt.timedelta(days=days)
+    salt = visitor_salt or os.urandom(16)
+    page_views = 0
+    visitors = set()
+    section_views = Counter()
+    section_visitors = defaultdict(set)
+    path_views = Counter()
+    path_visitors = defaultdict(set)
+    referrer_views = Counter()
+    filtered = Counter()
+
+    for record in records:
+        if record is None:
+            filtered["malformed"] += 1
+            continue
+        record_time = record.timestamp.astimezone(now.tzinfo)
+        if record_time < cutoff or record_time > now:
+            filtered["outside_window"] += 1
+            continue
+        if record.method not in {"GET", "HEAD"}:
+            filtered["non_page_methods"] += 1
+            continue
+        if record.status != 200:
+            filtered["non_success"] += 1
+            continue
+        if BOT_PATTERN.search(record.user_agent):
+            filtered["bots"] += 1
+            continue
+        if not is_page_path(record.path):
+            filtered["static_assets"] += 1
+            continue
+
+        visitor = anonymized_visitor(record, salt)
+        section = classify_page(record.path)
+        page_views += 1
+        visitors.add(visitor)
+        section_views[section] += 1
+        section_visitors[section].add(visitor)
+        path_views[record.path] += 1
+        path_visitors[record.path].add(visitor)
+        if record.referrer_host and not (
+            record.referrer_host == "funasr.com" or record.referrer_host.endswith(".funasr.com")
+        ):
+            referrer_views[record.referrer_host] += 1
+
+    sections = {
+        section: {
+            "page_views": section_views[section],
+            "unique_visitors": len(section_visitors[section]),
+        }
+        for section in SECTIONS
+    }
+    top_paths = [
+        {
+            "path": path,
+            "page_views": views,
+            "unique_visitors": len(path_visitors[path]),
+        }
+        for path, views in sorted(path_views.items(), key=lambda item: (-item[1], item[0]))[:top_limit]
+    ]
+    top_referrers = [
+        {"host": host, "page_views": views}
+        for host, views in sorted(referrer_views.items(), key=lambda item: (-item[1], item[0]))[:top_limit]
+    ]
+    filter_names = (
+        "malformed",
+        "outside_window",
+        "non_page_methods",
+        "non_success",
+        "bots",
+        "static_assets",
+    )
+    return {
+        "collected_at_utc": now.astimezone(dt.timezone.utc).isoformat(),
+        "window_start_utc": cutoff.astimezone(dt.timezone.utc).isoformat(),
+        "days": days,
+        "page_views": page_views,
+        "unique_visitors": len(visitors),
+        "sections": sections,
+        "top_paths": top_paths,
+        "top_referrers": top_referrers,
+        "filtered": {name: filtered[name] for name in filter_names},
+        "privacy": "Aggregate counts only; raw client IPs and user agents are never emitted.",
+    }
+
+
+def iter_log_lines(paths: Iterable[Path]) -> Iterator[str]:
+    for path in paths:
+        opener = gzip.open if path.suffix == ".gz" else open
+        with opener(path, "rt", encoding="utf-8", errors="replace") as handle:
+            yield from handle
+
+
+def resolve_log_paths(patterns: Sequence[str]) -> list[Path]:
+    paths = []
+    seen = set()
+    for pattern in patterns:
+        for match in sorted(glob.glob(pattern)):
+            path = Path(match)
+            if path.is_file() and path not in seen:
+                paths.append(path)
+                seen.add(path)
+    return paths
+
+
+def collect_traffic(
+    paths: Sequence[Path],
+    *,
+    now: dt.datetime,
+    days: int,
+    top_limit: int,
+) -> dict:
+    records = (parse_access_log_line(line) for line in iter_log_lines(paths))
+    result = aggregate_records(records, now=now, days=days, top_limit=top_limit)
+    result["source_file_count"] = len(paths)
+    return result
+
+
+def parse_now(value: Optional[str]) -> dt.datetime:
+    if value is None:
+        return dt.datetime.now(dt.timezone.utc)
+    parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if parsed.tzinfo is None:
+        raise argparse.ArgumentTypeError("--now must include a timezone")
+    return parsed
+
+
+def render_text(metrics: dict) -> str:
+    lines = [
+        f"funasr.com traffic ({metrics['days']} days)",
+        f"Page views: {metrics['page_views']:,}",
+        f"Approximate unique visitors: {metrics['unique_visitors']:,}",
+    ]
+    for section in SECTIONS:
+        values = metrics["sections"][section]
+        lines.append(
+            f"{section}: {values['page_views']:,} views / "
+            f"{values['unique_visitors']:,} approximate visitors"
+        )
+    return "\n".join(lines)
+
+
+def main(argv: Optional[Sequence[str]] = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--log",
+        action="append",
+        required=True,
+        help="Access-log path or glob. Repeat for multiple files.",
+    )
+    parser.add_argument("--days", type=int, default=30, help="Trailing window length (default: 30).")
+    parser.add_argument("--top", type=int, default=10, help="Number of top paths/referrers (default: 10).")
+    parser.add_argument("--now", help="Timezone-aware ISO timestamp for reproducible backfills.")
+    parser.add_argument("--format", choices=("text", "json"), default="text")
+    args = parser.parse_args(argv)
+
+    try:
+        now = parse_now(args.now)
+        paths = resolve_log_paths(args.log)
+        if not paths:
+            parser.error("no access-log files matched --log")
+        metrics = collect_traffic(paths, now=now, days=args.days, top_limit=args.top)
+    except (OSError, ValueError) as error:
+        parser.error(str(error))
+
+    if args.format == "json":
+        print(json.dumps(metrics, ensure_ascii=False, indent=2))
+    else:
+        print(render_text(metrics))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())