ztoolx commited on
Commit
e573d34
·
verified ·
1 Parent(s): aaacaa2
Files changed (1) hide show
  1. main.py +121 -0
main.py ADDED
@@ -0,0 +1,121 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from fastapi import FastAPI, HTTPException
2
+ from fastapi.middleware.cors import CORSMiddleware
3
+ import requests
4
+ import re
5
+ from pydantic import BaseModel
6
+
7
+ app = FastAPI()
8
+
9
+ # Enable CORS so your Vercel frontend can call this API
10
+ app.add_middleware(
11
+ CORSMiddleware,
12
+ allow_origins=["*"],
13
+ allow_credentials=True,
14
+ allow_methods=["*"],
15
+ allow_headers=["*"],
16
+ )
17
+
18
+ class TranscriptRequest(BaseModel):
19
+ url: str
20
+
21
+ class TranslateRequest(BaseModel):
22
+ text: str
23
+ targetLang: str
24
+
25
+ def extract_video_id(url: str) -> str:
26
+ pattern = r'(?:youtube\.com\/(?:[^\/]+\/.+\/|(?:v|e(?:mbed)?)\/|.*[?&]v=)|youtu\.be\/)([^"&?\/\s]{11})'
27
+ match = re.search(pattern, url)
28
+ return match.group(1) if match else None
29
+
30
+ @app.post("/api/transcript")
31
+ async def get_transcript(req: TranscriptRequest):
32
+ video_id = extract_video_id(req.url)
33
+ if not video_id:
34
+ raise HTTPException(status_code=400, detail="Invalid YouTube URL")
35
+
36
+ # We will try multiple Invidious instances in case one is down
37
+ instances = [
38
+ "https://inv.nadeko.net",
39
+ "https://invidious.fdn.fr",
40
+ "https://invidious.protokolla.fi",
41
+ "https://iv.datura.network"
42
+ ]
43
+
44
+ for host in instances:
45
+ try:
46
+ print(f"Trying {host} for video {video_id}...")
47
+ # 1. Fetch available captions from Invidious
48
+ res = requests.get(f"{host}/api/v1/captions/{video_id}?label=English", timeout=5)
49
+
50
+ if res.status_code != 200:
51
+ continue # Try next instance
52
+
53
+ data = res.json()
54
+ if not data.get("captions"):
55
+ continue # Try next instance
56
+
57
+ # 2. Find the English caption URL (or first available)
58
+ caption = next((c for c in data["captions"] if "English" in c.get("label", "")), data["captions"][0])
59
+ caption_url = caption.get("url")
60
+
61
+ if not caption_url:
62
+ continue
63
+
64
+ if caption_url.startswith("/"):
65
+ caption_url = f"{host}{caption_url}"
66
+
67
+ # 3. Fetch the actual caption XML
68
+ cap_res = requests.get(caption_url, timeout=5)
69
+ if cap_res.status_code != 200:
70
+ continue
71
+
72
+ # 4. Parse XML to plain text
73
+ texts = re.findall(r'<text[^>]*>([^<]*)</text>', cap_res.text)
74
+ if not texts:
75
+ continue
76
+
77
+ full_text = " ".join(texts)
78
+ full_text = full_text.replace("&amp;#39;", "'").replace("&amp;quot;", '"').replace("&amp;amp;", "&").replace("&#39;", "'").replace("\n", " ")
79
+
80
+ print(f"Successfully fetched transcript from {host}. Length: {len(full_text)} chars")
81
+ return {"text": full_text}
82
+
83
+ except Exception as e:
84
+ print(f"Failed on {host}: {e}")
85
+ continue # Try next instance
86
+
87
+ raise HTTPException(status_code=500, detail="Failed to fetch transcript. All sources timed out or are blocked.")
88
+
89
+ # ========================================================
90
+ # TRANSLATE API
91
+ # ========================================================
92
+ @app.post("/api/translate")
93
+ async def translate_text(req: TranslateRequest):
94
+ if not req.text or not req.targetLang:
95
+ raise HTTPException(status_code=400, detail="Missing text or target language")
96
+
97
+ try:
98
+ url = f"https://translate.googleapis.com/translate_a/single?client=gtx&sl=auto&tl={req.targetLang}&dt=t"
99
+ res = requests.post(url,
100
+ headers={
101
+ "Content-Type": "application/x-www-form-urlencoded",
102
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
103
+ },
104
+ data=f"q={requests.utils.quote(req.text)}",
105
+ timeout=10
106
+ )
107
+
108
+ if res.status_code != 200:
109
+ raise Exception(f"Google API returned {res.status_code}")
110
+
111
+ data = res.json()
112
+ translated = "".join([item[0] for item in data[0]])
113
+ return {"translatedText": translated}
114
+
115
+ except Exception as e:
116
+ raise HTTPException(status_code=500, detail=f"Translation failed: {str(e)}")
117
+
118
+ # KEEP-ALIVE ENDPOINT
119
+ @app.get("/api/health")
120
+ async def health_check():
121
+ return {"status": "ok", "message": "Transcript API is running!"}