DAQD

main.py

@@ -2,6 +2,7 @@ from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from curl_cffi import requests as cffi_requests
 import re
+import json
 from pydantic import BaseModel
 
 app = FastAPI()
@@ -34,127 +35,148 @@ def parse_caption_xml(xml_string: str) -> str:
     full_text = full_text.replace("'", "'").replace(""", '"').replace("&", "&").replace("'", "'")
     return full_text
 
-def fetch_via_innertube(video_id: str, payload: dict) -> str:
-    innertube_url = "https://www.youtube.com/youtubei/v1/player?key=AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8&prettyPrint=false"
-    
-    headers = {
-        "Content-Type": "application/json",
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
-        "Origin": "https://www.youtube.com",
-        "Referer": "https://www.youtube.com/",
-        "Cookie": "CONSENT=YES+1; SOCS=CAESHAgBEhJnd3NfMjAyMzEwMjUtMF9SQzIaAmVuIAEaBgiA-LaoBg"
-    }
-    
-    # Using chrome120 to avoid Hugging Face SSL bug with 131
-    response = cffi_requests.post(innertube_url, json=payload, headers=headers, impersonate="chrome120", timeout=10)
-    
-    if response.status_code != 200:
-        raise Exception(f"YouTube API returned status {response.status_code}")
-        
-    data = response.json()
-    
-    playability = data.get("playabilityStatus", {})
-    if playability.get("status") == "ERROR":
-        raise Exception(playability.get("reason", "Video unavailable"))
-        
+def extract_captions_from_data(data: dict) -> str:
     caption_tracks = data.get("captions", {}).get("playerCaptionsTracklistRenderer", {}).get("captionTracks", [])
-    
     if not caption_tracks:
-        raise Exception("No captions found in this client context.")
-        
-    selected_track = None
-    for track in caption_tracks:
-        if track.get("languageCode") == "en":
-            selected_track = track
-            break
-            
-    if not selected_track:
-        selected_track = caption_tracks[0]
-        
+        raise Exception("No captions found")
+    
+    selected_track = next((t for t in caption_tracks if t.get("languageCode") == "en"), caption_tracks[0])
     caption_url = selected_track.get("baseUrl")
     if not caption_url:
-        raise Exception("Caption URL missing.")
+        raise Exception("Caption URL missing")
         
-    caption_response = cffi_requests.get(caption_url, impersonate="chrome120", timeout=10)
-    
+    caption_response = cffi_requests.get(caption_url, impersonate="chrome120", timeout=8)
     if caption_response.status_code != 200:
-        raise Exception(f"Caption fetch failed with status {caption_response.status_code}")
+        raise Exception(f"Caption fetch failed: {caption_response.status_code}")
         
     full_text = parse_caption_xml(caption_response.text)
-    
     if not full_text.strip():
-        raise Exception("Caption XML was empty.")
-        
+        raise Exception("Caption XML empty")
     return full_text
 
+# ========================================================
+# METHOD 1: HTML Page Scrape (Exactly like Node.js package)
+# ========================================================
+def method1_html_scrape(video_id: str) -> str:
+    print(f"Trying Method 1: HTML Scrape for {video_id}...")
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+        "Accept": "text/html,application/xhtml+xml",
+        "Accept-Language": "en-US,en;q=0.9",
+        "Cookie": "CONSENT=YES+1; SOCS=CAESHAgBEhJnd3NfMjAyMzEwMjUtMF9SQzIaAmVuIAEaBgiA-LaoBg"
+    }
+    response = cffi_requests.get(f"https://www.youtube.com/watch?v={video_id}", headers=headers, impersonate="chrome120", timeout=8)
+    if response.status_code != 200:
+        raise Exception(f"HTML fetch failed: {response.status_code}")
+    
+    match = re.search(r'ytInitialPlayerResponse\s*=\s*(\{.*?\})\s*;', response.text, re.DOTALL)
+    if not match:
+        raise Exception("ytInitialPlayerResponse not found in HTML")
+    
+    data = json.loads(match.group(1))
+    return extract_captions_from_data(data)
+
+# ========================================================
+# METHOD 2: InnerTube API (MWEB - Mobile Web)
+# ========================================================
+def method2_innertube_mweb(video_id: str) -> str:
+    print(f"Trying Method 2: InnerTube MWEB for {video_id}...")
+    payload = {
+        "context": {
+            "client": {
+                "clientName": "MWEB",
+                "clientVersion": "2.20241210.00.00",
+                "hl": "en", "gl": "US"
+            }
+        },
+        "videoId": video_id
+    }
+    headers = {"Content-Type": "application/json", "Referer": "https://m.youtube.com/"}
+    response = cffi_requests.post("https://www.youtube.com/youtubei/v1/player?prettyPrint=false", json=payload, headers=headers, impersonate="chrome120", timeout=8)
+    if response.status_code != 200:
+        raise Exception(f"MWEB API failed: {response.status_code}")
+    return extract_captions_from_data(response.json())
+
+# ========================================================
+# METHOD 3: InnerTube API (TV Embedded)
+# ========================================================
+def method3_innertube_tv(video_id: str) -> str:
+    print(f"Trying Method 3: InnerTube TV for {video_id}...")
+    payload = {
+        "context": {
+            "client": {
+                "clientName": "TVHTML5_SIMPLY_EMBEDDED_PLAYER",
+                "clientVersion": "2.0",
+                "hl": "en", "gl": "US"
+            },
+            "thirdParty": {"embedUrl": "https://www.google.com"}
+        },
+        "videoId": video_id
+    }
+    headers = {"Content-Type": "application/json", "Referer": "https://www.youtube.com/"}
+    response = cffi_requests.post("https://www.youtube.com/youtubei/v1/player?prettyPrint=false", json=payload, headers=headers, impersonate="chrome120", timeout=8)
+    if response.status_code != 200:
+        raise Exception(f"TV API failed: {response.status_code}")
+    return extract_captions_from_data(response.json())
+
+# ========================================================
+# METHOD 4: Invidious API (Public Proxy)
+# ========================================================
+def method4_invidious(video_id: str) -> str:
+    print(f"Trying Method 4: Invidious Proxies for {video_id}...")
+    instances = [
+        "https://inv.nadeko.net",
+        "https://invidious.privacyredirect.com",
+        "https://iv.ggtyler.dev"
+    ]
+    for host in instances:
+        try:
+            res = cffi_requests.get(f"{host}/api/v1/captions/{video_id}?label=English", impersonate="chrome120", timeout=5)
+            if res.status_code != 200: continue
+            data = res.json()
+            if not data.get("captions"): continue
+            
+            caption = next((c for c in data["captions"] if "English" in c.get("label", "")), data["captions"][0])
+            caption_url = caption.get("url")
+            if not caption_url: continue
+            if caption_url.startswith("/"): caption_url = f"{host}{caption_url}"
+
+            cap_res = cffi_requests.get(caption_url, impersonate="chrome120", timeout=5)
+            if cap_res.status_code != 200: continue
+            
+            texts = re.findall(r'<text[^>]*>([^<]*)</text>', cap_res.text)
+            if not texts: continue
+            return " ".join(texts).replace("&#39;", "'")
+        except Exception:
+            continue
+    raise Exception("All Invidious instances failed")
+
+# ========================================================
+# MAIN ROUTE HANDLER
+# ========================================================
 @app.post("/api/transcript")
 async def get_transcript(req: TranscriptRequest):
     video_id = extract_video_id(req.url)
     if not video_id:
         raise HTTPException(status_code=400, detail="Invalid YouTube URL")
 
-    # We will try 3 different client contexts. YouTube strips captions from WEB on cloud IPs, 
-    # but usually leaves them for TV and Android contexts.
-    contexts = [
-        {
-            "name": "TV_EMBEDDED",
-            "payload": {
-                "context": {
-                    "client": {
-                        "clientName": "TVHTML5_SIMPLY_EMBEDDED_PLAYER",
-                        "clientVersion": "2.0",
-                        "hl": "en",
-                        "gl": "US"
-                    },
-                    "thirdParty": {
-                        "embedUrl": "https://www.google.com"
-                    }
-                },
-                "videoId": video_id
-            }
-        },
-        {
-            "name": "ANDROID",
-            "payload": {
-                "context": {
-                    "client": {
-                        "clientName": "ANDROID",
-                        "clientVersion": "19.29.37",
-                        "androidSdkVersion": 30,
-                        "hl": "en",
-                        "gl": "US"
-                    }
-                },
-                "videoId": video_id
-            }
-        },
-        {
-            "name": "WEB",
-            "payload": {
-                "context": {
-                    "client": {
-                        "clientName": "WEB",
-                        "clientVersion": "2.20241210.00.00",
-                        "hl": "en",
-                        "gl": "US"
-                    }
-                },
-                "videoId": video_id
-            }
-        }
+    methods = [
+        method1_html_scrape,
+        method2_innertube_mweb,
+        method3_innertube_tv,
+        method4_invidious
     ]
 
-    for ctx in contexts:
+    for method in methods:
         try:
-            print(f"Trying {ctx['name']} context for video {video_id}...")
-            full_text = fetch_via_innertube(video_id, ctx["payload"])
-            print(f"✅ Successfully fetched transcript via {ctx['name']}. Length: {len(full_text)} chars")
+            full_text = method(video_id)
+            print(f"✅ Success with {method.__name__}! Length: {len(full_text)} chars")
             return {"text": full_text}
         except Exception as e:
-            print(f"❌ {ctx['name']} failed: {e}")
+            print(f"❌ {method.__name__} failed: {e}")
             continue
 
-    raise HTTPException(status_code=500, detail="Failed to fetch transcript. YouTube is actively blocking server requests or captions are disabled.")
+    raise HTTPException(status_code=500, detail="Failed to fetch transcript after trying all methods. YouTube may be blocking this server IP entirely.")
 
 # ========================================================
 # TRANSLATE API
@@ -167,15 +189,11 @@ async def translate_text(req: TranslateRequest):
     try:
         url = f"https://translate.googleapis.com/translate_a/single?client=gtx&sl=auto&tl={req.targetLang}&dt=t"
         res = cffi_requests.post(url, 
-            headers={
-                "Content-Type": "application/x-www-form-urlencoded",
-                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
-            },
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
             data=f"q={cffi_requests.utils.quote(req.text)}",
             impersonate="chrome120",
             timeout=15
         )
-
         if res.status_code != 200:
             raise Exception(f"Google API returned {res.status_code}")