"""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, megaplay_master 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 SUBS = {} # ep_id -> external .vtt url (the player bakes no subs) 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 'data-type="sub"' in t: out.append({"code": "jpn", "name": "Sub", "embed_url": "sub"}) if 'data-type="dub"' in 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 _embed_payload(embed): """embed url -> (master_m3u8, [{label, src} subs], referer_or_None). Two player chains have appeared: zokoanime (window.__P blob, 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: j = _deobfuscate(blob.group(1)) 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) + "/" return megaplay_master(embed, BASE + "/") def _master_m3u8(ep_id, lang="eng"): global REFERER, STREAM_HOSTS t = _servers(ep_id) want = "sub" if lang in ("jpn", "sub") else "dub" # HD-1 has rotated between player chains (zokoanime blob, megaplay); # try every listed server of the wanted type until one resolves errs = [] for typ in (want, "sub") if want == "dub" else (want,): for name, h in re.findall( rf'data-type="{typ}"[\s\S]*?data-server-name="([^"]+)"' rf'[\s\S]*?data-hash="([^"]+)"', t): try: master, subs, refr = _embed_payload(base64.b64decode(h).decode()) except Exception as e: errs.append(f"{name}: {e}") continue if not master: errs.append(f"{name}: no source") continue REFERER = refr # prove the CDN serves us and register the hosts so the /stream proxy # allowlist survives CDN rotation r = _get(master, headers={"Referer": refr} if refr else {}, 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 # the 'lang' field is useless (everything says 'en'); the real language # is the label — and it isn't first (Chainsaw Man E1 leads with Arabic) pick = next((x for x in subs if (x or {}).get("label", "").lower() == "english"), None) if not pick and subs: pick = subs[0] if pick and pick.get("src"): SUBS[ep_id] = pick["src"] return master if not errs: return None # no servers listed at all: a legitimate "no dub" # listed but unresolvable (403s, stale player format) is a provider fault, # not "no dub" — raise so the ANI_SOURCE fallback chain tries the next source raise RuntimeError("hianime: no working server: " + "; ".join(errs[:3])) # 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} (or {}) for our own playlist/segment hosts, else None — non-None doubles as the /stream allowlist; {} = host is ours but the CDN needs no referer.""" if urlparse(url).netloc in STREAM_HOSTS: return {"Referer": REFERER} if REFERER else {} return None def subtitles(ep_id): """External .vtt for this episode, or None (anidb bakes subs in).""" return SUBS.get(ep_id)