File size: 5,014 Bytes
b4c1d87 7e89908 b4c1d87 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | import json, os, re, sys, urllib.parse, urllib.request, urllib.error
PORT = os.environ.get('TEST_PORT', '5001')
BASE = f'http://127.0.0.1:{PORT}'
UA = {'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/126 Safari/537.36'}
def req(url, data=None, timeout=60):
body = json.dumps(data).encode() if data is not None else None
headers = {**UA, 'Content-Type': 'application/json'} if data is not None else UA
r = urllib.request.Request(url, data=body, headers=headers)
try:
with urllib.request.urlopen(r, timeout=timeout) as resp:
return resp.status, dict(resp.headers), resp.read()
except urllib.error.HTTPError as e:
return e.code, dict(e.headers), e.read()
except Exception as e:
return 0, {}, str(e).encode()
def test_sources(anime_id, ep, lang='sub', providers=None):
providers = providers or ['anibd','anikoto','reanime','senshi','kaa','2dhive',
'animegg','animenosub','animedunya','anidbapp','anineko','mkissa']
ok = 0
for p in providers:
st, hd, body = req(f'{BASE}/api/watch/sources',
{'anime_id': anime_id, 'episode_number': ep, 'language': lang, 'provider': p})
try:
d = json.loads(body)
except Exception:
print(f' {p:12s} st={st} RAW={body[:100]!r}')
continue
vl = d.get('video_link') or ''
proxied = '/api/stream-proxy?' in vl
n_hls = len(d.get('hls_sources') or [])
n_emb = len(d.get('embed_sources') or [])
n_trk = len(d.get('tracks') or [])
usable = st == 200 and d.get('available') and (bool(vl) or n_hls or n_emb)
if usable:
ok += 1
print(f' {p:12s} st={st} avail={d.get("available")} type={d.get("source_type")} '
f'proxied={proxied} hls={n_hls} embeds={n_emb} tracks={n_trk}')
if vl and not proxied:
print(f' !! NOT PROXIED: {vl[:110]}')
if not usable and st == 200:
print(f' detail: {json.dumps(d)[:180]}')
print(f' -> {ok}/{len(providers)} usable')
return ok
def fetch_proxied(manifest_text):
"""Return the first rewritten URL from a proxied manifest (or None)."""
for ln in manifest_text.splitlines():
if '/api/stream-proxy' in ln:
return ln if ln.startswith('http') else f'{BASE}{ln}'
m = re.search(r'(/api/stream-proxy[^"\s]+)', manifest_text)
return (m.group(1) if m and m.group(1).startswith('http') else (f'{BASE}{m.group(1)}' if m else None))
def proxy_roundtrip(url, ref, depth=3):
"""Walk master → variant → segment through /api/stream-proxy."""
q = urllib.parse.urlencode({'url': url, 'ref': ref})
st, hd, body = req(f'{BASE}/api/stream-proxy?{q}')
if st != 200:
print(f' PROXY {st}: {body[:200]!r}')
return False
text = body.decode('utf-8', 'replace')
rew = [ln for ln in text.splitlines() if '/api/stream-proxy' in ln]
print(f' master st={st} ctype={hd.get("Content-Type")} lines={len(text.splitlines())} rewritten={len(rew)}')
print(' sample:', rew[0][:110] if rew else 'NONE')
cur = text
for i in range(depth):
nxt = fetch_proxied(cur)
if not nxt:
print(' no more rewritten URLs — done')
return True
st2, hd2, body2 = req(nxt, timeout=120)
ctype = hd2.get('Content-Type') or ''
print(f' hop{i+1} st={st2} ctype={ctype} bytes={len(body2)}')
if st2 != 200:
print(f' BODY: {body2[:200]!r}')
return False
if len(body2) > 50000:
print(f' hop{i+1} looks like a real segment ({len(body2)} bytes) — PLAYABLE')
return True
if 'mpegurl' not in ctype and 'dash' not in ctype:
print(f' hop{i+1} is binary/short — treat as segment, PLAYABLE')
return True
cur = body2.decode('utf-8', 'replace')
return True
print(f'=== target: {BASE} ===')
print('\n--- 1) FMA:B (5114) ep 1 sub — all providers ---')
n1 = test_sources(5114, 1)
print('\n--- 2) FMA:B (5114) ep 1 DUB ---')
n2 = test_sources(5114, 1, lang='dub', providers=['anibd','anikoto','reanime','anidbapp'])
print('\n--- 3) One Piece (21) ep 1 sub ---')
n3 = test_sources(21, 1, providers=['anibd','reanime','anikoto','2dhive'])
print('\n--- 4) Stream-proxy round-trip on a real anibd m3u8 ---')
proxy_roundtrip('https://playeng.animeapps.top/r2/cachehd/5114ebd1/index.m3u8',
'https://playeng.animeapps.top/')
print('\n--- 5) Embed source check (reanime, iframe-based) ---')
st, hd, body = req(f'{BASE}/api/watch/sources',
{'anime_id': 5114, 'episode_number': 1, 'language': 'sub', 'provider': 'reanime'})
if st == 200:
d = json.loads(body)
print(' type:', d.get('source_type'), '| embeds:', len(d.get('embed_sources') or []))
for e in (d.get('embed_sources') or [])[:2]:
print(' embed:', str(e)[:150])
print('\n=== DONE ===')
|