Spaces:
Sleeping
Sleeping
File size: 3,227 Bytes
11421c5 6eaacab 11421c5 6eaacab 1ea7a2c 11421c5 6eaacab 1ea7a2c 6eaacab 11421c5 1ea7a2c 11421c5 1ea7a2c 11421c5 1ea7a2c 11421c5 1ea7a2c 6eaacab 11421c5 6eaacab 11421c5 6eaacab 11421c5 1ea7a2c 11421c5 1ea7a2c 11421c5 1ea7a2c 11421c5 1ea7a2c 11421c5 6eaacab | 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 | import os
from fastapi import FastAPI, HTTPException
import yt_dlp
from supabase import create_client, Client
app = FastAPI()
# Your Supabase Credentials
SUPABASE_URL = "https://pfgmficmhmefvlotriuf.supabase.co"
SUPABASE_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InBmZ21maWNtaG1lZnZsb3RyaXVmIiwicm9sZSI6ImFub24iLCJpYXQiOjE3Njk4OTg3MDgsImV4cCI6MjA4NTQ3NDcwOH0.V_u7LMG7LWOzvGOPFrZdUGgESIB9Y6clu7UCfF1Xng0"
# Initialize Supabase
try:
supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
except Exception as e:
print(f"CRITICAL ERROR: Could not connect to Supabase: {e}")
@app.get("/")
def home():
return {"status": "Vortex TOT API is LIVE", "database": "Connected"}
@app.get("/extract")
def extract_video_data(url: str):
# 1. Check Supabase Cache First
try:
cached_data = supabase.table("video_cache").select("json_data").eq("video_url", url).execute()
if cached_data.data:
# If found, return instantly!
return cached_data.data[0]['json_data']
except Exception as e:
print(f"Cache check error: {e}")
# 2. If not in cache, run yt-dlp
ydl_opts = {
'quiet': True,
'no_warnings': True,
'skip_download': True,
'extract_flat': False,
}
try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(url, download=False)
# --- Logic for TikTok Slideshows ---
if 'entries' in info and (info.get('extractor') == 'TikTok' or 'tiktok' in url.lower()):
response_data = {
"type": "slideshow",
"title": info.get('title', 'TikTok Slideshow'),
"images": [e.get('url') for e in info['entries'] if e.get('url')],
"audio": info.get('url'),
"source": "TikTok"
}
# --- Logic for Standard Video/Audio ---
else:
formats = []
for f in info.get('formats', []):
if f.get('url'):
formats.append({
"id": f.get('format_id'),
"ext": f.get('ext'),
"res": f.get('resolution'),
"note": f.get('format_note'),
"url": f.get('url')
})
response_data = {
"type": "video",
"title": info.get('title'),
"thumbnail": info.get('thumbnail'),
"duration": info.get('duration'),
"formats": formats,
"source": info.get('extractor')
}
# 3. Save new result to Supabase
try:
supabase.table("video_cache").insert({
"video_url": url,
"json_data": response_data
}).execute()
except Exception as e:
print(f"Failed to cache: {e}")
return response_data
except Exception as e:
raise HTTPException(status_code=400, detail=str(e)) |