File size: 3,756 Bytes
d6fb459 196d41e 38fc268 d6fb459 38fc268 d6fb459 38fc268 d6fb459 1a181c7 e8b7d8e 1a181c7 94c404e 1333b51 d6fb459 94c404e 282d87d 12cf3bd 94c404e d6fb459 |
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 |
from pytubefix import YouTube
from pytubefix.cli import on_progress
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import logging as log
import subprocess
import json
from typing import Tuple
import random
# --- Logging Setup ---
log.basicConfig(
level=log.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
# --- FastAPI Setup ---
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Adjust in production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
def index():
return {"message": "Hello, World!"}
# --- Shell Command Runner ---
def cmd(command, check=True, shell=True, capture_output=True, text=True):
log.info(f"Executing command: {command}")
try:
result = subprocess.run(
command,
check=check,
shell=shell,
capture_output=capture_output,
text=text
)
log.info("Command executed successfully.")
return result
except subprocess.CalledProcessError as error:
log.error(f"Command failed: {error}")
raise HTTPException(status_code=500, detail="Token generation failed.")
# --- Token Generator ---
def generate_youtube_token() -> dict:
result = cmd("node test.js")
if not result or not result.stdout:
log.error("No output received from test.js")
raise HTTPException(status_code=500, detail="Invalid token response")
try:
data = json.loads(result.stdout)
log.info(f"Token data received: {data}")
return data
except json.JSONDecodeError as e:
log.error(f"Failed to decode token JSON: {e}")
raise HTTPException(status_code=500, detail="Invalid JSON from token generator")
# --- Token Verifier Function ---
def po_token_verifier() -> Tuple[str, str]:
token_object = generate_youtube_token()
return token_object["visitorData"], token_object["poToken"]
# --- Video Fetch Endpoint ---
@app.get("/api/video/{id}")
def video_id(id: str):
url = f"https://www.youtube.com/watch?v={id}"
log.info(f"Fetching YouTube video for ID: {id}")
proxy_list = [
"198.23.239.134:6540:widxbsal:cnu4fy74afh7",
"207.244.217.165:6712:widxbsal:cnu4fy74afh7",
"107.172.163.27:6543:widxbsal:cnu4fy74afh7",
"23.94.138.75:6349:widxbsal:cnu4fy74afh7",
"216.10.27.159:6837:widxbsal:cnu4fy74afh7",
"136.0.207.84:6661:widxbsal:cnu4fy74afh7",
"64.64.118.149:6732:widxbsal:cnu4fy74afh7",
"142.147.128.93:6593:widxbsal:cnu4fy74afh7",
"104.239.105.125:6655:widxbsal:cnu4fy74afh7",
"173.0.9.70:5653:widxbsal:cnu4fy74afh7"
]
proxy = random.choice(proxy_list)
ip, port, user, password = proxy.split(":")
# Construct the proxy mapping
proxies = {
"http": f"http://{user}:{password}@{ip}:{port}",
"https": f"http://{user}:{password}@{ip}:{port}"
}
try:
yt = YouTube(
url,
use_po_token=True,
po_token_verifier=po_token_verifier(),
proxies=proxies,
)
ys = yt.streams.get_highest_resolution()
if not ys:
log.warning("No suitable video stream found.")
raise HTTPException(status_code=404, detail="No suitable video stream found")
log.info(f"Video fetched: {yt.title}")
return {
"title": yt.title,
"url": ys.url,
"thumbnail": yt.thumbnail_url,
"description": yt.description
}
except Exception as e:
log.error(f"Failed to fetch video info: {e}")
raise HTTPException(status_code=500, detail=str(e)) |