runtime_analytics.py

"""Privacy-conscious, runtime-local analytics for the Gradio Space."""
from __future__ import annotations

import hashlib
import html
import ipaddress
import json
import os
import sqlite3
import threading
import time
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import quote, urlparse
from urllib.request import Request, urlopen


DB_PATH = Path(os.environ.get("AUTO_PAPER_ANALYTICS_DB", "/tmp/auto-paper-list-analytics.sqlite3"))
ACTIVE_WINDOW_SECONDS = 300
_SALT = os.urandom(32)
_LOCK = threading.Lock()


def _connect() -> sqlite3.Connection:
    connection = sqlite3.connect(DB_PATH, timeout=10)
    connection.row_factory = sqlite3.Row
    return connection


def initialize() -> None:
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    with _LOCK, _connect() as db:
        db.executescript(
            """
            CREATE TABLE IF NOT EXISTS visitors (
                user_id TEXT PRIMARY KEY,
                country TEXT NOT NULL,
                device TEXT NOT NULL,
                referrer TEXT NOT NULL,
                first_seen REAL NOT NULL,
                last_seen REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS sessions (
                session_id TEXT PRIMARY KEY,
                user_id TEXT NOT NULL,
                first_seen REAL NOT NULL,
                last_seen REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS generations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                created_at REAL NOT NULL,
                pdf_count INTEGER NOT NULL
            );
            """
        )


def _headers(request: object | None) -> dict[str, str]:
    raw = getattr(getattr(request, "request", None), "headers", {}) or {}
    return {str(key).lower(): str(value) for key, value in raw.items()}


def _public_ip(request: object | None) -> str | None:
    headers = _headers(request)
    candidates = [part.strip() for part in headers.get("x-forwarded-for", "").split(",")]
    client = getattr(getattr(getattr(request, "request", None), "client", None), "host", None)
    if client:
        candidates.append(str(client))
    for candidate in candidates:
        try:
            address = ipaddress.ip_address(candidate)
        except ValueError:
            continue
        if address.is_global:
            return str(address)
    return None


def _hash(value: str) -> str:
    return hashlib.sha256(_SALT + value.encode("utf-8", errors="ignore")).hexdigest()


def _identity(request: object | None) -> tuple[str, str]:
    ip = _public_ip(request)
    session_hash = str(getattr(request, "session_hash", "") or "anonymous-session")
    return _hash(ip or session_hash), ip or ""


def _session_id(request: object | None, user_id: str) -> str:
    session_hash = str(getattr(request, "session_hash", "") or user_id)
    return _hash(session_hash)


def _country_from_ip(ip: str) -> str:
    if not ip:
        return "Unknown"
    try:
        request = Request(
            f"https://ipwho.is/{quote(ip, safe='')}",
            headers={"User-Agent": "auto-paper-list-space/1.0"},
        )
        with urlopen(request, timeout=2.5) as response:
            payload = json.load(response)
        if payload.get("success", True) and payload.get("country"):
            return str(payload["country"])
    except Exception:
        pass
    return "Unknown"


def _device(request: object | None) -> str:
    user_agent = _headers(request).get("user-agent", "").lower()
    if any(token in user_agent for token in ("mobile", "android", "iphone", "ipad", "tablet")):
        return "Mobile"
    return "Desktop"


def _referrer(request: object | None) -> str:
    value = _headers(request).get("referer", "")
    if not value:
        return "Direct"
    parsed = urlparse(value)
    return parsed.netloc or "Direct"


def record_visit(request: object | None) -> None:
    initialize()
    user_id, ip = _identity(request)
    session_id = _session_id(request, user_id)
    now = time.time()

    with _LOCK, _connect() as db:
        existing = db.execute("SELECT country FROM visitors WHERE user_id = ?", (user_id,)).fetchone()
    country = str(existing["country"]) if existing else _country_from_ip(ip)

    with _LOCK, _connect() as db:
        db.execute(
            """
            INSERT INTO visitors(user_id, country, device, referrer, first_seen, last_seen)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(user_id) DO UPDATE SET
                country = excluded.country,
                device = excluded.device,
                referrer = excluded.referrer,
                last_seen = excluded.last_seen
            """,
            (user_id, country, _device(request), _referrer(request), now, now),
        )
        db.execute(
            """
            INSERT INTO sessions(session_id, user_id, first_seen, last_seen)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(session_id) DO UPDATE SET last_seen = excluded.last_seen
            """,
            (session_id, user_id, now, now),
        )


def record_generation(request: object | None, pdf_count: int) -> None:
    record_visit(request)
    user_id, _ = _identity(request)
    with _LOCK, _connect() as db:
        db.execute(
            "INSERT INTO generations(user_id, created_at, pdf_count) VALUES (?, ?, ?)",
            (user_id, time.time(), pdf_count),
        )


def _top(rows: list[sqlite3.Row], key: str, *, limit: int = 6) -> list[tuple[str, int]]:
    return Counter(str(row[key]) for row in rows).most_common(limit)


def _list_html(items: list[tuple[str, int]]) -> str:
    if not items:
        return '<div class="analytics-empty">No data yet</div>'
    return "".join(
        f'<div class="analytics-list-row"><span>{html.escape(name)}</span><strong>{count}</strong></div>'
        for name, count in items
    )


def render_dashboard() -> str:
    initialize()
    now = time.time()
    with _LOCK, _connect() as db:
        visitors = db.execute("SELECT * FROM visitors").fetchall()
        sessions = db.execute("SELECT * FROM sessions").fetchall()
        generations = db.execute("SELECT * FROM generations").fetchall()

    active_now = sum(row["last_seen"] >= now - ACTIVE_WINDOW_SECONDS for row in sessions)
    average_session = (
        sum(max(0, row["last_seen"] - row["first_seen"]) for row in sessions) / len(sessions)
        if sessions
        else 0
    )
    average_pdfs = sum(row["pdf_count"] for row in generations) / len(generations) if generations else 0
    clock = datetime.now(timezone(timedelta(hours=8))).strftime("%H:%M:%S UTC+8")

    cards = [
        ("All-Time Active Users", str(len(visitors))),
        ("All-Time Generations", str(len(generations))),
        ("Average PDFs / Generation", f"{average_pdfs:.1f}"),
        ("Active Now", str(active_now)),
        ("Average Session Time", f"{average_session:.0f}s"),
    ]
    card_html = "".join(
        f'<div class="analytics-card"><div>{label}</div><strong>{value}</strong></div>'
        for label, value in cards
    )

    return f"""
    <style>
      .analytics-panel {{ border:1px solid #d5deea; border-radius:12px; padding:22px; background:#f8fafc; color:#10213b; }}
      .analytics-head {{ display:flex; justify-content:space-between; gap:16px; align-items:start; }}
      .analytics-head h2 {{ margin:0 0 4px; font-size:25px; }}
      .analytics-subtle {{ color:#53627a; }}
      .analytics-clock {{ white-space:nowrap; color:#53627a; }}
      .analytics-cards {{ display:grid; grid-template-columns:repeat(5,minmax(0,1fr)); gap:12px; margin-top:18px; }}
      .analytics-card, .analytics-list {{ border:1px solid #d5deea; border-radius:10px; background:white; padding:16px; }}
      .analytics-card div {{ color:#53627a; min-height:42px; }}
      .analytics-card strong {{ display:block; margin-top:4px; font-size:30px; }}
      .analytics-lists {{ display:grid; grid-template-columns:repeat(3,minmax(0,1fr)); gap:12px; margin-top:12px; }}
      .analytics-list h3 {{ margin:0 0 12px; font-size:18px; }}
      .analytics-list-row {{ display:flex; justify-content:space-between; gap:12px; margin:9px 0; }}
      .analytics-empty {{ color:#7a8799; }}
      @media(max-width:900px) {{ .analytics-cards {{ grid-template-columns:repeat(2,1fr); }} .analytics-lists {{ grid-template-columns:1fr; }} }}
    </style>
    <section class="analytics-panel">
      <div class="analytics-head">
        <div><h2>Live Demo Analytics</h2><div class="analytics-subtle">All-time totals from this Space runtime. Clock: Beijing / Singapore Time (UTC+8).</div></div>
        <div class="analytics-clock">{clock}</div>
      </div>
      <div class="analytics-cards">{card_html}</div>
      <div class="analytics-lists">
        <div class="analytics-list"><h3>Countries / Regions</h3>{_list_html(_top(visitors, "country"))}</div>
        <div class="analytics-list"><h3>Devices</h3>{_list_html(_top(visitors, "device"))}</div>
        <div class="analytics-list"><h3>Referring Sites</h3>{_list_html(_top(visitors, "referrer"))}</div>
      </div>
    </section>
    """


initialize()