"""FastAPI application serving a small YouTube download interface."""
import asyncio
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
HTML = """
YouTube Downloader
YouTube Downloader
"""
app = FastAPI()
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}
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() -> None:
with yt_dlp.YoutubeDL(ydl_opts) as downloader:
downloader.download([url])
try:
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]
except Exception as exc:
cleanup_tmpdir(temp_dir)
raise HTTPException(status_code=500, detail=str(exc)) from exc
return FileResponse(
media_path,
filename=media_path.name,
background=BackgroundTask(cleanup_tmpdir, temp_dir),
)