platform/services/media/app/api/episodes.py

"""Episode listing, detail, and streaming endpoints."""

from __future__ import annotations

import os
import uuid
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Header, Query
from fastapi.responses import StreamingResponse, Response
from sqlalchemy import select, and_
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.requests import Request

from app.api.deps import get_user_id, get_db_session
from app.config import CONTENT_TYPES
from app.models import Episode, Show, Progress

router = APIRouter(prefix="/api/episodes", tags=["episodes"])


# ── Named convenience endpoints (must be before /{episode_id}) ──

@router.get("/recent")
async def recent_episodes(
    limit: int = Query(20, ge=1, le=100),
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    """Recently played episodes ordered by last_played_at."""
    stmt = (
        select(Episode, Show, Progress)
        .join(Progress, Progress.episode_id == Episode.id)
        .join(Show, Show.id == Episode.show_id)
        .where(Progress.user_id == user_id)
        .order_by(Progress.last_played_at.desc())
        .limit(limit)
    )
    result = await db.execute(stmt)
    return [_format(ep, show, prog) for ep, show, prog in result.all()]


@router.get("/in-progress")
async def in_progress_episodes(
    limit: int = Query(20, ge=1, le=100),
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    """Episodes with progress > 0 and not completed."""
    stmt = (
        select(Episode, Show, Progress)
        .join(Progress, Progress.episode_id == Episode.id)
        .join(Show, Show.id == Episode.show_id)
        .where(
            Progress.user_id == user_id,
            Progress.position_seconds > 0,
            Progress.is_completed == False,  # noqa: E712
        )
        .order_by(Progress.last_played_at.desc())
        .limit(limit)
    )
    result = await db.execute(stmt)
    return [_format(ep, show, prog) for ep, show, prog in result.all()]


# ── List ──

@router.get("")
async def list_episodes(
    show_id: Optional[str] = Query(None),
    status: Optional[str] = Query(None),
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    """List episodes with optional filters: show_id, status (unplayed|in_progress|completed)."""
    stmt = (
        select(Episode, Show, Progress)
        .join(Show, Show.id == Episode.show_id)
        .outerjoin(Progress, and_(Progress.episode_id == Episode.id, Progress.user_id == user_id))
        .where(Episode.user_id == user_id)
    )

    if show_id:
        stmt = stmt.where(Episode.show_id == uuid.UUID(show_id))

    if status == "unplayed":
        stmt = stmt.where(Progress.id == None)  # noqa: E711
    elif status == "in_progress":
        stmt = stmt.where(Progress.position_seconds > 0, Progress.is_completed == False)  # noqa: E712
    elif status == "completed":
        stmt = stmt.where(Progress.is_completed == True)  # noqa: E712

    stmt = stmt.order_by(Episode.published_at.desc().nullslast()).offset(offset).limit(limit)

    result = await db.execute(stmt)
    return [_format(ep, show, prog) for ep, show, prog in result.all()]


# ── Detail ──

@router.get("/{episode_id}")
async def get_episode(
    episode_id: str,
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    """Episode detail with progress."""
    stmt = (
        select(Episode, Show, Progress)
        .join(Show, Show.id == Episode.show_id)
        .outerjoin(Progress, and_(Progress.episode_id == Episode.id, Progress.user_id == user_id))
        .where(Episode.id == uuid.UUID(episode_id), Episode.user_id == user_id)
    )
    row = (await db.execute(stmt)).first()
    if not row:
        raise HTTPException(404, "Episode not found")

    ep, show, prog = row
    return _format(ep, show, prog)


# ── Stream local audio ──

@router.get("/{episode_id}/stream")
async def stream_episode(
    episode_id: str,
    request: Request,
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    """Stream local audio file with HTTP Range support for seeking."""
    ep = await db.get(Episode, uuid.UUID(episode_id))
    if not ep or ep.user_id != user_id:
        raise HTTPException(404, "Episode not found")

    show = await db.get(Show, ep.show_id)
    if not show or show.show_type != "local":
        raise HTTPException(400, "Streaming only available for local episodes")

    file_path = ep.audio_url
    if not file_path or not os.path.isfile(file_path):
        raise HTTPException(404, "Audio file not found")

    file_size = os.path.getsize(file_path)
    ext = os.path.splitext(file_path)[1].lower()
    content_type = CONTENT_TYPES.get(ext, "application/octet-stream")

    range_header = request.headers.get("range")
    return _range_response(file_path, file_size, content_type, range_header)


# ── Helpers ──

def _format(ep: Episode, show: Show, prog: Optional[Progress]) -> dict:
    return {
        "id": str(ep.id),
        "show_id": str(ep.show_id),
        "title": ep.title,
        "description": ep.description,
        "audio_url": ep.audio_url,
        "duration_seconds": ep.duration_seconds,
        "file_size_bytes": ep.file_size_bytes,
        "published_at": ep.published_at.isoformat() if ep.published_at else None,
        "artwork_url": ep.artwork_url or show.artwork_url,
        "show": {
            "id": str(show.id),
            "title": show.title,
            "author": show.author,
            "artwork_url": show.artwork_url,
            "show_type": show.show_type,
        },
        "progress": {
            "position_seconds": prog.position_seconds,
            "duration_seconds": prog.duration_seconds,
            "is_completed": prog.is_completed,
            "playback_speed": prog.playback_speed,
            "last_played_at": prog.last_played_at.isoformat() if prog.last_played_at else None,
        } if prog else None,
    }


def _range_response(file_path: str, file_size: int, content_type: str, range_header: Optional[str]):
    """Build a streaming response with optional Range support for seeking."""
    if range_header and range_header.startswith("bytes="):
        range_spec = range_header[6:]
        parts = range_spec.split("-")
        start = int(parts[0]) if parts[0] else 0
        end = int(parts[1]) if len(parts) > 1 and parts[1] else file_size - 1
        end = min(end, file_size - 1)

        if start >= file_size:
            return Response(status_code=416, headers={"Content-Range": f"bytes */{file_size}"})

        length = end - start + 1

        def iter_range():
            with open(file_path, "rb") as f:
                f.seek(start)
                remaining = length
                while remaining > 0:
                    chunk = f.read(min(65536, remaining))
                    if not chunk:
                        break
                    remaining -= len(chunk)
                    yield chunk

        return StreamingResponse(
            iter_range(),
            status_code=206,
            media_type=content_type,
            headers={
                "Content-Range": f"bytes {start}-{end}/{file_size}",
                "Content-Length": str(length),
                "Accept-Ranges": "bytes",
            },
        )

    def iter_file():
        with open(file_path, "rb") as f:
            while chunk := f.read(65536):
                yield chunk

    return StreamingResponse(
        iter_file(),
        media_type=content_type,
        headers={
            "Content-Length": str(file_size),
            "Accept-Ranges": "bytes",
        },
    )