Spaces:
Paused
Paused
File size: 4,303 Bytes
66e27d9 3b8a142 66e27d9 3b8a142 66e27d9 3b8a142 66e27d9 3b8a142 66e27d9 3b8a142 66e27d9 3b8a142 66e27d9 | 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 | import os
import time
import asyncio
import aiohttp
import subprocess
import urllib.parse
from fastapi import FastAPI, BackgroundTasks, HTTPException
from fastapi.responses import FileResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
# ==========================================
# 🚀 API SETUP & DOCUMENTATION
# ==========================================
app = FastAPI(
title="SILENT TECH Media API",
description="High-Speed API for YouTube Searching, MP4 fetching, and ultra-fast FFmpeg MP3 Extraction.",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
def cleanup_files(files: list):
for f in files:
if os.path.exists(f):
try:
os.remove(f)
except:
pass
# ==========================================
# 📡 ENDPOINTS (WHITE-LABELED)
# ==========================================
@app.get("/")
def read_root():
return {"status": "Online", "creator": "SILENT TECH", "message": "Visit /docs for the API Documentation!"}
@app.get("/api/search")
async def search_yt(query: str):
"""🔍 Search YouTube and return metadata."""
async with aiohttp.ClientSession() as session:
url = f"https://apis.davidcyril.name.ng/play?query={urllib.parse.quote(query)}"
async with session.get(url) as resp:
data = await resp.json()
if not data.get("status"):
raise HTTPException(status_code=404, detail="Not found")
# 🥷 THE NINJA MOVE: Overwrite the creator name!
data["creator"] = "SILENT TECH"
return data
@app.get("/api/ytmp4")
async def get_ytmp4(url: str):
"""🎬 Get a direct MP4 Download URL."""
async with aiohttp.ClientSession() as session:
api_url = f"https://apis.davidcyril.name.ng/download/ytmp4?url={urllib.parse.quote(url)}"
async with session.get(api_url) as resp:
data = await resp.json()
if not data.get("success"):
raise HTTPException(status_code=400, detail="Failed to fetch video link")
# 🥷 THE NINJA MOVE: Overwrite the creator name!
data["creator"] = "SILENT TECH"
return data
@app.get("/api/ytmp3")
async def get_ytmp3(url: str, background_tasks: BackgroundTasks):
"""🎵 Download MP4, convert to MP3 instantly via FFmpeg, and return the audio file."""
timestamp = int(time.time() * 1000)
temp_vid = f"temp_vid_{timestamp}.mp4"
temp_aud = f"temp_aud_{timestamp}.mp3"
background_tasks.add_task(cleanup_files, [temp_vid, temp_aud])
try:
async with aiohttp.ClientSession() as session:
api_url = f"https://apis.davidcyril.name.ng/download/ytmp4?url={urllib.parse.quote(url)}"
async with session.get(api_url) as resp:
data = await resp.json()
if not data.get("success") or not data.get("result", {}).get("download_url"):
raise HTTPException(status_code=400, detail="Failed to fetch stream")
download_url = data["result"]["download_url"]
title = data["result"].get("title", "Silent_Tech_Audio").replace("/", "_")
async with session.get(download_url) as video_resp:
with open(temp_vid, 'wb') as f:
while True:
chunk = await video_resp.content.read(2 * 1024 * 1024)
if not chunk: break
f.write(chunk)
command = [
"ffmpeg", "-y", "-i", temp_vid,
"-vn", "-acodec", "libmp3lame", "-q:a", "2",
temp_aud
]
process = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if process.returncode != 0:
raise Exception("FFmpeg audio extraction failed.")
return FileResponse(
path=temp_aud,
media_type="audio/mpeg",
filename=f"{title}.mp3"
)
except Exception as e:
background_tasks.add_task(cleanup_files, [temp_vid, temp_aud])
raise HTTPException(status_code=500, detail=str(e)) |