Spaces:
Running
Running
File size: 3,434 Bytes
61bd138 | 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 | """Static file server + YouTube audio download.
Auto-installs yt-dlp if missing. User just runs: python start_server.py
"""
import http.server
import json
import subprocess
import sys
from pathlib import Path
# Auto-install yt-dlp if missing
try:
import yt_dlp
except ImportError:
print("Installing yt-dlp...")
subprocess.check_call([
sys.executable, "-m", "pip", "install", "yt-dlp", "-q",
"--break-system-packages",
])
import yt_dlp
DOWNLOADS = Path(__file__).resolve().parent / "downloads"
DOWNLOADS.mkdir(exist_ok=True)
def extract_video_id(url):
"""Extract YouTube video ID from URL without downloading."""
with yt_dlp.YoutubeDL({"quiet": True, "skip_download": True}) as ydl:
info = ydl.extract_info(url, download=False)
return info.get("id"), info.get("title", "Unknown")
def download_audio(url):
"""Download YouTube audio as mp3, return (filepath, title). Reuses cached files."""
video_id, title = extract_video_id(url)
# Reuse cached file if already downloaded
mp3 = DOWNLOADS / f"{video_id}.mp3"
if mp3.exists():
print(f"Cache hit: {mp3.name}")
return mp3, title
opts = {
"format": "bestaudio/best",
"postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "mp3"}],
"outtmpl": str(DOWNLOADS / "%(id)s.%(ext)s"),
"noplaylist": True,
"quiet": True,
}
with yt_dlp.YoutubeDL(opts) as ydl:
ydl.download([url])
if not mp3.exists():
for f in DOWNLOADS.glob(f"{video_id}.*"):
mp3 = f
break
return mp3, title
class AppHandler(http.server.SimpleHTTPRequestHandler):
def end_headers(self):
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.send_header("Cross-Origin-Opener-Policy", "same-origin")
self.send_header("Cross-Origin-Embedder-Policy", "credentialless")
super().end_headers()
def do_OPTIONS(self):
self.send_response(204)
self.end_headers()
def do_POST(self):
if self.path == "/api/yt/download":
self._handle_download()
else:
self.send_error(404)
def _handle_download(self):
try:
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length)) if length else {}
url = body.get("url", "").strip()
if not url:
return self._json(400, {"error": "url required"})
mp3, title = download_audio(url)
self._json(200, {
"audio_url": f"/downloads/{mp3.name}",
"title": title,
})
except Exception as e:
self._json(500, {"error": str(e)})
def _json(self, code, data):
body = json.dumps(data).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8000
print(f"Robot Dance Party — http://localhost:{port}")
with http.server.HTTPServer(("", port), AppHandler) as s:
s.serve_forever()
|