diff --git a/server/app.py b/server/app.py index 828984f..3523936 100644 --- a/server/app.py +++ b/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] diff --git a/server/providers/__init__.py b/server/providers/__init__.py new file mode 100644 index 0000000..75bddbe --- /dev/null +++ b/server/providers/__init__.py @@ -0,0 +1 @@ +"""providers/ — anime source providers. See upstream.py for the interface.""" diff --git a/server/providers/anidb.py b/server/providers/anidb.py new file mode 100644 index 0000000..d45f42d --- /dev/null +++ b/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']*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']*>(.*?)', 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: ' 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) diff --git a/server/providers/anikoto.py b/server/providers/anikoto.py new file mode 100644 index 0000000..8cd6cd8 --- /dev/null +++ b/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})', + html): + slug, name = m.group(1), m.group(2).strip() + if slug in seen: + continue + seen.add(slug) + imgs = re.findall(r']*>(.*?)', t, re.S) + syn = re.search(r'class="synopsis[^"]*"[\s\S]*?class="content"[\s\S]*?>(.*?)', 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']+>', 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]*?
    ([\s\S]*?)
', 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) diff --git a/server/providers/hianime.py b/server/providers/hianime.py new file mode 100644 index 0000000..6f11fec --- /dev/null +++ b/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'

\s*]*title="([^"]*)"', + html): + slug, name = m.group(1), m.group(2) + if slug in seen: + continue + seen.add(slug) + img = re.search(r']+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 diff --git a/server/upstream.py b/server/upstream.py index 659abf4..bbe6624 100644 --- a/server/upstream.py +++ b/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/.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']*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']*>(.*?)', 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: ' 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= 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) diff --git a/static/app.js b/static/app.js index 96b4bf1..ee8b951 100644 --- a/static/app.js +++ b/static/app.js @@ -762,7 +762,7 @@ function videoEl() { function stopHls() { if (hls) { hls.destroy(); hls = null; } if (bufTimer) { clearInterval(bufTimer); bufTimer = null; } } -let currentEp = null, lastSavedPos = 0, playerBroken = false, nextCta = null; +let currentEp = null, lastSavedPos = 0, lastSeenPos = 0, playerBroken = false, nextCta = null; function playerOverlay(text, onAction, actionLabel = "Retry", spin = false) { const p = document.getElementById("player"); @@ -786,8 +786,9 @@ function playerFail(msg) { playerBroken = true; // a fatal error stops the periodic saves (timeupdate/pause never come // again); persist the position now or the user loses everything since - // the last 10s save when they leave - savePos(); + // the last 10s save when they leave. hls.js may have already reset + // currentTime to 0, so floor at the furthest point reached this episode + savePos(lastSeenPos); const { ep_id, ep_no } = currentEp; postEvent({type: "error", detail: msg, ep_id, ep: ep_no, lang, ua: navigator.userAgent}); playerOverlay("Playback failed: " + msg, () => play(ep_id, ep_no)); @@ -1008,6 +1009,7 @@ async function play(ep_id, ep_no) { } currentEp = {ep_id, ep_no, v}; lastSavedPos = 0; + lastSeenPos = 0; if (nextCta) { nextCta.remove(); nextCta = null; } const saved = history.find(x => x.slug_id === detail.slug_id && x.ep_id === ep_id); if (saved?.pos > 10) { @@ -1090,16 +1092,24 @@ const meta = v => ({...(Number.isFinite(v.duration) && v.duration > 0 ? {dur: v. // one savePos covers all video elements: it reads currentEp, so stale // closures from earlier plays can't write the wrong episode's position -function savePos() { - if (!currentEp || currentEp.v.currentTime <= 3) return; - lastSavedPos = currentEp.v.currentTime; +function savePos(floor = 0) { + if (!currentEp) return; + // a fatal frag error can reset video.currentTime to 0 before this runs + // (hls.js recovery); playerFail passes the furthest point reached as floor + const pos = Math.max(currentEp.v.currentTime, floor); + if (pos <= 3) return; + lastSavedPos = pos; postHistory({...meta(currentEp.v), title: detail.name, slug_id: detail.slug_id, season_id: seasonId, - ep_id: currentEp.ep_id, lang, ep: currentEp.ep_no, pos: currentEp.v.currentTime, + ep_id: currentEp.ep_id, lang, ep: currentEp.ep_no, pos, ...(epsList.length ? {total: epsList.length} : {})}); } function watchPos(v) { - v.addEventListener("timeupdate", () => { if (v.currentTime - lastSavedPos >= 10) savePos(); }); + // 'seeking' fires when currentTime is set, even before the target segment + // buffers — so a seek made just before a stream death is still captured + const seen = () => { lastSeenPos = Math.max(lastSeenPos, v.currentTime); }; + v.addEventListener("timeupdate", () => { seen(); if (v.currentTime - lastSavedPos >= 10) savePos(); }); + v.addEventListener("seeking", seen); v.addEventListener("pause", savePos); // pagehide is bound once at boot (per-play listeners accumulated and fired // with stale video elements) diff --git a/tests/test_sources.py b/tests/test_sources.py new file mode 100644 index 0000000..28ea0ec --- /dev/null +++ b/tests/test_sources.py @@ -0,0 +1,76 @@ +"""test_sources.py — ANI_SOURCE dispatch + fallback, offline (fake providers). + +Run: .venv/bin/pytest tests/test_sources.py -q +""" +import os +import sys +import types +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +FAKE = {"id": 1, "slug_id": "fake-1", "name": "Fake Anime", "poster": None} + + +def _fake(name, boom=False): + m = types.ModuleType(f"server.providers.{name}") + + def search(q, boom=boom): + if boom: + raise RuntimeError(f"{name} is down (test)") + return [FAKE] + + m.search = search + m.details = lambda s: {"id": 1, "name": "Fake Anime"} + m.episodes = lambda i: [{"id": 1, "number": 1, "filler": False}] + m.languages = lambda e: [{"code": "jpn", "name": "Sub", "embed_url": "x"}] + m.master_m3u8 = lambda e, l="eng": "https://cdn.example/master.m3u8" + sys.modules[m.__name__] = m + return m + + +def test_fallback_skips_dead_provider(): + _fake("good") + _fake("dead", boom=True) + os.environ["ANI_SOURCE"] = "dead,good" + from server import upstream + r = upstream.search("anything") + assert r == [FAKE], r + + +def test_all_dead_raises_with_all_errors(): + _fake("dead", boom=True) + os.environ["ANI_SOURCE"] = "dead" + from server import upstream + try: + upstream.search("x") + except RuntimeError as e: + assert "dead" in str(e) and "all sources failed" in str(e), e + else: + raise AssertionError("expected all-sources-failed") + + +def test_anidb_provider_imports(): + import server.providers.anidb as a # noqa: F401 + for fn in ("search", "details", "episodes", "languages", "master_m3u8"): + assert callable(getattr(a, fn)), fn + + +def test_default_is_hianime_first(): + os.environ.pop("ANI_SOURCE", None) + from server import upstream + assert upstream._providers()[:3] == ["hianime", "anidb", "anikoto"] + + +def test_hianime_provider_imports(): + import server.providers.hianime as a # noqa: F401 + for fn in ("search", "details", "episodes", "languages", "master_m3u8"): + assert callable(getattr(a, fn)), fn + assert callable(a.stream_meta) + + +def test_anikoto_provider_imports(): + import server.providers.anikoto as a # noqa: F401 + for fn in ("search", "details", "episodes", "languages", "master_m3u8"): + assert callable(getattr(a, fn)), fn