Spaces:
Build error
Build error
File size: 2,896 Bytes
4564216 8e27f17 2515173 8e27f17 9cb3fae 8e27f17 2515173 8e27f17 2515173 9cb3fae 2515173 4564216 2515173 9cb3fae 8e27f17 2515173 4564216 8e27f17 |
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 |
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
# --- 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}")
try:
yt = YouTube(
url,
use_po_token=True,
po_token_verifier=po_token_verifier
)
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)) |