Add per-user stats: play/error events + /api/stats + ?stats page

- server: 'play' event on successful stream fetch, 'error' on 404 stream
  failures; new POST /api/events for client-reported playback errors
  (only type=error accepted, fields allowlisted + length-capped)
- server: GET /api/stats?days=N — watch time from position-event pos
  deltas (<=90s cap: bigger jumps are seeks), episodes started/completed
  (>=95% of dur), days active, weekday histogram, top anime/searches,
  sub/dub split, error count
- frontend: ?stats page (cards, weekday bars, most-watched, top
  searches) + 'stats' link on home; playerFail now reports the error
- test: test_stats with synthetic viewing data (deltas, isolation, 422)

server/app.py

@@ -8,9 +8,10 @@ import base64
 import json
 import os
 import time
+from collections import Counter, OrderedDict
+from datetime import datetime
 from pathlib import Path
 from base64 import urlsafe_b64encode, urlsafe_b64decode
-from collections import OrderedDict
 from urllib.parse import quote, urljoin, urlparse
 
 from fastapi import Body, FastAPI, HTTPException, Query, Request
@@ -230,6 +231,91 @@ def api_searches_post(item: dict = Body(...), request: Request = None):
     return {"ok": True}
 
 
+@app.post("/api/events")
+def api_events(evt: dict = Body(...), request: Request = None):
+    # client-reported events; only 'error' is accepted (playback failures
+    # happen in the browser, not in the request log)
+    if evt.get("type") != "error":
+        raise HTTPException(422, "unknown event type")
+    c = db.connect()
+    db.event(c, request.state.user, "error",
+             {k: evt[k] for k in ("ep_id", "ep", "lang") if k in evt}
+             | {"detail": str(evt.get("detail", ""))[:200]})
+    c.commit()
+    c.close()
+    return {"ok": True}
+
+
+@app.get("/api/stats")
+def api_stats(request: Request, days: int = Query(30, ge=1, le=365)):
+    u = request.state.user
+    since = time.time() - days * 86400
+    c = db.connect()
+    positions = c.execute("SELECT ts, payload FROM events WHERE user=? AND type='position' "
+                          "AND ts>=? ORDER BY ts", (u, since)).fetchall()
+    plays = [r[0] for r in c.execute("SELECT payload FROM events WHERE user=? "
+                                     "AND type='play' AND ts>=?", (u, since))]
+    searches = [json.loads(r[0]).get("q") for r in c.execute(
+        "SELECT payload FROM events WHERE user=? AND type='search' AND ts>=? "
+        "ORDER BY ts", (u, since))]
+    n_errors = c.execute("SELECT COUNT(*) FROM events WHERE user=? AND type='error' "
+                         "AND ts>=?", (u, since)).fetchone()[0]
+    days_active = c.execute(
+        "SELECT COUNT(DISTINCT CAST(ts/86400 AS INTEGER)) FROM events "
+        "WHERE user=? AND ts>=?", (u, since)).fetchone()[0]
+    c.close()
+
+    # ponytail: watch time = sum of positive pos deltas <= 90s per (anime, ep).
+    # position events fire ~10s apart; a bigger jump is a seek/resume, not viewing
+    eps = {}
+    watch_sec = [0.0] * 7  # Mon..Sun
+    for ts, payload in positions:
+        e = json.loads(payload)
+        k = (str(e.get("slug_id")), e.get("ep_id"))
+        s = eps.setdefault(k, {"title": e.get("title"), "pos": None, "max": 0.0,
+                               "dur": 0.0, "lang": e.get("lang"), "secs": 0.0})
+        p = e.get("pos") or 0
+        if s["pos"] is not None and 0 < p - s["pos"] <= 90:
+            s["secs"] += p - s["pos"]
+            watch_sec[datetime.fromtimestamp(ts).weekday()] += p - s["pos"]
+        s["pos"] = p
+        s["max"] = max(s["max"], p)
+        s["dur"] = e.get("dur") or s["dur"]
+        if e.get("lang"):
+            s["lang"] = e["lang"]
+
+    per_anime, started, completed = {}, 0, 0
+    for (slug, _ep), s in eps.items():
+        a = per_anime.setdefault(slug, {"title": s["title"], "minutes": 0.0,
+                                        "started": 0, "completed": 0})
+        a["minutes"] += s["secs"] / 60
+        a["started"] += 1
+        started += 1
+        if s["dur"] > 0 and s["max"] >= s["dur"] * 0.95:
+            a["completed"] += 1
+            completed += 1
+    top = sorted(per_anime.values(), key=lambda a: -a["minutes"])[:10]
+    for a in top:
+        a["minutes"] = round(a["minutes"])
+
+    plays_by_lang = Counter(json.loads(p).get("lang", "jpn") for p in plays)
+    if not plays_by_lang:  # pre-`play`-event data: fall back to position lang
+        plays_by_lang = Counter(s["lang"] for s in eps.values() if s["lang"])
+
+    return {
+        "days": days,
+        "watch_minutes": round(sum(s["secs"] for s in eps.values()) / 60),
+        "episodes": {"started": started, "completed": completed},
+        "days_active": days_active,
+        "plays": len(plays),
+        "errors": n_errors,
+        "by_weekday": [int(x) for x in watch_sec],  # seconds, Mon..Sun
+        "top_anime": top,
+        "top_searches": [q for q, _ in Counter(searches).most_common(5) if q],
+        "sub_dub": {"jpn": plays_by_lang.get("jpn", 0), "eng": plays_by_lang.get("eng", 0)},
+    }
+
+
 def _b64(url: str) -> str:
     return base64.urlsafe_b64encode(url.encode()).decode()
 
@@ -276,22 +362,34 @@ def api_episodes(anime_id: int):
 
 
 @app.get("/api/episode/{ep_id}/stream")
-def api_stream(ep_id: int, lang: str = "jpn"):
+def api_stream(ep_id: int, lang: str = "jpn", request: Request = None):
     def go():
         master = _upstream(upstream.master_m3u8, ep_id, lang)
         if not master:
             raise HTTPException(404, f"no {lang} stream for episode {ep_id}")
         return {"token": _b64(master),
                 "qualities": _upstream(upstream.master_variants, master)}
+
     try:
-        return go()
+        out = go()
     except HTTPException as e:
         # one-shot retry with a cold cache: recovers rotated/expired
         # playlist URLs from the TTL cache
         if e.status_code == 502:
             upstream.clear_cache()
-            return go()
-        raise
+            out = go()
+        else:
+            c = db.connect()
+            db.event(c, request.state.user, "error",
+                     {"ep_id": ep_id, "lang": lang, "detail": str(e.detail)})
+            c.commit()
+            c.close()
+            raise
+    c = db.connect()
+    db.event(c, request.state.user, "play", {"ep_id": ep_id, "lang": lang})
+    c.commit()
+    c.close()
+    return out
 
 
 def _rewrite_m3u8(text: str, base: str) -> str: