"""FastAPI application serving a small YouTube download interface."""
import asyncio
import re
import shutil
import tempfile
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Literal
from uuid import uuid4
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 = """
YouTube Downloader
YouTube Downloader
"""
app = FastAPI()
@dataclass
class JobState:
"""Transient state for a browser-requested download."""
status: Literal["downloading", "finished", "error"]
percent: float = 0
speed: float | None = None
downloaded_bytes: int = 0
total_bytes: int | None = None
temp_dir: str = ""
file_path: str | None = None
filename: str | None = None
error_message: str | None = None
mode: Literal["mp3", "video"] = "mp3"
created_at: float = 0
_JOB_TTL_SECONDS = 30 * 60
_jobs: dict[str, JobState] = {}
_jobs_lock = threading.Lock()
_pending_jobs: set[asyncio.Future[None]] = set()
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)
def cleanup_job(job_id: str) -> None:
"""Remove a delivered job and its temporary download directory."""
with _jobs_lock:
job = _jobs.pop(job_id, None)
if job:
cleanup_tmpdir(job.temp_dir)
def cleanup_stale_jobs() -> None:
"""Best-effort removal of abandoned download jobs."""
cutoff = time.monotonic() - _JOB_TTL_SECONDS
with _jobs_lock:
stale_jobs = [
_jobs.pop(job_id)
for job_id, job in list(_jobs.items())
if job.created_at < cutoff
]
for job in stale_jobs:
cleanup_tmpdir(job.temp_dir)
@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(...)
) -> JSONResponse:
"""Start a download job and return its identifier immediately."""
if not url.strip() or not url.lower().startswith("http"):
raise HTTPException(status_code=400, detail="Please provide a valid URL starting with http.")
cleanup_stale_jobs()
job_id = str(uuid4())
temp_dir = tempfile.mkdtemp(prefix="youtube-download-")
output_template = str(Path(temp_dir) / "%(title)s.%(ext)s")
with _jobs_lock:
_jobs[job_id] = JobState(
status="downloading", temp_dir=temp_dir, mode=mode, created_at=time.monotonic()
)
def progress_hook(progress: dict[str, object]) -> None:
"""Copy yt-dlp progress updates into the job visible to the browser."""
with _jobs_lock:
job = _jobs.get(job_id)
if not job:
return
if progress.get("status") == "error":
job.status = "error"
job.error_message = "The download could not be prepared."
return
if progress.get("status") != "downloading":
return
downloaded = progress.get("downloaded_bytes")
total = progress.get("total_bytes") or progress.get("total_bytes_estimate")
speed = progress.get("speed")
job.downloaded_bytes = int(downloaded) if isinstance(downloaded, (int, float)) else 0
job.total_bytes = int(total) if isinstance(total, (int, float)) else None
job.speed = float(speed) if isinstance(speed, (int, float)) else None
if job.total_bytes:
job.percent = min(100, job.downloaded_bytes / job.total_bytes * 100)
ydl_opts: dict[str, object] = {
"outtmpl": output_template,
"noplaylist": True,
"progress_hooks": [progress_hook],
}
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 run_job() -> None:
try:
with yt_dlp.YoutubeDL(ydl_opts) as downloader:
info = downloader.extract_info(url, download=True)
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
with _jobs_lock:
job = _jobs.get(job_id)
if job:
job.status = "finished"
job.percent = 100
job.file_path = str(media_path)
job.filename = display_name
except Exception as exc:
cleanup_tmpdir(temp_dir)
with _jobs_lock:
job = _jobs.get(job_id)
if job:
job.status = "error"
job.error_message = str(exc)
future = asyncio.get_running_loop().run_in_executor(None, run_job)
_pending_jobs.add(future)
future.add_done_callback(_pending_jobs.discard)
return JSONResponse({"job_id": job_id})
@app.get("/progress/{job_id}")
async def progress(job_id: str) -> JSONResponse:
"""Return the browser-safe state of an active download job."""
with _jobs_lock:
job = _jobs.get(job_id)
if not job:
raise HTTPException(status_code=404, detail="Download job not found.")
payload = {
"status": job.status,
"percent": job.percent,
"speed": job.speed,
"downloaded_bytes": job.downloaded_bytes,
"total_bytes": job.total_bytes,
"error": job.error_message,
}
return JSONResponse(payload)
@app.get("/download/{job_id}/file")
async def download_file(job_id: str) -> FileResponse:
"""Stream a completed job's file, cleaning its state after delivery."""
with _jobs_lock:
job = _jobs.get(job_id)
if not job:
raise HTTPException(status_code=404, detail="Download job not found.")
if job.status != "finished" or not job.file_path or not job.filename:
raise HTTPException(status_code=409, detail="The download is not ready yet.")
file_path = job.file_path
filename = job.filename
return FileResponse(
file_path,
filename=filename,
background=BackgroundTask(cleanup_job, job_id),
)