-
Notifications
You must be signed in to change notification settings - Fork 48
/
Copy pathmain.py
150 lines (129 loc) · 5.96 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import json
import os
from urllib.parse import urlparse, parse_qs, urlencode
from urllib.request import urlopen
from typing import Optional, List
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
try:
from youtube_transcript_api import YouTubeTranscriptApi
except ImportError:
raise ImportError(
"`youtube_transcript_api` not installed. Please install using `pip install youtube_transcript_api`"
)
app = FastAPI(title="YouTube Tools API")
class YouTubeTools:
@staticmethod
def get_youtube_video_id(url: str) -> Optional[str]:
"""Function to get the video ID from a YouTube URL."""
parsed_url = urlparse(url)
hostname = parsed_url.hostname
if hostname == "youtu.be":
return parsed_url.path[1:]
if hostname in ("www.youtube.com", "youtube.com"):
if parsed_url.path == "/watch":
query_params = parse_qs(parsed_url.query)
return query_params.get("v", [None])[0]
if parsed_url.path.startswith("/embed/"):
return parsed_url.path.split("/")[2]
if parsed_url.path.startswith("/v/"):
return parsed_url.path.split("/")[2]
return None
@staticmethod
def get_video_data(url: str) -> dict:
"""Function to get video data from a YouTube URL."""
if not url:
raise HTTPException(status_code=400, detail="No URL provided")
try:
video_id = YouTubeTools.get_youtube_video_id(url)
if not video_id:
raise HTTPException(status_code=400, detail="Invalid YouTube URL")
except Exception:
raise HTTPException(status_code=400, detail="Error getting video ID from URL")
try:
params = {"format": "json", "url": f"https://www.youtube.com/watch?v={video_id}"}
oembed_url = "https://www.youtube.com/oembed"
query_string = urlencode(params)
full_url = oembed_url + "?" + query_string
with urlopen(full_url) as response:
response_text = response.read()
video_data = json.loads(response_text.decode())
clean_data = {
"title": video_data.get("title"),
"author_name": video_data.get("author_name"),
"author_url": video_data.get("author_url"),
"type": video_data.get("type"),
"height": video_data.get("height"),
"width": video_data.get("width"),
"version": video_data.get("version"),
"provider_name": video_data.get("provider_name"),
"provider_url": video_data.get("provider_url"),
"thumbnail_url": video_data.get("thumbnail_url"),
}
return clean_data
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error getting video data: {str(e)}")
@staticmethod
def get_video_captions(url: str, languages: Optional[List[str]] = None) -> str:
"""Get captions from a YouTube video."""
if not url:
raise HTTPException(status_code=400, detail="No URL provided")
try:
video_id = YouTubeTools.get_youtube_video_id(url)
if not video_id:
raise HTTPException(status_code=400, detail="Invalid YouTube URL")
except Exception:
raise HTTPException(status_code=400, detail="Error getting video ID from URL")
try:
captions = None
if languages:
captions = YouTubeTranscriptApi.get_transcript(video_id, languages=languages)
else:
captions = YouTubeTranscriptApi.get_transcript(video_id)
if captions:
return " ".join(line["text"] for line in captions)
return "No captions found for video"
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error getting captions for video: {str(e)}")
@staticmethod
def get_video_timestamps(url: str, languages: Optional[List[str]] = None) -> List[str]:
"""Generate timestamps for a YouTube video based on captions."""
if not url:
raise HTTPException(status_code=400, detail="No URL provided")
try:
video_id = YouTubeTools.get_youtube_video_id(url)
if not video_id:
raise HTTPException(status_code=400, detail="Invalid YouTube URL")
except Exception:
raise HTTPException(status_code=400, detail="Error getting video ID from URL")
try:
captions = YouTubeTranscriptApi.get_transcript(video_id, languages=languages or ["en"])
timestamps = []
for line in captions:
start = int(line["start"])
minutes, seconds = divmod(start, 60)
timestamps.append(f"{minutes}:{seconds:02d} - {line['text']}")
return timestamps
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error generating timestamps: {str(e)}")
class YouTubeRequest(BaseModel):
url: str
languages: Optional[List[str]] = None
@app.post("/video-data")
async def get_video_data(request: YouTubeRequest):
"""Endpoint to get video metadata"""
return YouTubeTools.get_video_data(request.url)
@app.post("/video-captions")
async def get_video_captions(request: YouTubeRequest):
"""Endpoint to get video captions"""
return YouTubeTools.get_video_captions(request.url, request.languages)
@app.post("/video-timestamps")
async def get_video_timestamps(request: YouTubeRequest):
"""Endpoint to get video timestamps"""
return YouTubeTools.get_video_timestamps(request.url, request.languages)
if __name__ == "__main__":
# Use environment variable for port, default to 8000 if not set
port = int(os.getenv("PORT", 8000))
host = os.getenv("HOST", "0.0.0.0")
uvicorn.run(app, host=host, port=port)