ani-web/server/upstream.py

"""upstream.py — the only module that talks to anidb.app (and keyless Jikan
for intro timings).

anidb.app sits behind Cloudflare: plain curl gets 403. All requests use
curl_cffi Chrome TLS impersonation (verified to pass the challenge).
Parsers are regex against live markup — keep them here so the layout
breaks in one place, not across the app.
"""
import re
import sys
import threading
import time

from curl_cffi import requests

BASE = "https://anidb.app"
H = dict(impersonate="chrome", timeout=20)

# ponytail: thread-local sessions (uvicorn threadpool); curl_cffi Sessions
# are not thread-safe, but per-thread keep-alive still reuses TCP/TLS across
# the many requests a worker handles. Saves a handshake on every call.
_tls = threading.local()

def _session():
    if not hasattr(_tls, "s"):
        _tls.s = requests.Session()
    return _tls.s

# ponytail: in-memory TTL cache for upstream data that never changes
# (details/episodes/langs/playlist URLs). No lock: worst case a duplicate
# fetch. A dead/rotated playlist URL is recovered by the one-shot
# clear_cache()+retry in api_stream.
_CACHE = {}  # key -> (expires, value)

def _cached(key, ttl, fn, *a, **kw):
    v = _CACHE.get(key)
    if v and v[0] > time.time():
        return v[1]
    val = fn(*a, **kw)
    _CACHE[key] = (time.time() + ttl, val)
    if len(_CACHE) > 256:
        _CACHE.clear()
    return val

def clear_cache():
    _CACHE.clear()

_ANIME_LINK = re.compile(r'anime/([a-z0-9-]+)-(\d+)"[^>]*title="([^"]*)"')


def _get(url, **kw):
    # ponytail: one retry (300ms) on transient upstream errors (Cloudflare
    # 403/503, 429, 5xx); 404 and other 4xx are final. Failing fast matters:
    # hls.js's fragLoadingTimeOut (20s) races curl's timeout (20s), so a
    # quick 502 lets hls.js's own frag retries (6x, backoff) do the waiting
    for attempt in range(2):
        r = _session().get(url, **{**H, **kw})  # kw overrides H (e.g. timeout)
        if not (r.status_code in (403, 429, 503) or r.status_code >= 500
                or "Just a moment" in r.text[:500]):
            r.raise_for_status()
            return r
        if attempt:
            raise RuntimeError(f"upstream {r.status_code}: {url}")
        time.sleep(0.3)


def _search(q):
    """Step 1: browse page -> [{id, slug_id, name, poster}]"""
    html = _get(f"{BASE}/browse", params={"q": q}).text
    out, seen = [], set()
    for m in re.finditer(
        r'<a href="https://anidb\.app/anime/([a-z0-9-]+-\d+)"[^>]*title="([^"]*)"[^>]*>(.*?)</a>',
        html, re.S):
        slug_id, name, block = m.group(1), m.group(2), m.group(3)
        if slug_id in seen:
            continue
        seen.add(slug_id)
        img = re.search(r'<img src="([^"]+)"', block)
        out.append({"id": int(slug_id.rsplit("-", 1)[1]), "slug_id": slug_id,
                    "name": name, "poster": img.group(1) if img else None})
    return out


def _details(slug_id):
    """Step 2: details page -> {name, synopsis, mal_id, poster, seasons}"""
    p = _get(f"{BASE}/anime/{slug_id}").text
    mal = re.search(r'myanimelist\.net/anime/(\d+)', p)
    syn = re.search(r'>Synopsis</h2>.*?<p[^>]*>(.*?)</p>', p, re.S)
    h1 = re.search(r'<h1[^>]*>(.*?)</h1>', p, re.S)
    post = re.search(r'src="(https://cdn\.xlsbox\.com/poster/[^"\']+)"', p)
    seasons, seen = [], set()
    i = p.find(">Seasons<")
    if i > 0:
        cnt = re.search(r">(\d+) entries in this series<", p[i:i + 400])
        n = int(cnt.group(1)) if cnt else 10  # fall back to cap if count markup changes
        for m in _ANIME_LINK.finditer(p[i:i + 20000]):
            sid = f"{m.group(1)}-{m.group(2)}"
            if sid in seen:
                continue
            seen.add(sid)
            seasons.append({"slug_id": sid, "name": m.group(3), "id": int(m.group(2))})
            if len(seasons) == n:
                break
    return {"id": int(slug_id.rsplit("-", 1)[1]),
            "name": h1.group(1).strip() if h1 else None,
            "synopsis": re.sub(r"<[^>]+>", "", syn.group(1)).strip() if syn else None,
            "mal_id": int(mal.group(1)) if mal else None,
            "poster": post.group(1) if post else None,
            "seasons": seasons}


def _episodes(anime_id):
    """Step 3: JSON API -> [{id, number, filler}]"""
    d = _get(f"{BASE}/api/frontend/anime/{anime_id}/episodes").json()
    return [{"id": e["id"], "number": e["number"], "filler": bool(e.get("filler"))}
            for e in d["episodes"]]


def _languages(ep_id):
    """Step 4: JSON API -> [{code, name, embed_url}] (jpn=sub, eng=dub)"""
    return _get(f"{BASE}/api/frontend/episode/{ep_id}/languages").json()["languages"]


def _parse_intros(eps):
    """Jikan episode list -> [{start, length}, ...], None where MAL has no timing"""
    out = []
    for e in eps:
        i = e.get("intro") or {}
        out.append({"start": int(i["start"]), "length": int(i["length"])}
                   if i.get("start") is not None else None)
    return out


def _intros(mal_id):
    """Jikan (keyless MAL mirror): per-episode intro start/length, seconds"""
    r = _get("https://api.jikan.moe/v4/anime/{}".format(mal_id) + "/episodes")
    return _parse_intros(r.json().get("data", {}).get("episodes", []))


def _master_m3u8(ep_id, lang="eng"):
    """Steps 5+6: embed page -> 'file: <master m3u8>' URL (None if unavailable)"""
    for l in languages(ep_id):
        if l["code"] == lang:
            m = re.search(r"file: '([^']*)'", _get(l["embed_url"]).text)
            if m:
                return m.group(1)
    return None


# cached public API (TTLs in seconds)
search = lambda q: _cached(f"search:{q.lower()}", 300, _search, q)
details = lambda s: _cached(f"details:{s}", 3600, _details, s)
episodes = lambda i: _cached(f"episodes:{i}", 300, _episodes, i)  # 5m: airing shows gain eps
languages = lambda e: _cached(f"languages:{e}", 3600, _languages, e)
master_m3u8 = lambda e, l="eng": _cached(f"master:{e}:{l}", 3600, _master_m3u8, e, l)
intros = lambda m: _cached(f"intros:{m}", 30 * 86400, _intros, m)  # static data


if __name__ == "__main__":
    # ponytail: assert self-check instead of a test framework
    results = search("chainsaw man")
    assert results and any(x["name"] == "Chainsaw Man" for x in results), results
    top = next(x for x in results if x["name"] == "Chainsaw Man")
    d = details(top["slug_id"])
    assert d["mal_id"] == 44511 and d["synopsis"], d
    assert [s["id"] for s in d["seasons"]] == [922, 923], d["seasons"]
    eps = episodes(top["id"])
    assert eps and eps[0]["number"] == 1, eps
    m = master_m3u8(eps[0]["id"], "eng")
    assert m and "EXT-X-STREAM" in _get(m).text, m
    print(f"OK: {top['name']} (MAL {d['mal_id']}, {len(d['seasons'])} seasons, "
          f"{len(eps)} eps) -> ep1 dub: {m[:70]}...")
    sys.exit(0)