"""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 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
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")

# ---------- 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")}
_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 <form action=.../account/delete> 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) and the vendored hls.min.js (immutable, 500KB — re-downloading
    # it on every page load hurt on mobile)
    r = await call_next(request)
    if request.url.path == "/hls.min.js":
        r.headers["Cache-Control"] = "public, max-age=2592000"
    elif not request.url.path.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/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),
                "qualities": _upstream(upstream.master_variants, 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


@app.get("/stream/{token}")
def stream(token: str):
    # ponytail: buffers whole response in memory; .ts segments are small
    url = _url(token)
    # 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.
    p = urlparse(url)
    if p.scheme != "https" or p.netloc != "hls.anidb.app":
        raise HTTPException(400, "bad stream token")
    hit = _STREAM_CACHE.get(url)
    if hit:
        _STREAM_CACHE.move_to_end(url)
        return Response(hit[1], media_type=hit[0])
    r = _upstream(upstream._get, url)
    body = r.content
    if url.endswith(".m3u8") or "mpegurl" in r.headers.get("content-type", ""):
        body = _rewrite_m3u8(body.decode(), url).encode()
    _STREAM_CACHE[url] = (r.headers.get("content-type", "application/octet-stream"), body)
    while len(_STREAM_CACHE) > _STREAM_MAX:
        _STREAM_CACHE.popitem(last=False)
    return Response(body, media_type=r.headers.get("content-type", "application/octet-stream"))


app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")