jozamudi
731f44b99c
server/app.py DONE_RATIO is injected into index.html like {{VERSION}}, app.js
reads it as window.DONE_RATIO (fallback = same default). Change the threshold
in exactly one place now.
718 lines
29 KiB
Python
718 lines
29 KiB
Python
"""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 subprocess
|
|
import threading
|
|
import time
|
|
from math import isfinite
|
|
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, HTMLResponse, JSONResponse, RedirectResponse, Response, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from itsdangerous import BadSignature, TimestampSigner
|
|
|
|
from . import auth, db, upstream, users
|
|
|
|
ROOT = Path(__file__).parent.parent
|
|
app = FastAPI(title="ani-web")
|
|
|
|
def _version():
|
|
"""Cache-bust token: git short SHA (+ -dirty). The content of a commit
|
|
never changes, so ?v=<sha> is a correct cache key — this replaces the
|
|
hand-bumped ?v= number (forgetting a bump = users stuck on stale JS).
|
|
Every deploy restarts the process (run.sh), so the read is fresh."""
|
|
try:
|
|
v = subprocess.run(["git", "describe", "--always", "--dirty"], cwd=ROOT,
|
|
capture_output=True, text=True, timeout=5).stdout.strip()
|
|
if v:
|
|
return v
|
|
except Exception:
|
|
pass
|
|
return "dev"
|
|
|
|
|
|
VERSION = _version()
|
|
|
|
# An episode counts as "watched" once you're this far through it (fraction of
|
|
# duration). One knob for client + server: injected into index.html below so
|
|
# app.js uses the same value as these stats. Change here only.
|
|
DONE_RATIO = 0.95
|
|
|
|
# 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": []}
|
|
# origin (primary provider pages/API) health — separate host, separate failure
|
|
# mode (observed 2026-09-03: CDN fine while origin 503'd "Under Maintenance")
|
|
_ORIGIN = {"state": "unknown", "msg": "", "ts": 0.0, "hist": []}
|
|
# ponytail: fixed threshold; if the CDN's normal throughput drifts below it
|
|
# the pill sits on "slow" — tune via ANIWEB_CDN_KBPS, not a percentile.
|
|
_CDN_THRESHOLD = float(os.environ.get("ANIWEB_CDN_KBPS", "40"))
|
|
|
|
|
|
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: ep_id 1 = One Piece E1 on the primary provider.
|
|
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(1, "jpn")
|
|
if not m:
|
|
raise RuntimeError("no master m3u8")
|
|
# hianime's CDN 403s without the embed-origin Referer (anidb needs none)
|
|
hdrs = upstream.stream_meta(m) or {}
|
|
base = m.rsplit("/", 1)[0] + "/"
|
|
v = first(upstream._get(m, timeout=10, headers=hdrs).text, base)
|
|
# hianime's variant playlists use URIs relative to THEIR OWN path
|
|
vbase = v.rsplit("/", 1)[0] + "/"
|
|
seg = first(upstream._get(v, timeout=10, headers=hdrs).text, vbase)
|
|
t = time.time()
|
|
r = upstream._session().get(seg, headers={**hdrs, "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 >= _CDN_THRESHOLD 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]
|
|
_origin_probe()
|
|
time.sleep(180)
|
|
|
|
|
|
def _origin_probe():
|
|
"""Hit the same kind of page users load (details page) every 3 min and
|
|
track OK/DOWN — this is the host that 503s 'Under Maintenance'.
|
|
ponytail: one URL, no history; the log line + ts tell the when."""
|
|
try:
|
|
r = upstream._get("https://hianime.at/watch/one-piece-1", timeout=10)
|
|
state, msg = "OK", f"http {r.status_code}"
|
|
except Exception as e:
|
|
state, msg = "DOWN", str(e)
|
|
prev = _ORIGIN["state"]
|
|
_ORIGIN.update(state=state, msg=msg, ts=time.time())
|
|
_ORIGIN["hist"].append({"t": time.time(), "state": state})
|
|
del _ORIGIN["hist"][:-20]
|
|
if state != prev:
|
|
print(f"[probe] ORIGIN {state} ({msg})", flush=True)
|
|
|
|
|
|
threading.Thread(target=_cdn_probe, daemon=True).start()
|
|
|
|
|
|
@app.get("/api/cdn")
|
|
async def cdn_status():
|
|
"""CDN + origin health (read-only, public). Top-level fields are the CDN
|
|
probe (existing shape); .origin is the anidb.app pages/API probe."""
|
|
return {**_CDN, "threshold": _CDN_THRESHOLD, "origin": _ORIGIN}
|
|
|
|
# ---------- 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
|
|
|
|
# Login throttling lives in server/auth.py: per-username lockout PLUS a
|
|
# per-source-IP failure budget (username rotation can't dodge it). The
|
|
# bucket logic is unit-tested there (tests/test_auth.py), without the app.
|
|
|
|
|
|
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
|
|
|
|
|
|
# (throttle buckets + client_ip live in server/auth.py — see auth.locked /
|
|
# auth.record; the old per-username-only _lock dict was the audit finding)
|
|
|
|
|
|
@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 <form action=.../account/delete> from a
|
|
# malicious page would work; forms can't set a custom header, the SPA and
|
|
# the login page both send one. /login is covered too: login needs no
|
|
# cookie, so an exempt /login let a cross-site forced-login POST swap a
|
|
# visitor's session onto the attacker's account. Runs *after* require_auth
|
|
# (registered earlier), so unauthenticated GETs still 302/401 first.
|
|
if request.method in ("POST", "DELETE"):
|
|
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")
|
|
def login(request: Request, payload: dict = Body(...)):
|
|
# SYNC def on purpose: the 0.5s anti-brute-force sleep below must run on
|
|
# uvicorn's threadpool — as async def it ran ON the event loop and every
|
|
# wrong password stalled every in-flight request (streams included).
|
|
u = payload.get("username", "").lower()
|
|
if not u:
|
|
return JSONResponse({"detail": "username required"}, status_code=400)
|
|
ip = auth.client_ip(request)
|
|
if auth.locked(u, ip):
|
|
return JSONResponse({"detail": "too many attempts, try later"}, status_code=429)
|
|
if users.verify(u, payload.get("password", "")):
|
|
auth.clear(u)
|
|
resp = JSONResponse({"ok": True, "user": u})
|
|
_session_cookie(resp, u)
|
|
return resp
|
|
auth.record(u, ip)
|
|
time.sleep(0.5) # slow down brute force
|
|
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")
|
|
slug = str(entry.get("slug_id") or "").strip()
|
|
if not slug:
|
|
raise HTTPException(422, "slug_id required") # a missing slug used to store a "None" row
|
|
entry["slug_id"] = slug
|
|
if entry.get("pos") is not None:
|
|
try:
|
|
pos = max(0.0, float(entry["pos"]))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "pos must be a number")
|
|
if not isfinite(pos):
|
|
raise HTTPException(422, "pos not finite")
|
|
if entry.get("dur"): # clamp above-duration junk out of the progress UI
|
|
try:
|
|
pos = min(pos, float(entry["dur"]))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
entry["pos"] = pos
|
|
# a missing/null pos stays null: the upsert keeps the existing row's
|
|
# payload, so a stale client can never clobber the saved position
|
|
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=CASE WHEN json_extract(excluded.payload, '$.pos') IS NULL "
|
|
"THEN history.payload ELSE excluded.payload END",
|
|
(u, entry["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; the diagnostics the browser can uniquely see
|
|
# (playback failures, capability probes) happen client-side, not in the
|
|
# request log — so the phone reports them here
|
|
if evt.get("type") not in ("error", "client-fail", "play-probe", "hls-err"):
|
|
raise HTTPException(422, "unknown event type")
|
|
c = db.connect()
|
|
db.event(c, request.state.user, evt["type"],
|
|
{k: evt[k] for k in ("ep_id", "ep", "lang", "msg", "detail",
|
|
"jsv", "branch", "hls_lib", "mse", "native_hls", "ua",
|
|
"step", "fatal", "details", "buffer", "retry", "http",
|
|
"dt", "n", "rs", "ns", "paused", "bufEnd", "ct",
|
|
"hls_state", "ov", "name") if k in evt})
|
|
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"] * DONE_RATIO:
|
|
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}")
|
|
out = {"token": _b64(master)}
|
|
sub = upstream.subtitles(ep_id) # external .vtt (hianime); anidb bakes subs in
|
|
if sub:
|
|
out["sub"] = _b64(sub)
|
|
return out
|
|
|
|
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, meta=None):
|
|
# 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, **({"headers": meta} if meta else {}))
|
|
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):
|
|
url = _url(token)
|
|
p = urlparse(url)
|
|
# 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).
|
|
# anidb's CDN is static (hls.anidb.app); hianime's CDN rotates, so its
|
|
# provider self-registers resolved hosts and supplies the Referer the
|
|
# CDN demands — stream_meta() non-None IS the allowlist entry.
|
|
meta = upstream.stream_meta(url) if p.scheme == "https" else None
|
|
if p.scheme != "https" or (p.netloc != "hls.anidb.app" and not meta):
|
|
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, meta)
|
|
# 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,
|
|
**({"headers": meta} if meta else {}))
|
|
except HTTPException:
|
|
r = _upstream(upstream.get_stream, url, timeout=12,
|
|
**({"headers": meta} if meta else {}))
|
|
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.get("/", include_in_schema=False)
|
|
@app.get("/index.html", include_in_schema=False)
|
|
def index():
|
|
# index.html is served (not mounted) so ?v={{VERSION}} becomes the git
|
|
# SHA and {{DONE_RATIO}} the watched threshold at startup; other statics
|
|
# go through the mount below
|
|
return HTMLResponse(_INDEX_HTML.replace("{{VERSION}}", VERSION)
|
|
.replace("{{DONE_RATIO}}", repr(DONE_RATIO)))
|
|
|
|
|
|
_INDEX_HTML = (ROOT / "static" / "index.html").read_text()
|
|
|
|
|
|
app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")
|