You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
171 lines
6.4 KiB
171 lines
6.4 KiB
"""FastAPI application serving a small YouTube download interface."""
|
|
|
|
import asyncio
|
|
import re
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import yt_dlp
|
|
from fastapi import FastAPI, Form, HTTPException, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
|
|
from starlette.background import BackgroundTask
|
|
|
|
|
|
_FILENAME_ALLOWED = re.compile(r"[^\w\s.,'()\-]", re.UNICODE)
|
|
_WHITESPACE = re.compile(r"\s+")
|
|
|
|
|
|
HTML = """<!doctype html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>YouTube Downloader</title>
|
|
<style>
|
|
body { align-items: center; background: #101114; color: #f4f4f5; display: flex;
|
|
font-family: system-ui, sans-serif; justify-content: center; min-height: 100vh; margin: 0; }
|
|
main { max-width: 36rem; padding: 2rem; width: 100%; }
|
|
input { box-sizing: border-box; font-size: 1rem; padding: .8rem; width: 100%; }
|
|
button { cursor: pointer; font-size: 1rem; margin: 1rem .5rem 0 0; padding: .75rem 1rem; }
|
|
#status { color: #fca5a5; min-height: 1.5rem; padding-top: 1rem; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<main>
|
|
<h1>YouTube Downloader</h1>
|
|
<input id="url" type="url" placeholder="Paste YouTube URL…" autocomplete="off">
|
|
<div>
|
|
<button type="button" onclick="downloadFile('mp3')">Download MP3</button>
|
|
<button type="button" onclick="downloadFile('video')">Download Video</button>
|
|
</div>
|
|
<div id="status" role="status"></div>
|
|
</main>
|
|
<script>
|
|
async function downloadFile(mode) {
|
|
const url = document.getElementById('url').value.trim();
|
|
const status = document.getElementById('status');
|
|
if (!url) {
|
|
status.textContent = 'Please paste a YouTube URL.';
|
|
return;
|
|
}
|
|
|
|
status.textContent = 'Preparing your download…';
|
|
const data = new FormData();
|
|
data.append('url', url);
|
|
data.append('mode', mode);
|
|
try {
|
|
const response = await fetch('/download', { method: 'POST', body: data });
|
|
const contentType = response.headers.get('content-type') || '';
|
|
if (!response.ok || contentType.includes('application/json')) {
|
|
const error = await response.json();
|
|
throw new Error(error.message || 'The download could not be prepared.');
|
|
}
|
|
|
|
const disposition = response.headers.get('content-disposition') || '';
|
|
const match = /filename\\*?=(?:UTF-8''|")?([^";]+)/i.exec(disposition);
|
|
const filename = match ? decodeURIComponent(match[1].replace(/"/g, '')) : 'download';
|
|
const blob = await response.blob();
|
|
const objectUrl = URL.createObjectURL(blob);
|
|
const link = document.createElement('a');
|
|
link.href = objectUrl;
|
|
link.download = filename;
|
|
document.body.appendChild(link);
|
|
link.click();
|
|
link.remove();
|
|
setTimeout(() => URL.revokeObjectURL(objectUrl), 0);
|
|
status.textContent = '';
|
|
} catch (error) {
|
|
status.textContent = error.message || 'The download could not be prepared.';
|
|
}
|
|
}
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
def sanitize_filename(title: str, fallback_id: str) -> str:
|
|
"""Return a safe, human-readable download filename stem."""
|
|
cleaned = _FILENAME_ALLOWED.sub("", title or "")
|
|
cleaned = _WHITESPACE.sub(" ", cleaned).strip()
|
|
return cleaned or fallback_id
|
|
|
|
|
|
def cleanup_tmpdir(path: str) -> None:
|
|
"""Remove a request's temporary download directory after streaming."""
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(_: Request, exc: HTTPException) -> JSONResponse:
|
|
"""Expose errors in the format expected by the browser client."""
|
|
message = exc.detail if isinstance(exc.detail, str) else "The request could not be processed."
|
|
return JSONResponse(status_code=exc.status_code, content={"message": message})
|
|
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(_: Request, __: RequestValidationError) -> JSONResponse:
|
|
"""Return malformed form submissions as a client-readable error."""
|
|
return JSONResponse(status_code=400, content={"message": "A URL and download mode are required."})
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def home() -> str:
|
|
"""Serve the single-page downloader interface."""
|
|
return HTML
|
|
|
|
|
|
@app.post("/download")
|
|
async def download(
|
|
url: str = Form(...), mode: Literal["mp3", "video"] = Form(...)
|
|
) -> FileResponse:
|
|
"""Download and stream a requested audio or video file."""
|
|
if not url.strip() or not url.lower().startswith("http"):
|
|
raise HTTPException(status_code=400, detail="Please provide a valid URL starting with http.")
|
|
|
|
temp_dir = tempfile.mkdtemp(prefix="youtube-download-")
|
|
output_template = str(Path(temp_dir) / "%(title)s.%(ext)s")
|
|
ydl_opts: dict[str, object] = {"outtmpl": output_template, "noplaylist": True}
|
|
if mode == "mp3":
|
|
ydl_opts.update(
|
|
{
|
|
"format": "bestaudio/best",
|
|
"postprocessors": [
|
|
{
|
|
"key": "FFmpegExtractAudio",
|
|
"preferredcodec": "mp3",
|
|
"preferredquality": "0",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
else:
|
|
ydl_opts.update({"format": "bestvideo+bestaudio/best", "merge_output_format": "mp4"})
|
|
|
|
def fetch_media() -> dict[str, object]:
|
|
with yt_dlp.YoutubeDL(ydl_opts) as downloader:
|
|
return downloader.extract_info(url, download=True)
|
|
|
|
try:
|
|
info = await asyncio.get_running_loop().run_in_executor(None, fetch_media)
|
|
downloads = [path for path in Path(temp_dir).iterdir() if path.is_file()]
|
|
if len(downloads) != 1:
|
|
raise RuntimeError("The downloaded file could not be identified.")
|
|
media_path = downloads[0]
|
|
title = str(info.get("title") or "")
|
|
video_id = str(info.get("id") or "download")
|
|
display_name = sanitize_filename(title, video_id) + media_path.suffix
|
|
except Exception as exc:
|
|
cleanup_tmpdir(temp_dir)
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
|
|
return FileResponse(
|
|
media_path,
|
|
filename=display_name,
|
|
background=BackgroundTask(cleanup_tmpdir, temp_dir),
|
|
)
|
|
|