led2/lightsync/api/audio.py

"""Audio upload, file listing, waveform, and beat analysis endpoints."""
import asyncio
import json
import re
import shutil
import sys
from pathlib import Path
from fastapi import APIRouter, HTTPException, Request, UploadFile, File
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from lightsync.audio.waveform import extract_peaks
from lightsync.audio.beats import analyze_async

router = APIRouter()

SUPPORTED_EXTENSIONS = {".mp3", ".wav", ".flac", ".ogg"}
SHOWS_DIR = Path("/app/shows")
PROGRESS_RE = re.compile(r'\[download\]\s+([\d.]+)%')


class YouTubeFetchRequest(BaseModel):
    url: str


@router.get("/files")
async def list_audio_files():
    """List audio files available in the shows directory."""
    SHOWS_DIR.mkdir(parents=True, exist_ok=True)
    files = sorted(
        p.name for p in SHOWS_DIR.iterdir()
        if p.is_file() and p.suffix.lower() in SUPPORTED_EXTENSIONS
    )
    return {"files": files, "directory": str(SHOWS_DIR)}


@router.post("/upload")
async def upload_audio(request: Request, file: UploadFile = File(...)):
    """Upload an audio file to the shows directory and run beat analysis.

    After upload, librosa beat detection runs in a thread and the result is
    cached in app.state.beats keyed by the absolute file path.
    """
    suffix = Path(file.filename).suffix.lower()
    if suffix not in SUPPORTED_EXTENSIONS:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}"
        )

    SHOWS_DIR.mkdir(parents=True, exist_ok=True)
    dest = SHOWS_DIR / file.filename

    with dest.open("wb") as f:
        shutil.copyfileobj(file.file, f)

    # Run beat analysis in background — cache result keyed by filepath
    dest_str = str(dest)
    try:
        beat_data = await analyze_async(dest_str)
        if not hasattr(request.app.state, "beats") or request.app.state.beats is None:
            request.app.state.beats = {}
        request.app.state.beats[dest_str] = beat_data
    except Exception as exc:
        # Beat analysis failure is non-fatal — upload still succeeds
        beat_data = None
        import logging
        logging.getLogger(__name__).warning("Beat analysis failed for %s: %s", dest_str, exc)

    return {
        "status": "uploaded",
        "path": dest_str,
        "filename": file.filename,
        "beats": beat_data,
    }


@router.get("/beats")
async def get_beats(request: Request, path: str):
    """Return cached beat data for a loaded file.

    Query params:
        path: absolute path to the audio file (must have been uploaded/analyzed)

    Returns:
        {"tempo": float, "beats": [float, ...], "file": str}
    """
    beats_cache: dict = getattr(request.app.state, "beats", None) or {}

    if path not in beats_cache:
        # Not cached — run analysis now (may be slow)
        p = Path(path)
        if not p.exists():
            raise HTTPException(status_code=404, detail=f"File not found: {path}")
        if p.suffix.lower() not in SUPPORTED_EXTENSIONS:
            raise HTTPException(status_code=400, detail=f"Unsupported format: {p.suffix}")
        try:
            beat_data = await analyze_async(path)
            if not hasattr(request.app.state, "beats") or request.app.state.beats is None:
                request.app.state.beats = {}
            request.app.state.beats[path] = beat_data
        except Exception as exc:
            raise HTTPException(status_code=500, detail=f"Beat analysis failed: {exc}") from exc
    else:
        beat_data = beats_cache[path]

    return {**beat_data, "file": path}


@router.get("/waveform")
async def get_waveform(path: str, peaks: int = 1000):
    """Get waveform peak data for a file (AUD-05).

    Query params:
        path: absolute path to audio file
        peaks: number of peak samples (default 1000, max 5000)
    """
    p = Path(path)
    if not p.exists():
        raise HTTPException(status_code=404, detail=f"File not found: {path}")

    num_peaks = min(max(peaks, 100), 5000)

    try:
        data = await asyncio.to_thread(extract_peaks, path, num_peaks)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Waveform extraction failed: {e}")

    return {"peaks": data, "count": len(data), "file": path}


@router.post("/fetch")
async def fetch_youtube(request: Request, body: YouTubeFetchRequest):
    """Download audio from YouTube URL via yt-dlp. Streams SSE progress events."""

    async def event_stream():
        filename = None
        try:
            proc = await asyncio.create_subprocess_exec(
                sys.executable, '-m', 'yt_dlp',
                '--extract-audio',
                '--audio-format', 'mp3',
                '--audio-quality', '0',
                '--newline',
                '--no-playlist',
                '-o', str(SHOWS_DIR / 'yt_%(id)s.%(ext)s'),
                body.url,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
            )

            async for line_bytes in proc.stdout:
                line = line_bytes.decode('utf-8', errors='replace').rstrip()
                # Progress line
                m = PROGRESS_RE.search(line)
                if m:
                    pct = float(m.group(1))
                    yield f"data: {json.dumps({'type': 'progress', 'pct': pct})}\n\n"
                # Destination line — captures actual output filename
                if 'Destination:' in line:
                    filename = line.split('Destination:')[-1].strip()

            await proc.wait()

            if proc.returncode != 0:
                yield f"data: {json.dumps({'type': 'error', 'msg': f'yt-dlp exited with code {proc.returncode}'})}\n\n"
                return

            if not filename:
                # Fallback: find most recently created mp3 in SHOWS_DIR
                mp3s = sorted(SHOWS_DIR.glob('yt_*.mp3'), key=lambda p: p.stat().st_mtime, reverse=True)
                if mp3s:
                    filename = str(mp3s[0])

            if filename:
                # Run beat analysis in background and cache
                try:
                    beat_data = await analyze_async(filename)
                    if not hasattr(request.app.state, 'beats') or request.app.state.beats is None:
                        request.app.state.beats = {}
                    request.app.state.beats[filename] = beat_data
                except Exception:
                    pass  # Non-fatal — beats can be loaded later

                fname = Path(filename).name
                yield f"data: {json.dumps({'type': 'done', 'path': filename, 'filename': fname})}\n\n"
            else:
                yield f"data: {json.dumps({'type': 'error', 'msg': 'Download completed but output file not found'})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'msg': str(e)})}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")