jozamudi
834ca80a8c
- 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
90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
"""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)
|