"""app.py — ani-web: FastAPI over upstream.py + HLS proxy + static UI. Browser playback: hls.anidb.app sends no CORS headers, so the player is pointed at /stream/{token}; the proxy fetches upstream (with the same Cloudflare workaround) and rewrites playlist URIs to go through itself. """ import base64 import json import os import threading import time from collections import Counter, OrderedDict from datetime import datetime from pathlib import Path from base64 import urlsafe_b64encode, urlsafe_b64decode from urllib.parse import quote, urljoin, urlparse from fastapi import Body, FastAPI, HTTPException, Query, Request from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from itsdangerous import BadSignature, TimestampSigner from . import db, upstream, users ROOT = Path(__file__).parent.parent app = FastAPI(title="ani-web") # latest probe result + last 20 samples (~1h at 3-min spacing), for GET /api/cdn _CDN = {"state": "unknown", "kbps": 0.0, "ts": 0.0, "hist": []} def _cdn_probe(): """Background probe: pull 64KB from the CDN every 3 min and log state flips, so server.log shows when a degraded window ends (CDN RECOVERED). Probe target: any valid episode (36019 = JJK S2 E25). ponytail: single fixed episode; if it ever disappears, the probe logs DEGRADED forever — swap the ID, don't build a picker.""" def first(pl, base): l = next(x.strip() for x in pl.splitlines() if x.strip() and not x.startswith("#")) return l if l.startswith("http") else base + l state = None while True: try: m = upstream.master_m3u8(36019, "jpn") if not m: raise RuntimeError("no master m3u8") base = m.rsplit("/", 1)[0] + "/" v = first(upstream._get(m, timeout=10).text, base) seg = first(upstream._get(v, timeout=10).text, base) t = time.time() r = upstream._session().get(seg, headers={"Range": "bytes=0-65535"}, impersonate="chrome", timeout=10) kbps = len(r.content) / 1024 / max(time.time() - t, 0.01) except Exception: kbps = 0.0 now = "OK" if kbps >= 40 else "DEGRADED" if now != state: print(f"[probe] CDN {now} ({kbps:.0f} KB/s)", flush=True) state = now _CDN.update(state=now, kbps=round(kbps), ts=time.time()) _CDN["hist"].append({"t": time.time(), "kbps": round(kbps)}) del _CDN["hist"][:-20] time.sleep(180) threading.Thread(target=_cdn_probe, daemon=True).start() @app.get("/api/cdn") async def cdn_status(): """CDN health (read-only, public): latest probe + ~1h of history.""" return _CDN # ---------- auth: multi-user, signed cookies ---------- users.migrate_legacy() # file-based legacy import; db.init() absorbs its output db.init() _SIGNER = TimestampSigner(users.get_secret()) _SESSION = "aniweb_session" _SESSION_AGE = 30 * 86400 _PUBLIC = {("/login", "GET"), ("/login", "POST"), ("/api/cdn", "GET")} _SECURE = os.environ.get("ANIWEB_SECURE") == "1" # set when served over HTTPS # ponytail: in-memory lockout per username, single process — 5 fails -> 15 min _lockout: dict = {} def _session_user(request) -> str | None: try: payload = _SIGNER.unsign(request.cookies.get(_SESSION, ""), max_age=_SESSION_AGE) except BadSignature: return None return payload.decode().split(":", 1)[1] if payload.startswith(b"user:") else None def _lock(user: str, pw: str) -> str: lo = _lockout.setdefault(user, {"fails": 0, "until": 0.0}) if time.time() < lo["until"]: return "locked" if users.verify(user, pw): lo.update(fails=0, until=0.0) return "ok" lo["fails"] += 1 if lo["fails"] >= 5: lo.update(fails=0, until=time.time() + 900) time.sleep(0.5) # slow down brute force return "bad" @app.middleware("http") async def csrf_header(request, call_next): # ponytail: a cross-site form POST (top-level navigation) does carry the # SameSite=Lax cookie, so
from a # malicious page would work; forms can't set a custom header, the SPA # always sends it. Runs *after* require_auth (registered earlier), so # unauthenticated requests still get 302/401 first. if request.method in ("POST", "DELETE") and request.url.path != "/login": if request.headers.get("x-aniweb") != "1": return JSONResponse({"detail": "forbidden"}, status_code=403) return await call_next(request) @app.middleware("http") async def require_auth(request, call_next): if (request.url.path, request.method) in _PUBLIC: return await call_next(request) user = _session_user(request) # stateless cookie -> re-verify the account still exists, so # deleted users' cookies die immediately if user is not None and user not in users.load(): user = None if not user: if request.url.path.startswith("/api/"): return JSONResponse({"detail": "not authenticated"}, status_code=401) next_ = request.url.path + (f"?{request.url.query}" if request.url.query else "") return RedirectResponse(f"/login?next={quote(next_, safe='')}", status_code=302) request.state.user = user return await call_next(request) def _session_cookie(resp, user: str): resp.set_cookie(_SESSION, _SIGNER.sign(b"user:" + user.encode()).decode(), max_age=_SESSION_AGE, httponly=True, samesite="lax", secure=_SECURE) @app.get("/login", include_in_schema=False) def login_page(): return FileResponse(ROOT / "static" / "login.html") @app.post("/login") async def login(payload: dict = Body(...)): u = payload.get("username", "").lower() if not u: return JSONResponse({"detail": "username required"}, status_code=400) state = _lock(u, payload.get("password", "")) if state == "ok": resp = JSONResponse({"ok": True, "user": u}) _session_cookie(resp, u) return resp if state == "locked": return JSONResponse({"detail": "too many attempts, try later"}, status_code=429) return JSONResponse({"detail": "wrong username or password"}, status_code=401) @app.get("/api/me", include_in_schema=False) def api_me(request: Request): return {"user": request.state.user} @app.post("/account/delete") def delete_account(request: Request): users.remove(request.state.user) resp = JSONResponse({"ok": True}) resp.delete_cookie(_SESSION, secure=_SECURE) return resp @app.post("/logout") def logout(): resp = JSONResponse({"ok": True}) resp.delete_cookie(_SESSION, secure=_SECURE) return resp def _load_hist(user: str): c = db.connect() rows = c.execute("SELECT payload FROM history WHERE user=? ORDER BY t DESC", (user,)).fetchall() c.close() return [json.loads(r[0]) for r in rows] @app.get("/api/history") def api_history(request: Request): # no-store: this JSON mutates while a tab is open (position saves); # a stale copy has caused "poster gone" confusion before. # ?all=1: full list for the /history page (no top-5 cap) entries = sorted(_load_hist(request.state.user), key=lambda e: e.get("t", 0), reverse=True) if request.query_params.get("all") != "1": entries = entries[:5] return JSONResponse(entries, headers={"Cache-Control": "no-store"}) @app.delete("/api/history") def api_history_delete(request: Request, slug_id: str, ep_id: str): # ep_id is a stable INTEGER in the schema (ghost rows died in the import), # so a strict match suffices try: ep = int(ep_id) except ValueError: return {"removed": 0} c = db.connect() cur = c.execute("DELETE FROM history WHERE user=? AND slug_id=? AND ep_id=?", (request.state.user, slug_id, ep)) c.commit() c.close() return {"removed": cur.rowcount} @app.post("/api/history") def api_history_post(entry: dict = Body(...), request: Request = None): if "ep_id" not in entry: raise HTTPException(422, "ep_id required") # normalize: ghost rows were born when ep_id arrived as str against an # int row (or vice versa) — int() makes the stored type stable try: entry["ep_id"] = int(entry["ep_id"]) except (TypeError, ValueError): raise HTTPException(422, "ep_id must be an integer") entry["t"] = time.time() u = request.state.user c = db.connect() # upsert: the (user, slug_id, ep_id) PK *is* the merge key, so no # read-modify-write (and no lost-update race between the read and write) c.execute("INSERT INTO history (user, slug_id, ep_id, t, payload) VALUES (?,?,?,?,?) " "ON CONFLICT(user, slug_id, ep_id) DO UPDATE SET " "t=excluded.t, payload=excluded.payload", (u, str(entry.get("slug_id")), entry["ep_id"], entry["t"], json.dumps(entry))) # cap: newest 50 per user c.execute("DELETE FROM history WHERE user=? AND rowid NOT IN " "(SELECT rowid FROM history WHERE user=? ORDER BY t DESC LIMIT 50)", (u, u)) db.event(c, u, "position", entry) # metrics: watch-time trajectory c.commit() c.close() return entry def _load_searches(user: str): c = db.connect() rows = c.execute("SELECT q FROM searches WHERE user=? ORDER BY t DESC", (user,)).fetchall() c.close() return [r[0] for r in rows] @app.get("/api/searches") def api_searches(request: Request): return JSONResponse(_load_searches(request.state.user), headers={"Cache-Control": "no-store"}) @app.post("/api/searches") def api_searches_post(item: dict = Body(...), request: Request = None): q = str(item.get("q", "")).strip()[:60] if not q: raise HTTPException(422, "q required") u = request.state.user # NOCASE PK: re-posting a seen query (any casing) conflicts on the row, # so the upsert rewrites q to the newest casing, then caps at 8 c = db.connect() c.execute("INSERT INTO searches (user, q, t) VALUES (?,?,?) " "ON CONFLICT(user, q) DO UPDATE SET q=excluded.q, t=excluded.t", (u, q, time.time())) c.execute("DELETE FROM searches WHERE user=? AND rowid NOT IN " "(SELECT rowid FROM searches WHERE user=? ORDER BY t DESC LIMIT 8)", (u, u)) db.event(c, u, "search", {"q": q}) c.commit() c.close() 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), slug: str = Query(None)): u = request.state.user since = time.time() - days * 86400 c = db.connect() 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( "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] 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; a backward jump >60s within an ep counts as a rewatch eps = {} watch_sec = [0.0] * 7 # Mon..Sun by_day = {} # local YYYY-MM-DD -> seconds for ts, payload in positions: e = json.loads(payload) 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: d = p - s["pos"] s["secs"] += d s["eng" if e.get("lang") == "eng" else "jpn"] += d dt = datetime.fromtimestamp(ts) watch_sec[dt.weekday()] += d k = dt.strftime("%Y-%m-%d") by_day[k] = by_day.get(k, 0) + d 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, 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 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 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, "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 "by_day": {k: int(v) for k, v in by_day.items()}, # seconds per local day "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() def _url(token: str) -> str: return base64.urlsafe_b64decode(token.encode()).decode() def _upstream(fn, *a, **kw): try: return fn(*a, **kw) except HTTPException: raise except Exception as e: raise HTTPException(502, f"upstream error: {e}") @app.middleware("http") async def no_cache(request, call_next): # ponytail: no-store on everything except /stream (immutable HLS data, # cached per upstream URL) and the vendored/static JS+CSS (re-downloaded # on every page load hurt on mobile) r = await call_next(request) p = request.url.path if p == "/hls.min.js": r.headers["Cache-Control"] = "public, max-age=2592000" elif p in ("/app.js", "/style.css"): r.headers["Cache-Control"] = "public, max-age=300" elif not p.startswith("/stream"): r.headers["Cache-Control"] = "no-store" return r @app.get("/api/search") def api_search(q: str = Query(min_length=1)): return _upstream(upstream.search, q) @app.get("/api/anime/{slug_id}") def api_anime(slug_id: str): return _upstream(upstream.details, slug_id) @app.get("/api/intros/{mal_id}") def api_intros(mal_id: int): # keyless Jikan/MAL intro timings; Jikan down or no data = no skip # button, never an error that could touch the player try: return upstream.intros(mal_id) except Exception: return [] @app.get("/api/anime/{anime_id}/episodes") def api_episodes(anime_id: int): return _upstream(upstream.episodes, anime_id) @app.get("/api/episode/{ep_id}/stream") 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)} try: 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() 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: # ponytail: rewrites plain URIs only; URI-list attributes in # #EXT-X-* tags (e.g. SUBTITLES) are ignored — fine for video-only playlists return "\n".join( line if line.startswith("#") or not line else "/stream/" + _b64(urljoin(base, line)) for line in text.splitlines()) + "\n" _STREAM_CACHE = OrderedDict() # ponytail: LRU of proxied HLS data (immutable); _STREAM_MAX = 24 # segments ~1-2MB -> <= ~48MB # ponytail: HLS VOD content is immutable and the token IS the exact upstream # URL, so per-URL browser caching can't go stale — re-opens, re-seeks and # auto-next come from the disk cache instead of proxy + CDN _STREAM_CACHE_HDRS = {"Cache-Control": "public, max-age=86400"} def _cache_put(url, ct, body): _STREAM_CACHE[url] = (ct, body) while len(_STREAM_CACHE) > _STREAM_MAX: _STREAM_CACHE.popitem(last=False) def _seg_stream(r, url, ct): """Forward the CDN's bytes as they arrive; keep the LRU warm for completed downloads (a truncated one would poison the cache).""" import time t0, t_first, buf, complete = time.time(), None, [], False try: for c in r.iter_content(chunk_size=256 * 1024): if t_first is None: t_first = time.time() buf.append(c) yield c complete = True finally: r.close() if complete and buf: _cache_put(url, ct, b"".join(buf)) # observability: a slow segment is the fragLoadTimeOut culprit. Log # only the slow ones so normal playback stays quiet. TTFB>3s or # total>8s means the CDN (or our upstream hop) stalled — that's what # the browser times out on. If these show ~0s but the user still # fails, the stall is in *their* path (HTTPS proxy / network), not ours. ttfb = (t_first - t0) if t_first is not None else None total = time.time() - t0 if (ttfb is not None and ttfb > 3) or total > 8: name = url.rsplit("/", 1)[-1][:40] print(f"[stream-slow] {name} ttfb={ttfb:.1f}s total={total:.1f}s " f"bytes={sum(len(b) for b in buf)} complete={complete}", flush=True) def _stream_playlist(url): # buffered: the full body is needed to rewrite URIs through the proxy # ponytail: 12s < hls.js's fragLoadingTimeOut (20s) — on a hang the proxy # 502s first, so hls.js retries the frag (backoff) instead of freezing # on a fatal timeout race r = _upstream(upstream._get, url, timeout=12) body = r.content if "mpegurl" in r.headers.get("content-type", ""): body = _rewrite_m3u8(body.decode(), url).encode() ct = r.headers.get("content-type", "application/octet-stream") _cache_put(url, ct, body) return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS) @app.get("/stream/{token}") def stream(token: str): # ponytail: token is plain base64, so allowlist the host — otherwise any # logged-in user could use this as a GET proxy to arbitrary URLs (SSRF). # Verified 2026-08: master, variant and segment URLs all live on # hls.anidb.app; if the CDN moves, this 400s loudly and the list updates. url = _url(token) p = urlparse(url) if p.scheme != "https" or p.netloc != "hls.anidb.app": raise HTTPException(400, "bad stream token") # cache hit (a playlist, or a segment downloaded earlier): serve at once hit = _STREAM_CACHE.get(url) if hit: _STREAM_CACHE.move_to_end(url) return Response(hit[1], media_type=hit[0], headers=_STREAM_CACHE_HDRS) if url.endswith(".m3u8"): return _stream_playlist(url) # segment miss: try parallel byte-range lanes first (4 concurrent fetches # beat per-connection CDN throttling); ANY failure falls back to the # single streaming path, so this can never be worse than before t0 = time.time() try: body, ct = _upstream(upstream.fetch_parallel, url) except HTTPException: r = _upstream(upstream.get_stream, url, timeout=12) ct = r.headers.get("content-type", "application/octet-stream") if time.time() - t0 > 8: name = url.rsplit("/", 1)[-1][:40] print(f"[stream-slow] {name} parallel-fallback elapsed={time.time()-t0:.1f}s", flush=True) return StreamingResponse(_seg_stream(r, url, ct), media_type=ct, headers=_STREAM_CACHE_HDRS) if time.time() - t0 > 8: name = url.rsplit("/", 1)[-1][:40] print(f"[stream-slow] {name} parallel total={time.time()-t0:.1f}s " f"bytes={len(body)}", flush=True) _cache_put(url, ct, body) return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS) app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")