"""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 '
No data yet
' return "".join( f'
{html.escape(name)}{count}
' 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'
{label}
{value}
' for label, value in cards ) return f"""

Live Demo Analytics

All-time totals from this Space runtime. Clock: Beijing / Singapore Time (UTC+8).
{clock}
{card_html}

Countries / Regions

{_list_html(_top(visitors, "country"))}

Devices

{_list_html(_top(visitors, "device"))}

Referring Sites

{_list_html(_top(visitors, "referrer"))}
""" initialize()