Stats UI: per-anime progress rows, status chips, on-hold, streak, detail line

- server: /api/stats top_anime rows enriched — poster, total eps, cur_ep,
  last_seen, status (finished/ongoing/hold heuristic), pace (ep/day),
  sub/dub watch-time split, rewatches (backward jump >60s); new on_hold
  + best_streak top-level keys; optional ?slug= filter (json_extract)
- frontend: 'Most watched' rows = poster + meta + status chip + completion
  bar (ongoing -> hold -> finished order); stat cards swap in on-hold and
  best streak, errors demoted to a one-liner; anime detail page gets a
  progress line (E12/24 · 4h 30m · 2.1 ep/day · status); history cards
  dim + label 'on hold' after 7 days; savePos now includes total eps
- css: .chip, .stat-anime rows, .card.hold, .detail-prog; drop dead .stat-row
- test: test_stats asserts the new fields + slug filter

server/app.py

@@ -247,12 +247,18 @@ def api_events(evt: dict = Body(...), request: Request = None):
 
 
 @app.get("/api/stats")
-def api_stats(request: Request, days: int = Query(30, ge=1, le=365)):
+def api_stats(request: Request, days: int = Query(30, ge=1, le=365),
+              slug: str = Query(None)):
     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()
+    if slug:
+        positions = c.execute("SELECT ts, payload FROM events WHERE user=? AND type='position' "
+                              "AND ts>=? AND json_extract(payload, '$.slug_id')=? ORDER BY ts",
+                              (u, since, slug)).fetchall()
+    else:
+        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(
@@ -260,43 +266,87 @@ def api_stats(request: Request, days: int = Query(30, ge=1, le=365)):
         "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]
+    day_set = [r[0] for r in c.execute(
+        "SELECT DISTINCT CAST(ts/86400 AS INTEGER) FROM events WHERE user=? AND ts>=? "
+        "ORDER BY 1", (u, since))]
     c.close()
+    days_active = len(day_set)
+    best = streak = 0  # max consecutive-day run (streak)
+    for i, d in enumerate(day_set):
+        streak = streak + 1 if i and d == day_set[i - 1] + 1 else 1
+        best = max(best, streak)
 
     # 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
+    # position events fire ~10s apart; a bigger jump is a seek/resume, not
+    # viewing; a backward jump >60s within an ep counts as a rewatch
     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})
+        k = (str(e.get("slug_id")), e.get("season_id"), e.get("ep_id"))
+        s = eps.setdefault(k, {"title": e.get("title"), "poster": e.get("poster"),
+                               "total": e.get("total"), "last_ts": 0.0, "last_ep": None,
+                               "pos": None, "max": 0.0, "dur": 0.0, "lang": e.get("lang"),
+                               "secs": 0.0, "rewinds": 0, "jpn": 0.0, "eng": 0.0,
+                               "days": set()})
         p = e.get("pos") or 0
+        s["days"].add(ts // 86400)
+        s["last_ts"] = ts
+        if e.get("ep") is not None:
+            s["last_ep"] = e["ep"]
+        if e.get("poster"):
+            s["poster"] = e["poster"]
+        if e.get("total"):
+            s["total"] = e["total"]
         if s["pos"] is not None and 0 < p - s["pos"] <= 90:
             s["secs"] += p - s["pos"]
+            s["eng" if e.get("lang") == "eng" else "jpn"] += p - s["pos"]
             watch_sec[datetime.fromtimestamp(ts).weekday()] += p - s["pos"]
+        elif s["pos"] is not None and p < s["pos"] - 60:
+            s["rewinds"] += 1
         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})
+    per_anime, started, completed, on_hold = {}, 0, 0, 0
+    for (slug, _season, _ep), s in eps.items():
+        a = per_anime.setdefault(slug, {"title": s["title"], "poster": s["poster"],
+                                        "total": s["total"], "last_seen": 0.0,
+                                        "cur_ep": s["last_ep"], "minutes": 0.0,
+                                        "started": 0, "completed": 0, "rewatches": 0,
+                                        "sub": 0.0, "dub": 0.0, "days": set()})
         a["minutes"] += s["secs"] / 60
         a["started"] += 1
+        a["rewatches"] += s["rewinds"]
+        a["sub"] += s["jpn"]
+        a["dub"] += s["eng"]
+        a["days"] |= s["days"]
+        a["poster"] = a["poster"] or s["poster"]
+        a["total"] = a["total"] or s["total"]
+        a["title"] = a["title"] or s["title"]
         started += 1
+        if s["last_ts"] > a["last_seen"]:
+            a["last_seen"] = s["last_ts"]
+            a["cur_ep"] = s["last_ep"]
         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:
+    now = time.time()
+    for a in per_anime.values():
+        a["pace"] = round(a["started"] / len(a["days"]), 1)
+        # ponytail: finished = every started ep completed and (total unknown,
+        # or all eps done) — heuristic, fine for a v1 chip
+        fin = a["completed"] == a["started"] and (a["total"] is None
+                                                  or a["completed"] >= a["total"])
+        a["status"] = ("finished" if fin else
+                       "ongoing" if now - a["last_seen"] < 7 * 86400 else "hold")
+        if a["status"] == "hold":
+            on_hold += 1
         a["minutes"] = round(a["minutes"])
+        a["sub"], a["dub"] = int(a["sub"]), int(a["dub"])
+    top = sorted(per_anime.values(), key=lambda a: -a["minutes"])[:10]
 
     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
@@ -307,6 +357,8 @@ def api_stats(request: Request, days: int = Query(30, ge=1, le=365)):
         "watch_minutes": round(sum(s["secs"] for s in eps.values()) / 60),
         "episodes": {"started": started, "completed": completed},
         "days_active": days_active,
+        "best_streak": best,
+        "on_hold": on_hold,
         "plays": len(plays),
         "errors": n_errors,
         "by_weekday": [int(x) for x in watch_sec],  # seconds, Mon..Sun