Provider layer + hianime source (primary)

- providers/{anidb,anikoto,hianime}.py: per-source modules behind the
  ANI_SOURCE fallback dispatch (upstream.py _call/_providers)
- hianime.py: pystardust/ani-cli#1897 flow — search/episode-list/
  servers API, HD-1 hash -> base64 embed url, window.__P blob
  (base64 + XOR 'otaku-embed-v1') -> master m3u8; registers CDN hosts
  and the required Referer via stream_meta()
- upstream.py: get_stream/fetch_parallel accept extra headers;
  default source order hianime,anidb,anikoto (anidb.app is 503)
- app.py: /stream allowlist accepts provider-registered hosts and
  sends their Referer; status probe follows the primary provider
  (hianime variant URIs are relative to the variant path)
- app.js: error-time position save floors at furthest point reached
  (hls.js resets currentTime to 0 before playerFail on fatal errors)
- tests: default-order + hianime import checks; live browser suite green

server/app.py

@@ -45,15 +45,15 @@ VERSION = _version()
 
 # 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 (anidb.app pages/API) health — separate host, separate failure mode
-# (observed 2026-09-03: CDN fine while origin 503'd "Under Maintenance")
+# 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}
 
 
 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).
+    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):
@@ -64,14 +64,18 @@ def _cdn_probe():
     state = None
     while True:
         try:
-            m = upstream.master_m3u8(36019, "jpn")
+            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).text, base)
-            seg = first(upstream._get(v, timeout=10).text, base)
+            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={"Range": "bytes=0-65535"},
+            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:
@@ -92,8 +96,7 @@ def _origin_probe():
     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://anidb.app/anime/jujutsu-kaisen-season-2-2554",
-                          timeout=10)
+        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)
@@ -606,12 +609,12 @@ def _seg_stream(r, url, ct):
                   f"bytes={sum(len(b) for b in buf)} complete={complete}", flush=True)
 
 
-def _stream_playlist(url):
+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)
+    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()
@@ -622,13 +625,15 @@ def _stream_playlist(url):
 
 @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":
+    # 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)
@@ -636,15 +641,17 @@ def stream(token: str):
         _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)
+        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)
+        body, ct = _upstream(upstream.fetch_parallel, url,
+                             **({"headers": meta} if meta else {}))
     except HTTPException:
-        r = _upstream(upstream.get_stream, url, timeout=12)
+        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]

server/providers/__init__.py

@@ -0,0 +1 @@
+"""providers/ — anime source providers. See upstream.py for the interface."""

server/providers/anidb.py

@@ -0,0 +1,89 @@
+"""providers/anidb.py — anidb.app source (primary).
+
+Sits behind Cloudflare: plain curl gets 403. All requests go through the
+shared `_get` (curl_cffi Chrome impersonation). Parsers are regex against
+live markup — they break here, in one place, when the layout changes.
+"""
+import re
+
+from ..upstream import _cached, _get
+
+BASE = "https://anidb.app"
+
+_ANIME_LINK = re.compile(r'anime/([a-z0-9-]+)-(\d+)"[^>]*title="([^"]*)"')
+
+
+def _search(q):
+    """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):
+    """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):
+    """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):
+    """JSON API -> [{code, name, embed_url}] (jpn=sub, eng=dub)"""
+    return _get(f"{BASE}/api/frontend/episode/{ep_id}/languages").json()["languages"]
+
+
+def _master_m3u8(ep_id, lang="eng"):
+    """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
+
+
+# Provider interface. Cache keys are namespaced with "anidb:" so a shared
+# cache dict can't leak another provider's value into this one.
+search = lambda q: _cached(f"anidb:search:{q.lower()}", 300, _search, q)
+details = lambda s: _cached(f"anidb:details:{s}", 3600, _details, s)
+episodes = lambda i: _cached(f"anidb:episodes:{i}", 300, _episodes, i)  # 5m: airing shows gain eps
+languages = lambda e: _cached(f"anidb:languages:{e}", 3600, _languages, e)
+master_m3u8 = lambda e, l="eng": _cached(f"anidb:master:{e}:{l}", 3600, _master_m3u8, e, l)

server/providers/anikoto.py

@@ -0,0 +1,137 @@
+"""providers/anikoto.py — anikototv.to (AniKoto / Hianime) fallback source.
+
+Flow: search page -> watch page (numeric anime id) -> ajax episode list
+(per-episode token) -> ajax server list (sub/dub groups) -> ajax/server?get=
+(MegaPlay embed page) -> getSourcesNew?id= (master m3u8).
+
+MegaPlay's CDN domains rotate per getSources call and some (e.g. norami.top)
+hard-403 non-browser clients, so _master_m3u8 re-rolls the playlist a few
+times until a CDN domain actually serves us.
+"""
+import re
+
+from ..upstream import _cached, _get
+
+BASE = "https://anikototv.to"
+XHR = {"X-Requested-With": "XMLHttpRequest"}
+_MEGAPLAY_ID = re.compile(r'data-id="(\d+)"')
+
+
+def _search(q):
+    html = _get(f"{BASE}/search", params={"keyword": q}).text
+    out, seen = [], set()
+    for m in re.finditer(
+            r'href="https://anikototv\.to/watch/([a-z0-9-]+)[^"]*"[^>]*>\s*([^<]{2,80})</a>',
+            html):
+        slug, name = m.group(1), m.group(2).strip()
+        if slug in seen:
+            continue
+        seen.add(slug)
+        imgs = re.findall(r'<img src="(https://cdn\.anipixcdn\.co/[^"]+)"',
+                          html[max(0, m.start() - 600):m.start()])
+        out.append({"id": 0, "slug_id": slug, "name": name,
+                    "poster": imgs[-1] if imgs else None})
+    return out
+
+
+def _details(slug):
+    t = _get(f"{BASE}/watch/{slug}").text
+    h1 = re.search(r'<h1[^>]*>(.*?)</h1>', t, re.S)
+    syn = re.search(r'class="synopsis[^"]*"[\s\S]*?class="content"[\s\S]*?>(.*?)</div>', t)
+    post = re.search(r'(https://cdn\.anipixcdn\.co/thumbnail/[^"\']+)', t)
+    aid = re.search(r'data-anime-id="(\d+)"', t)
+    return {"id": int(aid.group(1)) if aid else 0,
+            "name": re.sub(r"<[^>]+>", "", h1.group(1)).strip() if h1 else None,
+            "synopsis": re.sub(r"<[^>]+>", "", syn.group(1)).strip() if syn else None,
+            "mal_id": None,
+            "poster": post.group(1) if post else None,
+            "seasons": []}
+
+
+def _ep_entries(anime_id):
+    d = _get(f"{BASE}/ajax/episode/list/{anime_id}", headers=XHR).json()
+    out = []
+    for tag in re.findall(r'<a[^>]+>', d.get("result") or ""):
+        num = re.search(r'data-num="(\d+)"', tag)
+        tok = re.search(r'data-ids="([^"]+)"', tag)
+        if num and tok:
+            out.append({"num": int(num.group(1)), "token": tok.group(1),
+                        "sub": 'data-sub="1"' in tag, "dub": 'data-dub="1"' in tag})
+    return out
+
+
+def _ep_entry(ep_id):
+    # ep_id encodes (anime_id, num): anime_id * 1000 + num (num < 1000)
+    anime_id, num = divmod(ep_id, 1000)
+    return next((e for e in _ep_entries(anime_id) if e["num"] == num), None)
+
+
+def _episodes(anime_id):
+    return [{"id": anime_id * 1000 + e["num"], "number": e["num"], "filler": False}
+            for e in _ep_entries(anime_id)]
+
+
+def _languages(ep_id):
+    e = _ep_entry(ep_id)
+    if not e:
+        return []
+    out = []
+    if e["sub"]:
+        out.append({"code": "jpn", "name": "Sub", "embed_url": "sub"})
+    if e["dub"]:
+        out.append({"code": "eng", "name": "Dub", "embed_url": "dub"})
+    return out
+
+
+def _server_link_ids(token, want):
+    d = _get(f"{BASE}/ajax/server/list", params={"servers": token}, headers=XHR).json()
+    res = d.get("result") or ""
+    for m in re.finditer(r'data-type="(\w+)"[\s\S]*?<ul>([\s\S]*?)</ul>', res):
+        if m.group(1) == want:
+            return re.findall(r'data-link-id="([^"]+)"', m.group(2))
+    return re.findall(r'data-link-id="([^"]+)"', res)  # no group split: take all
+
+
+def _player_file(embed_url):
+    """MegaPlay AND its re-skins (vidtube etc.): same player libs, so the
+    same getSourcesNew endpoint on the embed's own host."""
+    page = _get(embed_url, timeout=15)
+    m = _MEGAPLAY_ID.search(page.text)
+    if not m:
+        return None
+    host = re.match(r'https?://([^/]+)/', embed_url).group(1)
+    d = _get(f"https://{host}/stream/getSourcesNew?id={m.group(1)}",
+             headers={"Referer": embed_url, "X-Requested-With": "XMLHttpRequest"}).json()
+    return (d.get("sources") or {}).get("file")
+
+
+def _master_m3u8(ep_id, lang="eng"):
+    e = _ep_entry(ep_id)
+    if not e:
+        return None
+    want = "sub" if lang in ("jpn", "sub") else "dub"
+    ids = _server_link_ids(e["token"], want) or _server_link_ids(e["token"], "sub")
+    errs = []
+    for lid in ids[:3]:
+        d = _get(f"{BASE}/ajax/server", params={"get": lid}, headers=XHR).json()
+        url = (d.get("result") or {}).get("url")
+        if not url:
+            continue
+        for _ in range(2):  # re-roll: CDN domains rotate, most 403 our IP
+            try:
+                f = _player_file(url)
+                if not f:
+                    break
+                _get(f, timeout=10)  # prove this CDN serves us before committing
+                return f
+            except Exception as ex:
+                errs.append(str(ex)[:60])
+    raise RuntimeError("anikoto: no working server: " + "; ".join(errs[:3]))
+
+
+# Provider interface (cache keys namespaced "anikoto:").
+search = lambda q: _cached(f"anikoto:search:{q.lower()}", 300, _search, q)
+details = lambda s: _cached(f"anikoto:details:{s}", 3600, _details, s)
+episodes = lambda i: _cached(f"anikoto:episodes:{i}", 300, _episodes, i)
+languages = lambda e: _cached(f"anikoto:languages:{e}", 3600, _languages, e)
+master_m3u8 = lambda e, l="eng": _cached(f"anikoto:master:{e}:{l}", 3600, _master_m3u8, e, l)

server/providers/hianime.py

@@ -0,0 +1,126 @@
+"""providers/hianime.py — hianime.at source (pystardust/ani-cli#1897 flow).
+
+Flow: search page -> ajax episode list (numeric anime id) -> ajax server list
+(sub/dub groups) -> HD-1 hash (base64 embed url) -> embed page
+(window.__P = base64(json XOR "otaku-embed-v1")) -> master m3u8.
+
+The CDN hard-403s without the embed origin as Referer, so _master_m3u8
+records it in REFERER and registers the playlist hosts in STREAM_HOSTS;
+the /stream proxy reads both via stream_meta().
+"""
+import base64
+import json
+import re
+from urllib.parse import urljoin, urlparse
+
+from ..upstream import _cached, _get
+
+BASE = "https://hianime.at"
+KEY = b"otaku-embed-v1"
+
+REFERER = None          # embed origin, set when a master playlist is resolved
+STREAM_HOSTS = set()    # playlist/segment hosts the /stream proxy may fetch
+
+
+def _search(q):
+    html = _get(f"{BASE}/search", params={"keyword": q}).text
+    # the top-10 sidebar repeats the result markup; cut it off (ani-cli#1897)
+    html = html.split('id="main-sidebar"')[0]
+    out, seen = [], set()
+    for m in re.finditer(
+            r'<h3 class="film-name">\s*<a href="[^"]*/([^"/]+)"[^>]*title="([^"]*)"',
+            html):
+        slug, name = m.group(1), m.group(2)
+        if slug in seen:
+            continue
+        seen.add(slug)
+        img = re.search(r'<img[^>]+src="([^"]+)"', html[m.start():m.start() + 3000])
+        out.append({"id": int(slug.rsplit("-", 1)[1]) if slug.rsplit("-", 1)[1].isdigit() else 0,
+                    "slug_id": slug, "name": name,
+                    "poster": img.group(1) if img else None})
+    return out
+
+
+def _details(slug):
+    t = _get(f"{BASE}/watch/{slug}").text
+    # no details/synopsis page: og tags carry name + poster
+    title = re.search(r'property="og:title" content="([^"]+)"', t)
+    post = re.search(r'property="og:image" content="([^"]+)"', t)
+    name = re.sub(r'\s*Episode\s*\d+.*$', "", title.group(1)) if title else None
+    name = re.sub(r'\s*\|.*$', "", name).strip() if name else None
+    return {"id": int(slug.rsplit("-", 1)[1]) if slug.rsplit("-", 1)[1].isdigit() else 0,
+            "name": name, "synopsis": None, "mal_id": None,
+            "poster": post.group(1) if post else None,
+            "seasons": []}  # hianime has no seasons block (ani-cli#1897 re-searches)
+
+
+def _episodes(anime_id):
+    t = _get(f"{BASE}/api/theme/episode/list/{anime_id}").text.replace("\\", "")
+    out = []
+    for num, epid in re.findall(r'data-number="([^"]*)"[\s\S]*?data-id="(\d+)"', t):
+        out.append({"id": int(epid), "number": int(num) if num.isdigit() else num,
+                    "filler": False})
+    return out
+
+
+def _servers(ep_id):
+    return _get(f"{BASE}/api/theme/episode/servers",
+                params={"episodeId": ep_id}).text.replace("\\", "")
+
+
+def _languages(ep_id):
+    t = _servers(ep_id)
+    out = []
+    if re.search(r'data-type="sub"[\s\S]*?data-server-name="HD-1"', t):
+        out.append({"code": "jpn", "name": "Sub", "embed_url": "sub"})
+    if re.search(r'data-type="dub"[\s\S]*?data-server-name="HD-1"', t):
+        out.append({"code": "eng", "name": "Dub", "embed_url": "dub"})
+    return out
+
+
+def _deobfuscate(blob):
+    raw = base64.b64decode(blob)
+    return json.loads(bytes(b ^ KEY[i % len(KEY)] for i, b in enumerate(raw)))
+
+
+def _master_m3u8(ep_id, lang="eng"):
+    global REFERER, STREAM_HOSTS
+    t = _servers(ep_id)
+    want = "sub" if lang in ("jpn", "sub") else "dub"
+    for typ in (want, "sub") if want == "dub" else (want,):
+        m = re.search(rf'data-type="{typ}"[\s\S]*?data-server-name="HD-1"'
+                      r'[\s\S]*?data-hash="([^"]+)"', t)
+        if not m:
+            continue
+        embed = base64.b64decode(m.group(1)).decode()
+        REFERER = re.match(r'https?://[^/]+', embed).group(0) + "/"
+        j = _deobfuscate(re.search(r'window\.__P="([^"]+)"',
+                                   _get(embed, timeout=15).text).group(1))
+        master = j.get("src")
+        if not master:
+            continue
+        # prove the CDN serves us (it 403s without the referer) and register
+        # the hosts so the /stream proxy allowlist survives CDN rotation
+        r = _get(master, headers={"Referer": REFERER}, timeout=10)
+        hosts = {urlparse(master).netloc}
+        for line in r.text.splitlines():
+            if line and not line.startswith("#"):
+                hosts.add(urlparse(urljoin(master, line)).netloc)
+        STREAM_HOSTS |= hosts
+        return master
+    return None
+
+
+# Provider interface. Cache keys are namespaced with "hianime:".
+search = lambda q: _cached(f"hianime:search:{q.lower()}", 300, _search, q)
+details = lambda s: _cached(f"hianime:details:{s}", 3600, _details, s)
+episodes = lambda i: _cached(f"hianime:episodes:{i}", 300, _episodes, i)  # 5m: airing shows gain eps
+languages = lambda e: _cached(f"hianime:languages:{e}", 3600, _languages, e)
+master_m3u8 = lambda e, l="eng": _cached(f"hianime:master:{e}:{l}", 3600, _master_m3u8, e, l)
+
+
+def stream_meta(url):
+    """{Referer} for our own playlist/segment hosts, else None."""
+    if REFERER and urlparse(url).netloc in STREAM_HOSTS:
+        return {"Referer": REFERER}
+    return None

server/upstream.py

@@ -1,19 +1,30 @@
-"""upstream.py — the only module that talks to anidb.app (and keyless Jikan
-for intro timings).
+"""upstream.py — provider-agnostic source layer.
 
-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.
+A *provider* is a module in providers/ implementing the same interface:
+
+    search(q)                  -> [{id, slug_id, name, poster}]
+    details(slug_id)           -> {id, name, synopsis, mal_id, poster, seasons}
+    episodes(anime_id)         -> [{id, number, filler}]
+    languages(ep_id)           -> [{code, name, embed_url}]
+    master_m3u8(ep_id, lang)   -> master m3u8 url | None
+
+Switching sources = add providers/<name>.py, then:
+    ANI_SOURCE=other,anidb      # ordered fallback: first to succeed wins
+(default ANI_SOURCE=hianime,anidb,anikoto). Exceptions trigger the fallback; a legitimate
+None from master_m3u8 ("no dub") is final, not an error.
+
+Shared HTTP helpers (_get/get_stream/fetch_parallel, Chrome TLS
+impersonation, TTL cache) live here so every provider gets them for free;
+they work on any URL, not just anidb's.
 """
+import importlib
+import os
 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
@@ -26,10 +37,9 @@ def _session():
         _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.
+# ponytail: in-memory TTL cache shared by all providers (key namespace is
+# the provider's job). 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):
@@ -45,8 +55,6 @@ def _cached(key, ttl, fn, *a, **kw):
 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
@@ -64,7 +72,7 @@ def _get(url, **kw):
         time.sleep(0.3)
 
 
-def get_stream(url, timeout=12):
+def get_stream(url, timeout=12, headers=None):
     """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
@@ -72,7 +80,8 @@ def get_stream(url, timeout=12):
     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})
+        r = _session().get(url, **{**H, "timeout": timeout, "stream": True,
+                                  "headers": headers or {}})
         if r.status_code in (403, 429, 503) or r.status_code >= 500:
             r.close()
             if attempt:
@@ -83,18 +92,17 @@ def get_stream(url, timeout=12):
         return r
 
 
-def fetch_parallel(url, lanes=16, lane=32 * 1024, timeout=12):
+def fetch_parallel(url, lanes=16, lane=32 * 1024, timeout=12, headers=None):
     """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)."""
+    back to the single-request path — never worse than before."""
     from concurrent.futures import ThreadPoolExecutor
-    head = _session().head(url, **{**H, "timeout": timeout})
+    head = _session().head(url, **{**H, "timeout": timeout,
+                                  "headers": headers or {}})
     head.raise_for_status()
     total = int(head.headers.get("content-length") or 0)
     ct = head.headers.get("content-type", "application/octet-stream")
@@ -110,7 +118,8 @@ def fetch_parallel(url, lanes=16, lane=32 * 1024, timeout=12):
     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)
+        r = _get(url, headers={**(headers or {}), "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}")
@@ -120,66 +129,11 @@ def fetch_parallel(url, lanes=16, lane=32 * 1024, timeout=12):
         list(ex.map(grab, range(n)))
     body = b"".join(parts)
     if len(body) != total:
-        raise RuntimeError(f"stitch mismatch {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'<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"]
-
+# ---------- Jikan intros: provider-independent (keyless MAL mirror) ----------
 
 def _parse_intros(eps):
     """Jikan episode list -> one entry per episode: {title?, start?, length?}.
@@ -204,27 +158,56 @@ def _intros(mal_id):
     return _parse_intros(eps)
 
 
-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)
+intros = lambda m: _cached(f"jikan:intros:{m}", 30 * 86400, _intros, m)  # static data
+
+
+# ---------- source dispatch ----------
+
+def _providers():
+    # ponytail: env read per call, so tests and runtime can re-point without
+    # re-importing; comma list = ordered fallback
+    return [p.strip() for p in os.environ.get("ANI_SOURCE", "hianime,anidb,anikoto").split(",")
+            if p.strip()]
+
+
+def _prov(name):
+    try:
+        return importlib.import_module(f".providers.{name}", __package__)
+    except Exception:
+        return None
+
+
+def _call(name, *a, **kw):
+    errs = []
+    for p in _providers():
+        try:
+            return getattr(_prov(p), name)(*a, **kw)
+        except Exception as e:
+            errs.append(f"{p}: {e}")
+    raise RuntimeError("all sources failed -> " + " | ".join(errs))
+
+
+search = lambda q: _call("search", q)
+details = lambda s: _call("details", s)
+episodes = lambda i: _call("episodes", i)
+languages = lambda e: _call("languages", e)
+master_m3u8 = lambda e, l="eng": _call("master_m3u8", e, l)
+
+
+def stream_meta(url):
+    """Provider hook: extra upstream headers for a /stream URL (hianime's
+    CDN demands the embed origin as Referer), or None if no enabled
+    provider claims the host — that doubles as the allowlist."""
+    for p in _providers():
+        fn = getattr(_prov(p), "stream_meta", None)
+        if fn and fn(url):
+            return fn(url)
     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
+    # ponytail: assert self-check instead of a test framework.
+    # Needs a live source: ANI_SOURCE=<provider> python -m server.upstream
     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")
@@ -237,4 +220,5 @@ if __name__ == "__main__":
     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]}...")
+    import sys
     sys.exit(0)