SFV

main.py

@@ -1,6 +1,6 @@
 from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
-from curl_cffi import requests as cffi_requests
+import requests
 import re
 import json
 from pydantic import BaseModel
@@ -35,148 +35,85 @@ def parse_caption_xml(xml_string: str) -> str:
     full_text = full_text.replace("'", "'").replace(""", '"').replace("&", "&").replace("'", "'")
     return full_text
 
-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")
-    
-    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")
-        
-    caption_response = cffi_requests.get(caption_url, impersonate="chrome120", timeout=8)
-    if caption_response.status_code != 200:
-        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 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}...")
+def fetch_via_proxy(proxy_base: str, target_url: str, timeout: int = 15) -> requests.Response:
+    """Fetches a URL through a public CORS/Web proxy."""
+    encoded_url = requests.utils.quote(target_url, safe='')
+    proxy_url = f"{proxy_base}{encoded_url}"
     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
+        "Accept": "text/html,application/json"
     }
-    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)
+    response = requests.get(proxy_url, headers=headers, timeout=timeout)
     if response.status_code != 200:
-        raise Exception(f"MWEB API failed: {response.status_code}")
-    return extract_captions_from_data(response.json())
+        raise Exception(f"Proxy returned status {response.status_code}")
+    return response
 
-# ========================================================
-# 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")
 
-    methods = [
-        method1_html_scrape,
-        method2_innertube_mweb,
-        method3_innertube_tv,
-        method4_invidious
+    # List of reliable public web proxies
+    proxies = [
+        "https://api.allorigins.win/raw?url=",
+        "https://corsproxy.io/?",
+        "https://api.codetabs.com/v1/proxy?quest="
     ]
 
-    for method in methods:
+    for proxy in proxies:
         try:
-            full_text = method(video_id)
-            print(f"✅ Success with {method.__name__}! Length: {len(full_text)} chars")
+            print(f"Trying proxy {proxy} for video {video_id}...")
+            
+            # 1. Fetch YouTube Watch Page HTML via Proxy
+            yt_url = f"https://www.youtube.com/watch?v={video_id}"
+            response = fetch_via_proxy(proxy, yt_url, timeout=15)
+            html = response.text
+            
+            # 2. Find the ytInitialPlayerResponse JSON in the HTML
+            match = re.search(r'ytInitialPlayerResponse\s*=\s*(\{.*?\})\s*;', html, re.DOTALL)
+            if not match:
+                print(f"Failed to find player data via {proxy}")
+                continue
+                
+            data = json.loads(match.group(1))
+            
+            # Check playability
+            playability = data.get("playabilityStatus", {})
+            if playability.get("status") == "ERROR":
+                raise Exception(playability.get("reason", "Video unavailable"))
+                
+            # 3. Extract caption tracks
+            caption_tracks = data.get("captions", {}).get("playerCaptionsTracklistRenderer", {}).get("captionTracks", [])
+            if not caption_tracks:
+                print(f"No captions found via {proxy}")
+                continue
+                
+            # Prefer English, fallback to first available
+            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:
+                continue
+                
+            print(f"Found captions, fetching XML via proxy...")
+            
+            # 4. Fetch the actual caption XML via the same proxy
+            caption_response = fetch_via_proxy(proxy, caption_url, timeout=10)
+            
+            # 5. Parse the XML into plain text
+            full_text = parse_caption_xml(caption_response.text)
+            
+            if not full_text.strip():
+                continue
+                
+            print(f"✅ Successfully fetched transcript via {proxy}. Length: {len(full_text)} chars")
             return {"text": full_text}
+            
         except Exception as e:
-            print(f"❌ {method.__name__} failed: {e}")
+            print(f"❌ Proxy {proxy} failed: {e}")
             continue
 
-    raise HTTPException(status_code=500, detail="Failed to fetch transcript after trying all methods. YouTube may be blocking this server IP entirely.")
+    raise HTTPException(status_code=500, detail="Failed to fetch transcript. All proxy methods failed or video has no captions.")
 
 # ========================================================
 # TRANSLATE API
@@ -188,12 +125,15 @@ 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"},
-            data=f"q={cffi_requests.utils.quote(req.text)}",
-            impersonate="chrome120",
+        res = 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"
+            },
+            data=f"q={requests.utils.quote(req.text)}",
             timeout=15
         )
+
         if res.status_code != 200:
             raise Exception(f"Google API returned {res.status_code}")