- showHome(): most recent entry renders as a hero card (poster, episode/lang/position, progress bar, Resume); the rest become mini poster cards with progress + relative timestamp - history entries now persist dur (video.duration) and poster; /api/anime returns poster (parsed from the details page) - old entries without dur render without a bar (backward compatible) - browser test asserts the seeded entry, not a section header that only existed via a cross-test side effect
115 lines
4.3 KiB
Python
115 lines
4.3 KiB
Python
"""upstream.py — the only module that talks to anidb.app.
|
|
|
|
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
|
|
|
|
from curl_cffi import requests
|
|
|
|
BASE = "https://anidb.app"
|
|
H = dict(impersonate="chrome", timeout=20)
|
|
|
|
_ANIME_LINK = re.compile(r'anime/([a-z0-9-]+)-(\d+)"[^>]*title="([^"]*)"')
|
|
|
|
|
|
def _get(url, **kw):
|
|
r = requests.get(url, **H, **kw)
|
|
if r.status_code in (403, 503) or "Just a moment" in r.text[:500]:
|
|
raise RuntimeError(f"blocked by cloudflare: {url}")
|
|
r.raise_for_status()
|
|
return r
|
|
|
|
|
|
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"]
|
|
|
|
|
|
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)
|
|
return None
|
|
|
|
|
|
def master_variants(master_url):
|
|
"""Step 7: parse #EXT-X-STREAM-INF lines -> ['1920x1080', ...] (for display)"""
|
|
text = _get(master_url).text
|
|
return re.findall(r"RESOLUTION=(\d+x\d+)", text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# ponytail: assert self-check instead of a test framework
|
|
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")
|
|
d = details(top["slug_id"])
|
|
assert d["mal_id"] == 44511 and d["synopsis"], d
|
|
assert [s["id"] for s in d["seasons"]] == [922, 923], d["seasons"]
|
|
eps = episodes(top["id"])
|
|
assert eps and eps[0]["number"] == 1, eps
|
|
m = master_m3u8(eps[0]["id"], "eng")
|
|
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]}...")
|
|
sys.exit(0)
|