_stream_test.py

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 ===')