Implement analytics feature
Add server-side visit analytics collection, a public-page ping endpoint, and a full-screen AnalyticsView for admins. Backend: - Add pagerite/analytics.py: Analytics/Visit model, Store, and persistence - Wire /_a ping endpoint and GET /_api/analytics into pagerite/app.py Frontend: - Add full-screen AnalyticsView with visitor charts and transition map - Add VisitorCharts and TransitionGraph subcomponents - Add analytics JS helpers in frontend/src/analytics/ - Send navigation pings from frontend/src/pagerite.js - Mount AnalyticsView from frontend/src/main.js - Document the feature in docs/analytics.md and update AGENTS.md
This commit is contained in:
@@ -0,0 +1,174 @@
|
||||
"""Server-side visit analytics (collection only; see docs/analytics.md).
|
||||
|
||||
Events come from navigation pings POSTed to /_a by pagerite.js: the first
|
||||
ping on page load starts a visit, later pings extend it, and pings with no
|
||||
known session start a fresh one (missing data, not dropped). The document
|
||||
GET handler only stashes the entry referer (external https origin) in an
|
||||
in-memory IP -> referer table, consumed when the ping starts the visit;
|
||||
nothing is counted without a ping (bots and admin browsing stay invisible).
|
||||
The session map is in-memory only; IPs are never persisted.
|
||||
|
||||
Data is a msgspec Struct JSON-dumped to its own file (not the kanta db),
|
||||
rewritten atomically on every recorded event.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import msgspec
|
||||
|
||||
|
||||
class Visit(msgspec.Struct, omit_defaults=True):
|
||||
"""One visit: the initial-load data plus everything seen afterwards.
|
||||
|
||||
``trail`` holds page paths and external exit origins in first-seen
|
||||
order; re-visiting an already seen page does not append. The entry
|
||||
page itself is in ``entry``, not in the trail.
|
||||
"""
|
||||
|
||||
start: datetime
|
||||
entry: str
|
||||
#: External https origin of the initial load, "" for direct visits.
|
||||
referer: str = ""
|
||||
trail: list[str] = []
|
||||
|
||||
|
||||
class Analytics(msgspec.Struct, omit_defaults=True):
|
||||
"""Root of the analytics JSON file. Append-only by design: old data is
|
||||
dropped by deleting list entries / bucket keys."""
|
||||
|
||||
visits: list[Visit] = []
|
||||
#: Page transition matrix: from -> to -> count. ``from`` is the referer
|
||||
#: origin or "(direct)" for initial loads, a page path for pings.
|
||||
transitions: dict[str, dict[str, int]] = {}
|
||||
#: Page views per 5-minute bucket: path -> bucket ISO -> count (sparse).
|
||||
views: dict[str, dict[str, int]] = {}
|
||||
#: New visits per 5-minute bucket: bucket ISO -> count (sparse).
|
||||
site_visits: dict[str, int] = {}
|
||||
|
||||
|
||||
def _bucket(now: datetime) -> str:
|
||||
"""Start of the 5-minute interval containing ``now``, as ISO string."""
|
||||
return now.replace(minute=now.minute // 5 * 5, second=0, microsecond=0).isoformat()
|
||||
|
||||
|
||||
def _origin(url: str) -> str | None:
|
||||
"""The origin part of an https URL (scheme://host[:port]), else None."""
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
except ValueError:
|
||||
return None
|
||||
if parsed.scheme != "https" or not parsed.netloc:
|
||||
return None
|
||||
return f"https://{parsed.netloc}"
|
||||
|
||||
|
||||
_SEGMENT = re.compile(r"[a-z0-9][a-z0-9_-]*")
|
||||
|
||||
|
||||
def _internal_path(to: str) -> str | None:
|
||||
"""A valid internal page path ("/" or slug segments), else None."""
|
||||
path = to.split("?")[0].split("#")[0].strip("/")
|
||||
if not path:
|
||||
return "/"
|
||||
if all(_SEGMENT.fullmatch(seg) for seg in path.split("/")):
|
||||
return f"/{path}"
|
||||
return None
|
||||
|
||||
|
||||
class Store:
|
||||
"""In-memory analytics data plus the (IP, UA) -> visit session map."""
|
||||
|
||||
def __init__(self, path: Path) -> None:
|
||||
self.path = path
|
||||
self.data = Analytics()
|
||||
if path.exists():
|
||||
try:
|
||||
self.data = msgspec.json.decode(path.read_bytes(), type=Analytics)
|
||||
except (msgspec.DecodeError, OSError):
|
||||
pass # corrupt/unreadable file: start fresh
|
||||
#: (ip, user-agent) -> index of the current visit in data.visits
|
||||
self.sessions: dict[tuple[str, str], int] = {}
|
||||
#: ip -> external https origin of the latest document GET carrying
|
||||
#: one, stashed for the visit the client's initial ping starts.
|
||||
#: Internal or absent referers never touch the table.
|
||||
self.pending_referers: dict[str, str] = {}
|
||||
|
||||
def _save(self) -> None:
|
||||
"""Rewrite the JSON file atomically (temp file + rename)."""
|
||||
try:
|
||||
fd, tmp = tempfile.mkstemp(
|
||||
dir=self.path.parent, prefix=self.path.name, suffix=".tmp"
|
||||
)
|
||||
with os.fdopen(fd, "wb") as f:
|
||||
f.write(msgspec.json.encode(self.data))
|
||||
os.replace(tmp, self.path)
|
||||
except OSError:
|
||||
pass # analytics must never break page serving
|
||||
|
||||
def _count(self, table: dict[str, int], key: str) -> None:
|
||||
table[key] = table.get(key, 0) + 1
|
||||
|
||||
def _new_visit(self, entry: str, referer: str, key: tuple[str, str]) -> Visit:
|
||||
now = datetime.now(UTC)
|
||||
visit = Visit(start=now, entry=entry, referer=referer)
|
||||
self.data.visits.append(visit)
|
||||
self.sessions[key] = len(self.data.visits) - 1
|
||||
self._count(self.data.site_visits, _bucket(now))
|
||||
self._count(self.data.views.setdefault(entry, {}), _bucket(now))
|
||||
self._count(
|
||||
self.data.transitions.setdefault(referer or "(direct)", {}), entry
|
||||
)
|
||||
return visit
|
||||
|
||||
def entry_referer(self, referer: str, own_origin: str, ip: str) -> None:
|
||||
"""Stash the entry referer of a document GET for ping attribution.
|
||||
|
||||
Nothing is counted here — the client's initial /_a ping starts the
|
||||
visit (only non-admin clients ping). Only a cross-origin https
|
||||
referer updates the table; an internal or absent referer leaves any
|
||||
stashed origin untouched.
|
||||
"""
|
||||
if not referer:
|
||||
return
|
||||
origin = _origin(referer)
|
||||
if origin is None or origin == own_origin:
|
||||
return
|
||||
self.pending_referers[ip] = origin
|
||||
|
||||
def ping(self, from_: str, to: str, ip: str, ua: str) -> None:
|
||||
"""Record a client navigation ping ({from, to} from pagerite.js).
|
||||
|
||||
``to`` is an internal path ("/...") or an https origin for exit
|
||||
links; anything else is ignored. The transition is always counted;
|
||||
the trail only grows on first sight of a page within the visit.
|
||||
A ping with no known session starts a fresh visit, consuming the
|
||||
referer stashed by the document GET if there is one.
|
||||
"""
|
||||
if to.startswith("/") and not to.startswith("//"):
|
||||
target = _internal_path(to) or ""
|
||||
else:
|
||||
target = _origin(to) or ""
|
||||
if not target or (not to.startswith("/") and target != to):
|
||||
return
|
||||
key = (ip, ua)
|
||||
index = self.sessions.get(key)
|
||||
fr = (_internal_path(from_) or "(direct)") if from_ else "(direct)"
|
||||
if index is None or index >= len(self.data.visits):
|
||||
# No known session: the initial ping of a fresh page load (or
|
||||
# missing data after a server restart) — start a visit.
|
||||
visit = self._new_visit(target, self.pending_referers.pop(ip, ""), key)
|
||||
else:
|
||||
visit = self.data.visits[index]
|
||||
now = datetime.now(UTC)
|
||||
if target.startswith("/"):
|
||||
self._count(self.data.views.setdefault(target, {}), _bucket(now))
|
||||
self._count(self.data.transitions.setdefault(fr, {}), target)
|
||||
# First-seen only: repeat pages and repeated exits don't append.
|
||||
if visit.entry != target and target not in visit.trail:
|
||||
visit.trail.append(target)
|
||||
self._save()
|
||||
+59
-1
@@ -20,15 +20,17 @@ from contextlib import asynccontextmanager
|
||||
from datetime import UTC, datetime
|
||||
from email.utils import format_datetime
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import blake3
|
||||
import msgspec
|
||||
from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||||
from fastapi_vue import Frontend
|
||||
from kanta import Kanta
|
||||
from pydantic import BaseModel
|
||||
|
||||
from pagerite import seed, views
|
||||
from pagerite import analytics, seed, views
|
||||
from pagerite.__main__ import DEVMODE
|
||||
from pagerite.data import (
|
||||
Data,
|
||||
@@ -43,6 +45,12 @@ from pagerite.markdown import has_h1, render, toggle_task
|
||||
|
||||
DB_PATH = os.getenv("PAGERITE_DB", "pagerite.kantadb")
|
||||
|
||||
# Visit analytics go to their own JSON file, not the kanta database.
|
||||
ANALYTICS_PATH = Path(
|
||||
os.getenv("PAGERITE_ANALYTICS", DB_PATH.replace(".kantadb", "") + ".analytics.json")
|
||||
)
|
||||
analytics_store = analytics.Store(ANALYTICS_PATH)
|
||||
|
||||
# Our own data root; kanta edits it in place, reads are plain attribute access.
|
||||
data = Data()
|
||||
kanta = Kanta(DB_PATH, data)
|
||||
@@ -495,6 +503,41 @@ async def delete_page(path: str) -> None:
|
||||
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*$")
|
||||
|
||||
|
||||
def _client_ip(request: Request) -> str:
|
||||
"""Client IP: first X-Forwarded-For hop (we sit behind a proxy), else
|
||||
the direct peer."""
|
||||
forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
|
||||
return forwarded or (request.client.host if request.client else "")
|
||||
|
||||
|
||||
class AnalyticsPing(BaseModel):
|
||||
"""Navigation ping from pagerite.js (see docs/analytics.md)."""
|
||||
|
||||
fr: str = ""
|
||||
to: str
|
||||
|
||||
|
||||
@app.post("/_a", status_code=204)
|
||||
async def analytics_ping(ping: AnalyticsPing, request: Request) -> None:
|
||||
"""Record a navigation ping ({fr, to}); fire-and-forget, never fails."""
|
||||
analytics_store.ping(
|
||||
ping.fr, ping.to, _client_ip(request),
|
||||
request.headers.get("user-agent", ""),
|
||||
)
|
||||
|
||||
|
||||
def _track_entry(path: str, request: Request) -> None:
|
||||
"""Stash the referer of the document GET for the initial ping.
|
||||
|
||||
Nothing is counted on the GET itself — the client's /_a ping starts the
|
||||
visit, so bots and admin browsing never register.
|
||||
"""
|
||||
own_origin = f"https://{urlparse(str(request.base_url)).netloc}"
|
||||
analytics_store.entry_referer(
|
||||
request.headers.get("referer", ""), own_origin, _client_ip(request)
|
||||
)
|
||||
|
||||
|
||||
def _http_date(dt: datetime) -> str:
|
||||
"""RFC 7231 date for the Last-Modified header."""
|
||||
return format_datetime(dt.astimezone(UTC), usegmt=True)
|
||||
@@ -519,6 +562,18 @@ def _check_reserved(path: str) -> None:
|
||||
)
|
||||
|
||||
|
||||
@app.get("/_api/analytics")
|
||||
async def get_analytics() -> Response:
|
||||
"""The collected visit analytics as JSON (see docs/analytics.md).
|
||||
|
||||
Admin-only via the /_api forward-auth gate, like every management
|
||||
endpoint. Powers the full-screen analytics viewer in the frontend.
|
||||
"""
|
||||
return Response(
|
||||
msgspec.json.encode(analytics_store.data), media_type="application/json"
|
||||
)
|
||||
|
||||
|
||||
@app.websocket("/_api/ws/editor")
|
||||
async def editor_ws(ws: WebSocket) -> None:
|
||||
"""Editor session: open pages, render previews, save — over one socket.
|
||||
@@ -699,6 +754,7 @@ async def show_page(request: Request, path: str) -> HTMLResponse | Response:
|
||||
etag = f'"{path}@{node.modified.timestamp()}v{data.version}"'
|
||||
if request.headers.get("if-none-match") == etag:
|
||||
return Response(status_code=304)
|
||||
_track_entry(path, request)
|
||||
return HTMLResponse(
|
||||
views.render_page(data.menu, path, data.brand, data.custom_css, data.theme, data.favicon, data.brand_html, str(request.base_url).rstrip("/")),
|
||||
headers={
|
||||
@@ -710,6 +766,7 @@ async def show_page(request: Request, path: str) -> HTMLResponse | Response:
|
||||
if node is not None and node.published and node.content is None:
|
||||
# Category label without a landing page: placeholder with the pen
|
||||
# to create it (404 — no page here, but the node is real).
|
||||
_track_entry(path, request)
|
||||
return HTMLResponse(
|
||||
views.render_category(data.menu, path, data.brand, data.custom_css, data.theme, data.favicon, data.brand_html),
|
||||
404,
|
||||
@@ -724,4 +781,5 @@ async def show_page(request: Request, path: str) -> HTMLResponse | Response:
|
||||
for slug, item in sorted_nodes(data.menu):
|
||||
if item.published:
|
||||
return RedirectResponse(f"/{slug}")
|
||||
_track_entry(path, request)
|
||||
return HTMLResponse(views.render_not_found(data.menu, path, data.brand, data.custom_css, data.theme, data.favicon, data.brand_html), 404)
|
||||
|
||||
Reference in New Issue
Block a user