lofi-app/src/main.py

import logging
import time
from urllib.parse import urljoin

import httpx
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from functools import lru_cache

from src.channels import CHANNELS
from src.config import settings
from src.modules.discovery import find_live_video
from src.modules.stream_extractor import extract_audio_stream

logging.basicConfig(
    level=settings.LOG_LEVEL.upper(),
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

app = FastAPI(title="Lofi Radio Backend", version="0.1.0")

_http_client = httpx.Client(
    timeout=httpx.Timeout(30.0),
    follow_redirects=True,
    headers={
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://www.youtube.com/",
    },
)

# Cache: video_id -> (stream_url, stream_type, expires_at)
_stream_cache: dict[str, tuple[str, str, float]] = {}
CACHE_TTL = 15 * 60  # 15 minutes


def _get_cached_stream(video_id: str) -> tuple[str, str]:
    """Get or refresh cached stream for a video. Returns (url, stream_type)."""
    now = time.time()
    if video_id in _stream_cache:
        url, stype, expires = _stream_cache[video_id]
        if now < expires:
            return url, stype

    info = extract_audio_stream(video_id)
    if not info:
        raise HTTPException(status_code=503, detail="Unable to extract stream")

    _stream_cache[video_id] = (info["url"], info["streamType"], now + CACHE_TTL)
    return info["url"], info["streamType"]


def _evict_expired_cache():
    """Evict expired entries from the stream cache."""
    now = time.time()
    expired = [vid for vid, (_, _, exp) in _stream_cache.items() if now >= exp]
    for vid in expired:
        del _stream_cache[vid]


def _fetch_playlist(playlist_url: str) -> str:
    """Fetch HLS playlist content from YouTube."""
    resp = _http_client.get(playlist_url, timeout=15)
    if resp.status_code != 200:
        raise HTTPException(status_code=502, detail="Failed to fetch playlist")
    return resp.text


def _rewrite_playlist(playlist_content: str, video_id: str) -> str:
    """Rewrite HLS playlist segment URIs to go through server proxy."""
    proxy_base = f"/api/proxy/segment?video={video_id}"
    lines = playlist_content.split("\n")
    result = []
    seg_idx = 0
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            if stripped.startswith("#EXT-X-TARGETDURATION") or stripped.startswith("#EXT-X-MEDIA-SEQUENCE") or stripped.startswith("#EXT-X-DISCONTINUITY") or stripped.startswith("#EXT-X-MAP") or stripped.startswith("#EXT-X-BYTERANGE") or stripped.startswith("#EXTINF") or stripped.startswith("#EXTM3U") or stripped.startswith("#EXT-X-VERSION") or stripped.startswith("#EXT-X-STREAM-INF") or stripped.startswith("#EXT-X-KEY") or stripped.startswith("#EXT-X-ENDLIST") or stripped.startswith("#EXT-X-TIMING") or stripped.startswith("#EXT-X-SKIP"):
                result.append(line)
            else:
                result.append(line)
        else:
            result.append(f"{proxy_base}&idx={seg_idx}")
            seg_idx += 1
    return "\n".join(result)


def _extract_segments(playlist_content: str, playlist_url: str) -> list[str]:
    """Extract segment URLs from HLS playlist."""
    segments = []
    for line in playlist_content.split("\n"):
        stripped = line.strip()
        if stripped and not stripped.startswith("#"):
            seg_url = urljoin(playlist_url, stripped)
            segments.append(seg_url)
    return segments


@app.get("/api/proxy/hls")
def proxy_hls(video: str = Query(...)):
    """Proxy HLS playlist - fetches fresh playlist and rewrites segments."""
    _evict_expired_cache()
    stream_url, stream_type = _get_cached_stream(video)
    if stream_type == "hls":
        content = _fetch_playlist(stream_url)
        rewritten = _rewrite_playlist(content, video)
        return StreamingResponse(
            iter([rewritten]),
            media_type="application/x-mpegURL",
            headers={"Cache-Control": "no-cache"},
        )
    else:
        m3u8 = (
            f"#EXTM3U\n"
            f"#EXT-X-VERSION:3\n"
            f"#EXT-X-TARGETDURATION:30\n"
            f"#EXT-X-MEDIA-SEQUENCE:0\n"
            f"#EXTINF:30.0,\n"
            f"/api/proxy/segment?video={video}&idx=0\n"
        )
        return StreamingResponse(
            iter([m3u8]),
            media_type="application/x-mpegURL",
            headers={"Cache-Control": "no-cache"},
        )


@app.get("/api/proxy/segment")
def proxy_segment(video: str = Query(...), idx: int = Query(...)):
    """Proxy individual HLS segment."""
    stream_url, _ = _get_cached_stream(video)
    content = _fetch_playlist(stream_url)
    segments = _extract_segments(content, stream_url)

    if idx < 0 or idx >= len(segments):
        raise HTTPException(status_code=404, detail="Segment not found")

    seg_url = segments[idx]
    resp = _http_client.get(seg_url, timeout=30)
    if resp.status_code != 200:
        raise HTTPException(status_code=502, detail="Failed to fetch segment")

    return StreamingResponse(
        iter([resp.content]),
        media_type="video/MP2T",
        headers={"Cache-Control": "no-cache"},
    )


def _stream_direct(url: str):
    """Stream direct audio from YouTube."""
    with httpx.stream("GET", url, timeout=300, headers={
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://www.youtube.com/",
    }) as resp:
        for chunk in resp.iter_bytes(64 * 1024):
            yield chunk


@app.get("/api/proxy/audio")
def proxy_audio(video: str = Query(...)):
    """Proxy direct audio stream."""
    _evict_expired_cache()
    stream_url, stream_type = _get_cached_stream(video)
    if stream_type != "direct":
        raise HTTPException(status_code=503, detail="Not a direct stream")
    return StreamingResponse(
        _stream_direct(stream_url),
        media_type="audio/*",
        headers={"Cache-Control": "no-cache"},
    )


app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:5173",
        "http://localhost:5175",
        "http://frontend:80",
    ],
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["*"],
)


@app.get("/api/channels")
def list_channels() -> list[dict]:
    """List all channels with thumbnails."""
    return [
        {
            "id": channel["id"],
            "name": channel["name"],
            "handle": channel.get("handle", ""),
            "description": channel.get("description", ""),
            "isLive": True,
            "videoId": None,
            "thumbnail": channel.get("thumbnail"),
        }
        for channel in CHANNELS
    ]


@app.get("/api/channels/{channel_id}/live")
async def check_channel_live(channel_id: str) -> dict:
    """Check if a specific channel is currently live."""
    channel = next((c for c in CHANNELS if c["id"] == channel_id), None)
    if not channel:
        raise HTTPException(status_code=404, detail="Channel not found")

    video_id, thumbnail = await find_live_video(channel["id"], channel.get("handle", ""))
    return {
        "channelId": channel_id,
        "name": channel["name"],
        "isLive": video_id is not None,
        "videoId": video_id,
        "thumbnail": thumbnail,
    }


@app.get("/api/channel/{channel_id}/latest")
async def get_channel_latest(channel_id: str) -> dict:
    """Find the latest video for a channel."""
    channel = next((c for c in CHANNELS if c["id"] == channel_id), None)
    if not channel:
        raise HTTPException(status_code=404, detail="Channel not found")

    video_id, thumbnail = await find_live_video(channel["id"], channel.get("handle", ""))
    if not video_id:
        raise HTTPException(status_code=404, detail="No videos found for this channel")

    return {
        "channelId": channel_id,
        "videoId": video_id,
        "thumbnail": thumbnail,
    }


@app.get("/api/stream/{video_id}")
def get_stream(video_id: str) -> dict:
    """Get proxied stream URL for a YouTube video."""
    stream_info = extract_audio_stream(video_id)
    if not stream_info:
        raise HTTPException(
            status_code=503, detail="Unable to extract stream for this video"
        )

    # Cache the stream URL (server's IP-bound)
    _stream_cache[video_id] = (
        stream_info["url"],
        stream_info["streamType"],
        time.time() + CACHE_TTL
    )

    if stream_info["streamType"] == "hls":
        stream_info["url"] = f"/api/proxy/hls?video={video_id}"
    else:
        stream_info["url"] = f"/api/proxy/audio?video={video_id}"

    return stream_info


@app.get("/api/now-playing")
async def now_playing() -> dict:
    """Get the current active live stream from any channel."""
    for channel in CHANNELS:
        video_id, thumbnail = await find_live_video(channel["id"], channel.get("handle", ""))
        if video_id:
            stream_info = extract_audio_stream(video_id)
            if stream_info:
                _stream_cache[video_id] = (
                    stream_info["url"],
                    stream_info["streamType"],
                    time.time() + CACHE_TTL
                )
                if stream_info["streamType"] == "hls":
                    stream_info["url"] = f"/api/proxy/hls?video={video_id}"
                else:
                    stream_info["url"] = f"/api/proxy/audio?video={video_id}"
                stream_info["channel"] = {
                    "id": channel["id"],
                    "name": channel["name"],
                    "handle": channel.get("handle", ""),
                    "description": channel.get("description", ""),
                }
                stream_info["thumbnail"] = thumbnail
                return stream_info

    return {"channel": None, "videoId": None, "url": None}