"""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 get_stream(url, timeout=12):
"""Segments: open the upstream response WITHOUT reading the body, so the
proxy can forward bytes as the CDN emits them. (A buffered fetch made the
browser's time-to-first-byte = the CDN's whole upload time, which tripped
hls.js's fragLoadTimeOut on slow CDN moments.)
ponytail: skips the 'Just a moment' body check — challenge pages arrive
as 403/503 (handled below), and reading the body would defeat streaming."""
for attempt in range(2):
r = _session().get(url, **{**H, "timeout": timeout, "stream": True})
if r.status_code in (403, 429, 503) or r.status_code >= 500:
r.close()
if attempt:
raise RuntimeError(f"upstream {r.status_code}: {url}")
time.sleep(0.3)
continue
r.raise_for_status()
return r
def fetch_parallel(url, lanes=16, lane=32 * 1024, timeout=12):
"""One upstream file across N parallel byte-range lanes (16, like yt-dlp's
-N 16 / ani-cli's download path).
Per-connection CDN throttling is the common slow-segment cause; N lanes
turn one 60s crawl into N concurrent crawls (I/O-bound, GIL irrelevant).
Returns (body, content-type); raises on ANY failure so the caller falls
back to the single-request path — never worse than before.
Verified 2026-08 against hls.anidb.app: HEAD -> content-length,
ranges -> 206, multi-lane -> byte-identical stitch (sha256)."""
from concurrent.futures import ThreadPoolExecutor
head = _session().head(url, **{**H, "timeout": timeout})
head.raise_for_status()
total = int(head.headers.get("content-length") or 0)
ct = head.headers.get("content-type", "application/octet-stream")
if total == 0:
raise RuntimeError("no content-length")
if total <= 2 * lane: # too small to be worth N connections
r = _get(url, timeout=timeout)
return r.content, ct
n = min(lanes, total // lane)
size = (total + n - 1) // n
parts = [None] * n
def grab(i):
a = i * size
b = total - 1 if i == n - 1 else (i + 1) * size - 1
r = _get(url, headers={"Range": f"bytes={a}-{b}"}, timeout=timeout)
if r.status_code not in (200, 206) or len(r.content) != b - a + 1:
raise RuntimeError(f"lane {i}: http {r.status_code} "
f"len={len(r.content)} want={b - a + 1}")
parts[i] = r.content
with ThreadPoolExecutor(max_workers=n) as ex:
list(ex.map(grab, range(n)))
body = b"".join(parts)
if len(body) != total:
raise RuntimeError(f"stitch mismatch {len(body)} != {total}")
return body, ct
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']*title="([^"]*)"[^>]*>(.*?)',
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' {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.*?
]*>(.*?)
', p, re.S) h1 = re.search(r'