cloudflare-worker/test_worker_local.py

"""اختبار محلي لـ Cloudflare Worker - محاكاة بيئة Cloudflare"""
import json
import sys
sys.path.insert(0, '/home/z/my-project/DownTube')

# اختبار YouTube Innertube API مباشرة
import httpx
import asyncio

INNERTUBE_API_KEY = 'AIzaSyA8eiZmM1FaDVjRy-df2KTyQ_vz_yYM39w'
INNERTUBE_API_URL = 'https://www.youtube.com/youtubei/v1/player'

async def test_innertube():
    """اختبار YouTube Innertube API مع عملاء مختلفين"""
    video_id = "dQw4w9WgXcQ"  # Rick Astley
    
    clients = [
        {
            "name": "ANDROID",
            "body": {
                "videoId": video_id,
                "context": {
                    "client": {
                        "clientName": "ANDROID",
                        "clientVersion": "19.29.37",
                        "androidSdkVersion": 30,
                        "hl": "en",
                        "gl": "US",
                    }
                },
                "playbackContext": {
                    "contentPlaybackContext": {
                        "html5Preference": "HTML5_PREF_WANTS",
                    }
                },
                "contentCheckOk": True,
                "racyCheckOk": True,
            }
        },
        {
            "name": "IOS",
            "body": {
                "videoId": video_id,
                "context": {
                    "client": {
                        "clientName": "IOS",
                        "clientVersion": "19.29.1",
                        "deviceModel": "iPhone16,2",
                        "hl": "en",
                        "gl": "US",
                    }
                }
            }
        },
        {
            "name": "MWEB",
            "body": {
                "videoId": video_id,
                "context": {
                    "client": {
                        "clientName": "MWEB",
                        "clientVersion": "2.20240726.01.00",
                        "hl": "en",
                        "gl": "US",
                    }
                }
            }
        },
    ]
    
    async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
        for c in clients:
            print(f"\n{'='*60}")
            print(f"Testing client: {c['name']}")
            print(f"{'='*60}")
            
            try:
                r = await client.post(
                    f"{INNERTUBE_API_URL}?key={INNERTUBE_API_KEY}&prettyPrint=false",
                    json=c["body"],
                    headers={
                        "Content-Type": "application/json",
                        "User-Agent": "com.google.android.youtube/19.29.37 (Linux; U; Android 14)",
                    }
                )
                
                print(f"Status: {r.status_code}")
                
                if r.status_code == 200:
                    data = r.json()
                    status = data.get("playabilityStatus", {}).get("status", "UNKNOWN")
                    print(f"Playability: {status}")
                    
                    if status == "OK":
                        title = data.get("videoDetails", {}).get("title", "?")
                        print(f"Title: {title}")
                        
                        sd = data.get("streamingData", {})
                        formats = sd.get("formats", [])
                        adaptive = sd.get("adaptiveFormats", [])
                        
                        print(f"Combined formats: {len(formats)}")
                        print(f"Adaptive formats: {len(adaptive)}")
                        
                        # Check for URLs
                        has_url = any(f.get("url") for f in formats)
                        has_cipher = any(f.get("signatureCipher") or f.get("cipher") for f in formats)
                        print(f"Has direct URLs: {has_url}")
                        print(f"Has cipher: {has_cipher}")
                        
                        # Show first format with URL
                        for f in formats[:3]:
                            quality = f.get("qualityLabel", f.get("quality", "?"))
                            has_direct_url = bool(f.get("url"))
                            has_sig = bool(f.get("signatureCipher") or f.get("cipher"))
                            print(f"  Format: {quality} | URL: {has_direct_url} | Cipher: {has_sig}")
                        
                        # Show video+audio adaptive formats
                        video_fmts = [f for f in adaptive if f.get("mimeType","").startswith("video/")]
                        audio_fmts = [f for f in adaptive if f.get("mimeType","").startswith("audio/")]
                        print(f"\nVideo adaptive: {len(video_fmts)}")
                        for f in video_fmts[:3]:
                            quality = f.get("qualityLabel", "?")
                            has_direct_url = bool(f.get("url"))
                            has_sig = bool(f.get("signatureCipher") or f.get("cipher"))
                            print(f"  {quality} | URL: {has_direct_url} | Cipher: {has_sig}")
                        
                        print(f"\nAudio adaptive: {len(audio_fmts)}")
                        for f in audio_fmts[:2]:
                            quality = f.get("audioQuality", "?")
                            has_direct_url = bool(f.get("url"))
                            has_sig = bool(f.get("signatureCipher") or f.get("cipher"))
                            print(f"  {quality} | URL: {has_direct_url} | Cipher: {has_sig}")
                        
                        # This client works! Return success
                        return True
                    else:
                        reason = data.get("playabilityStatus", {}).get("reason", "")
                        print(f"Reason: {reason}")
                else:
                    print(f"Response: {r.text[:200]}")
                    
            except Exception as e:
                print(f"Error: {e}")
    
    return False

result = asyncio.run(test_innertube())
print(f"\n{'='*60}")
print(f"RESULT: {'SUCCESS' if result else 'FAILED'}")
print(f"{'='*60}")