Youtube-to-mp3-ripper/backend/main.py

from __future__ import annotations

import traceback
from concurrent.futures import ThreadPoolExecutor
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from threading import Lock
from typing import Any
from uuid import uuid4

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from backend.services.converter import (
    ConversionError,
    clean_source_files,
    convert_audio,
    download_audio,
    normalize_format,
    normalize_quality,
    sanitize_filename,
)
from backend.services.youtube import YouTubeServiceError, fetch_video_info, validate_youtube_url


BASE_DIR = Path(__file__).resolve().parent
PROJECT_DIR = BASE_DIR.parent
FRONTEND_DIR = PROJECT_DIR / "frontend"
DOWNLOADS_DIR = BASE_DIR / "downloads"
DOWNLOADS_DIR.mkdir(parents=True, exist_ok=True)

app = FastAPI(title="Local YouTube Audio Converter")
app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")

executor = ThreadPoolExecutor(max_workers=2)
jobs_lock = Lock()
jobs: dict[str, "Job"] = {}


class InfoRequest(BaseModel):
    url: str


class ConvertRequest(BaseModel):
    url: str
    format: str
    quality: str | None = "192"


@dataclass
class Job:
    id: str
    url: str
    format: str
    quality: str
    status: str = "queued"
    message: str = "Queued"
    metadata: dict[str, Any] | None = None
    filename: str | None = None
    file_path: str | None = None
    error: str | None = None
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


@app.get("/")
def index() -> FileResponse:
    return FileResponse(FRONTEND_DIR / "index.html")


@app.post("/api/info")
def api_info(payload: InfoRequest) -> dict[str, Any]:
    try:
        validate_youtube_url(payload.url)
        return fetch_video_info(payload.url)
    except YouTubeServiceError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


@app.post("/api/convert")
def api_convert(payload: ConvertRequest) -> dict[str, str]:
    try:
        valid_url = validate_youtube_url(payload.url)
        format_name = normalize_format(payload.format)
        quality = normalize_quality(format_name, payload.quality)
    except (YouTubeServiceError, ConversionError) as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    job_id = str(uuid4())
    job = Job(id=job_id, url=valid_url, format=format_name, quality=quality)

    with jobs_lock:
        jobs[job_id] = job

    executor.submit(_run_conversion, job_id)
    return {"job_id": job_id}


@app.get("/api/status/{job_id}")
def api_status(job_id: str) -> dict[str, Any]:
    job = _get_job(job_id)
    payload = asdict(job)
    payload.pop("file_path", None)
    if job.status == "complete":
        payload["download_url"] = f"/api/download/{job_id}"
    return payload


@app.get("/api/download/{job_id}")
def api_download(job_id: str) -> FileResponse:
    job = _get_job(job_id)

    if job.status != "complete" or not job.file_path or not job.filename:
        raise HTTPException(status_code=404, detail="This conversion is not ready for download.")

    path = Path(job.file_path)
    if not path.exists() or not path.is_file():
        raise HTTPException(status_code=404, detail="The converted file no longer exists.")

    return FileResponse(path, filename=job.filename, media_type="application/octet-stream")


def _run_conversion(job_id: str) -> None:
    job = _get_job(job_id)
    workdir = DOWNLOADS_DIR / job_id

    try:
        _update_job(job_id, status="validating", message="Validating URL")
        validate_youtube_url(job.url)

        _update_job(job_id, status="fetching_info", message="Fetching video info")
        metadata = fetch_video_info(job.url)
        _update_job(job_id, metadata=metadata)

        base_name = sanitize_filename(metadata.get("title") or "youtube-audio")
        filename = f"{base_name}.{job.format}"
        output_path = workdir / filename

        _update_job(job_id, status="downloading", message="Downloading audio")
        source_path = download_audio(job.url, workdir)

        _update_job(job_id, status="converting", message="Converting")
        final_path = convert_audio(source_path, output_path, job.format, job.quality)
        clean_source_files(workdir)

        _update_job(
            job_id,
            status="complete",
            message="Ready to download",
            filename=filename,
            file_path=str(final_path),
            error=None,
        )
    except (YouTubeServiceError, ConversionError) as exc:
        clean_source_files(workdir)
        _update_job(job_id, status="error", message="Error", error=str(exc))
    except Exception as exc:
        clean_source_files(workdir)
        traceback.print_exc()
        _update_job(job_id, status="error", message="Error", error=f"Unexpected error: {exc}")


def _get_job(job_id: str) -> Job:
    with jobs_lock:
        job = jobs.get(job_id)

    if not job:
        raise HTTPException(status_code=404, detail="Job not found.")
    return job


def _update_job(job_id: str, **changes: Any) -> None:
    with jobs_lock:
        job = jobs.get(job_id)
        if not job:
            return

        for key, value in changes.items():
            setattr(job, key, value)
        job.updated_at = datetime.now(timezone.utc).isoformat()