Add animekai source with 1anime direct-MP4 backend (independent CDN)

animekai.be fronted two server backends per episode: megaplay.buzz
(same player/CDN family as hianime — 403s our egress IP) and
my.1anime.site (direct progressive MP4, independent CDN — verified
serving our IP at ~2.4MB/s). This is the rescue path around the
MegaPlay Cloudflare block, per the ani-cli #1945/#1918 work.

- providers/animekai.py: standard interface, per-episode backend
  dispatch (1anime first, then megaplay), same listed-but-dead ->
  raise semantics as hianime. All endpoints ride the shared TTL cache
  (no re-fetching on repeat UI hits).
- upstream.py: megaplay_master() moved out of hianime — the MegaPlay
  chain (embed -> data-id -> getSourcesNew -> enc decrypt -> CDN proof)
  is now shared by every provider fronting that player.
- app.py: /stream forwards Range for progressive files (206 +
  Content-Range passthrough, partial bodies never LRU-cached, 8MB cap
  on the parallel path so a 250MB MP4 can't pin RAM). api_stream
  reports hls=true/false from the URL — the player kind is a property
  of the URL, so future stream types slot in without app changes.
- app.js: hls===false -> native <video> (existing Range-seeking),
  hls.js path untouched.
- run.sh: ANI_SOURCE=animekai,hianime,anidb,anikoto (revert to the old
  chain when the MegaPlay block lifts).
- tests: test_animekai.py (offline: interface, enc decrypt, /stream
  Range via monkeypatched network) + test_animekai_live.py (opt-in
  ANI_LIVE=1 end-to-end, skipped by default so the suite never touches
  animekai.be).

server/app.py

@@ -565,6 +565,9 @@ def api_stream(ep_id: int, lang: str = "jpn", request: Request = None):
         if not master:
             raise HTTPException(404, f"no {lang} stream for episode {ep_id}")
         out = {"token": _b64(master)}
+        # the player kind is a property of the URL, not the provider: .m3u8
+        # -> hls.js, anything else -> native <video> (app.js branches on this)
+        out["hls"] = master.endswith(".m3u8")
         sub = upstream.subtitles(ep_id)  # external .vtt (hianime); anidb bakes subs in
         if sub:
             out["sub"] = _b64(sub)
@@ -615,9 +618,11 @@ def _cache_put(url, ct, body):
         _STREAM_CACHE.popitem(last=False)
 
 
-def _seg_stream(r, url, ct):
+def _seg_stream(r, url, ct, cache=True):
     """Forward the CDN's bytes as they arrive; keep the LRU warm for
-    completed downloads (a truncated one would poison the cache)."""
+    completed downloads (a truncated one would poison the cache). cache=False
+    for ranged file chunks — a partial body under the full-URL key would
+    poison the LRU."""
     import time
     t0, t_first, buf, complete = time.time(), None, [], False
     try:
@@ -629,7 +634,7 @@ def _seg_stream(r, url, ct):
         complete = True
     finally:
         r.close()
-        if complete and buf:
+        if complete and buf and cache:
             _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
@@ -659,7 +664,7 @@ def _stream_playlist(url, meta=None):
 
 
 @app.get("/stream/{token}")
-def stream(token: str):
+def stream(token: str, request: Request = None):
     url = _url(token)
     p = urlparse(url)
     # ponytail: token is plain base64, so allowlist the host — otherwise any
@@ -677,6 +682,19 @@ def stream(token: str):
         return Response(hit[1], media_type=hit[0], headers=_STREAM_CACHE_HDRS)
     if url.endswith(".m3u8"):
         return _stream_playlist(url, meta)
+    # progressive file (1anime's direct MP4): the browser seeks with Range —
+    # forward it, stream the chunk back, forward 206/Content-Range, never
+    # cache (partial bodies under the full-URL key would poison the LRU)
+    rng = request.headers.get("range") if request else None
+    if rng:
+        r = _upstream(upstream.get_stream, url, timeout=30,
+                      headers={**(meta or {}), "Range": rng})
+        ct = r.headers.get("content-type", "video/mp4")
+        return StreamingResponse(
+            _seg_stream(r, url, ct, cache=False),
+            status_code=r.status_code, media_type=ct,
+            headers={k: v for k, v in r.headers.items()
+                     if k.lower() in ("content-range", "accept-ranges")})
     # 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
@@ -698,7 +716,10 @@ def stream(token: str):
         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)
+    # ponytail: 8MB cap — the LRU is for playlists/segments; a multi-hundred-MB
+    # progressive file (1anime MP4) would evict everything and pin RAM
+    if len(body) < 8 * 1024 * 1024:
+        _cache_put(url, ct, body)
     return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS)
 
 

server/providers/animekai.py

@@ -0,0 +1,196 @@
+"""providers/animekai.py — animekai.be source (hianime-family CMS, own ID space).
+
+Each episode lists one or more servers, allocated per-episode across TWO
+backends, handled by separate _backend_* functions (add a third backend =
+add one branch in _master_m3u8):
+
+- my.1anime.site  -> direct progressive MP4 (<source> on the embed page).
+  Independent CDN; its file host demands a Referer (403 without).
+- megaplay.buzz   -> MegaPlay player, shared with hianime via
+  upstream.megaplay_master. Same CDN family.
+
+master_m3u8() prefers 1anime first (independent of the MegaPlay CDN), then
+megaplay. The returned "master" is just "the URL the player fetches":
+.m3u8 -> hls.js, anything else -> native <video> through the /stream
+proxy's Range path (app.py decides by suffix — providers stay agnostic).
+
+Politeness: every endpoint rides the shared _cached TTL cache (search
+300s, details 3600s, episodes 300s — airing shows gain eps, sources
+3600s, master 3600s), so repeated UI requests cost the provider one fetch.
+"""
+import re
+from urllib.parse import urlparse
+
+from ..upstream import _cached, _get, megaplay_master
+
+BASE = "https://animekai.be"
+RETRIES = {"sub": 300, "details": 3600, "episodes": 300, "sources": 3600, "master": 3600}
+
+# ponytail: process-local slug<->id registry. details(slug) always precedes
+# episodes(anime_id) in the app flow (search -> details -> episodes), so an
+# id reaching episodes() was minted by this process's details(). A restart
+# orphans in-flight ids (user re-navigates) — same failure mode as any
+# source switch.
+_REG, _BYID = {}, {}
+_N = 0
+
+# host -> Referer its file CDN demands; {} entry = allowed, no headers.
+# stream_meta() non-None is the /stream proxy's allowlist.
+STREAM_HOSTS = {}
+
+
+def _search(q):
+    html = _get(f"{BASE}/browse", params={"keyword": q, "sort": "most_viewed"}, timeout=20).text
+    out, seen = [], set()
+    for m in re.finditer(
+            r'<a class="title" href="[^"]*/watch/([^"?/]+)"[^>]*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[max(0, m.start() - 400):m.end() + 200])
+        out.append({"id": 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}", timeout=20).text
+    global _N
+    if slug not in _REG:
+        _N += 1
+        _REG[slug] = _N
+        _BYID[_N] = slug
+    title = re.search(r'property="og:title" content="([^"]+)"', t)
+    name = re.sub(r'\s*Episode\s*\d+.*$', "", title.group(1)) if title else None
+    name = re.sub(r'\s*[-|–]\s*AnimeKai\s*$', "", name or "").strip() or None
+    post = re.search(r'property="og:image" content="([^"]+)"', t)
+    syn = re.search(r'<div class="desc text-expand"[^>]*>([\s\S]*?)</div>', t)
+    synopsis = re.sub(r"<[^>]+>", "", syn.group(1)).strip() if syn else None
+    return {"id": _REG[slug], "name": name, "synopsis": synopsis,
+            "mal_id": None,
+            "poster": post.group(1) if post else None,
+            "seasons": []}
+
+
+def _episodes(anime_id):
+    slug = _BYID.get(anime_id)
+    if slug is None:
+        raise RuntimeError(f"animekai: unknown anime id {anime_id} (stale after restart?)")
+    t = _get(f"{BASE}/watch/{slug}", timeout=20).text
+    out, seen = [], set()
+    for num in re.findall(rf'/watch/{re.escape(slug)}/ep-(\d+)"', t):
+        n = int(num)
+        if n in seen:
+            continue
+        seen.add(n)
+        out.append({"id": anime_id * 1000 + n, "number": n, "filler": False})
+    out.sort(key=lambda e: (isinstance(e["number"], str), e["number"]))
+    return out
+
+
+def _sources(slug, num):
+    d = _get(f"{BASE}/watch/{slug}/ep/{num}/sources", timeout=20).json()
+    return d.get("sources") or {}
+
+
+_sources_cached = lambda slug, num: _cached(
+    f"animekai:sources:{slug}:{num}", RETRIES["sources"], _sources, slug, num)
+
+
+def _languages(ep_id):
+    anime_id, num = divmod(ep_id, 1000)
+    slug = _BYID.get(anime_id)
+    if slug is None:
+        return []
+    s = _sources_cached(slug, num)
+    out = []
+    if s.get("sub"):
+        out.append({"code": "jpn", "name": "Sub", "embed_url": "sub"})
+    if s.get("dub"):
+        out.append({"code": "eng", "name": "Dub", "embed_url": "dub"})
+    return out
+
+
+def _backend_oneanime(source_url):
+    """1anime embed -> direct MP4 url, or None. The file CDN 403s without a
+    Referer, so the proof fetch doubles as the referer discovery."""
+    page = _get(source_url, headers={"Referer": BASE + "/"}, timeout=15).text
+    m = re.search(r'<source src="([^"]+)"', page)
+    if not m:
+        return None
+    url = m.group(1)
+    up = urlparse(source_url)
+    host = urlparse(url).netloc
+    # prove it serves us AND learn the referer the file host demands
+    for refr in (f"{up.scheme}://{up.netloc}/", f"https://{host}/", None):
+        try:
+            _get(url, headers={"Range": "bytes=0-1023", **({"Referer": refr} if refr else {})},
+                 timeout=10)
+            STREAM_HOSTS[host] = {"Referer": refr} if refr else {}
+            return url
+        except Exception:
+            continue
+    return None
+
+
+def _backend_megaplay(source_url):
+    """MegaPlay embed -> master m3u8 url, or None (shared chain)."""
+    master, _subs, refr = megaplay_master(source_url, BASE + "/")
+    if not master:
+        return None
+    host = urlparse(master).netloc
+    STREAM_HOSTS[host] = {"Referer": refr} if refr else {}
+    return master
+
+
+_BACKENDS = (("1anime", _backend_oneanime), ("megaplay", _backend_megaplay))
+
+
+def _master_m3u8(ep_id, lang="eng"):
+    anime_id, num = divmod(ep_id, 1000)
+    slug = _BYID.get(anime_id)
+    if slug is None:
+        raise RuntimeError(f"animekai: unknown anime id {anime_id} (stale after restart?)")
+    s = _sources_cached(slug, num)
+    want = "sub" if lang in ("jpn", "sub") else "dub"
+    # 1anime first: independent CDN, so it rides through MegaPlay outages
+    for mode in ((want, "sub") if want == "dub" else (want,)):
+        servers = s.get(mode) or []
+        if not servers:
+            continue
+        for _name, url in ((x.get("server_name"), x.get("source_url")) for x in servers):
+            for _bname, backend in _BACKENDS:
+                if url and _bname in url:
+                    try:
+                        out = backend(url)
+                    except Exception:
+                        out = None
+                    if out:
+                        return out
+    # distinguish "no servers listed" (legit None = no dub) from "listed but
+    # dead" (raise -> ANI_SOURCE fallback), same semantics as hianime
+    if s.get(want) or (want == "dub" and s.get("sub")):
+        raise RuntimeError(f"animekai: ep {num} servers listed but none resolved")
+    return None
+
+
+# Provider interface. Cache keys namespaced "animekai:"; TTLs keep the
+# provider from seeing the same request twice (see module docstring).
+search = lambda q: _cached(f"animekai:search:{q.lower()}", RETRIES["sub"], _search, q)
+details = lambda s: _cached(f"animekai:details:{s}", RETRIES["details"], _details, s)
+episodes = lambda i: _cached(f"animekai:episodes:{i}", RETRIES["episodes"], _episodes, i)
+languages = lambda e: _cached(f"animekai:languages:{e}", RETRIES["sources"], _languages, e)
+master_m3u8 = lambda e, l="eng": _cached(f"animekai:master:{e}:{l}", RETRIES["master"], _master_m3u8, e, l)
+
+
+def stream_meta(url):
+    """{Referer} (or {}) for our own file hosts, else None — non-None
+    doubles as the /stream allowlist."""
+    return STREAM_HOSTS.get(urlparse(url).netloc)
+
+
+def subtitles(ep_id):
+    # 1anime MP4s carry no separate subs; megaplay subs are in the tracks
+    # the player bakes in — nothing external to hand the frontend
+    return None

server/providers/hianime.py

@@ -13,7 +13,7 @@ import json
 import re
 from urllib.parse import urljoin, urlparse
 
-from ..upstream import _cached, _get, megaplay_sources
+from ..upstream import _cached, _get, megaplay_master
 
 BASE = "https://hianime.at"
 KEY = b"otaku-embed-v1"
@@ -88,9 +88,8 @@ def _embed_payload(embed):
     """embed url -> (master_m3u8, [{label, src} subs], referer_or_None).
 
     Two player chains have appeared: zokoanime (window.__P blob,
-    base64+XOR) and megaplay (page needs the hianime referer; data-id ->
-    getSourcesNew; CDN domains rotate and most 403 us, so re-roll until
-    one serves)."""
+    base64+XOR — hianime-specific) and megaplay (shared: upstream.
+    megaplay_master, used by animekai too)."""
     page = _get(embed, timeout=15).text
     blob = re.search(r'window\.__P="([^"]+)"', page)
     if blob:
@@ -98,30 +97,7 @@ def _embed_payload(embed):
         subs = [{"label": (x or {}).get("label", ""), "src": (x or {}).get("src")}
                 for x in j.get("subtitles") or []]
         return j.get("src"), subs, re.match(r'https?://[^/]+', embed).group(0) + "/"
-    # megaplay: without the hianime referer the page is an error stub with no data-id
-    if not re.search(r'data-id="\d+"', page):
-        page = _get(embed, headers={"Referer": BASE + "/"}, timeout=15).text
-    mid = re.search(r'data-id="(\d+)"', page)
-    if not mid:
-        return None, [], None
-    host = re.match(r'https?://([^/]+)', embed).group(1)
-    # ponytail: one attempt — the enc payload is deterministic per data-id,
-    # so re-rolling getSourcesNew would just spam the provider
-    d = megaplay_sources(_get(f"https://{host}/stream/getSourcesNew?id={mid.group(1)}",
-             headers={"Referer": embed, "X-Requested-With": "XMLHttpRequest"},
-             timeout=15).json())
-    src = (d.get("sources") or {}).get("file")
-    if not src:
-        return None, [], None
-    subs = [{"label": (x or {}).get("label", ""), "src": (x or {}).get("file")}
-            for x in d.get("tracks") or []]
-    for cand, refr in (({"Referer": embed}, embed), ({}, None)):
-        try:
-            _get(src, headers=cand, timeout=10)
-            return src, subs, refr
-        except Exception:
-            continue
-    return None, [], None
+    return megaplay_master(embed, BASE + "/")
 
 
 def _master_m3u8(ep_id, lang="eng"):

server/upstream.py

@@ -79,6 +79,39 @@ def megaplay_sources(d):
     return d
 
 
+def megaplay_master(embed, referer):
+    """MegaPlay embed url -> (master_url, [{label, src} subs], referer_or_None).
+
+    Shared by every provider fronting the MegaPlay player (hianime,
+    animekai): embed page (referer-gated) -> data-id -> getSourcesNew
+    (megaplay_sources) -> prove the CDN serves us, remembering which
+    Referer variant worked. referer = the front site's home (the embed
+    page 403s a stub without it). One attempt: the enc payload is
+    deterministic per data-id, re-rolling would just spam the provider.
+    The caller registers the returned master's hosts in its own
+    /stream allowlist."""
+    page = _get(embed, headers={"Referer": referer}, timeout=15).text
+    mid = re.search(r'data-id="(\d+)"', page)
+    if not mid:
+        return None, [], None
+    host = re.match(r'https?://([^/]+)', embed).group(1)
+    d = megaplay_sources(_get(f"https://{host}/stream/getSourcesNew?id={mid.group(1)}",
+             headers={"Referer": embed, "X-Requested-With": "XMLHttpRequest"},
+             timeout=15).json())
+    src = (d.get("sources") or {}).get("file")
+    if not src:
+        return None, [], None
+    subs = [{"label": (x or {}).get("label", ""), "src": (x or {}).get("file")}
+            for x in d.get("tracks") or []]
+    for cand, refr in (({"Referer": embed}, embed), ({}, None)):
+        try:
+            _get(src, headers=cand, timeout=10)
+            return src, subs, refr
+        except Exception:
+            continue
+    return None, [], None
+
+
 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: