live/backend/routes/streams.py

from fastapi import APIRouter, HTTPException, Depends, Query, Request
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from typing import Optional
from database import get_db
from routes.auth import require_auth
from services.stream_service import StreamService
from services.media_service import media_service
from services.load_balancer import lb
from config import settings

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


class StreamCreate(BaseModel):
    title: str
    description: Optional[str] = None
    cover_url: Optional[str] = None
    scheduled_at: Optional[str] = None


class StreamUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    cover_url: Optional[str] = None
    scheduled_at: Optional[str] = None


def _get_client_ip(request: Request) -> str | None:
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    client = request.headers.get("X-Real-IP")
    if client:
        return client
    if request.client:
        return request.client.host
    return None


@router.get("")
async def list_streams(db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    streams = await svc.get_all_streams()
    return [
        {
            "id": s.id,
            "title": s.title,
            "description": s.description,
            "cover_url": s.cover_url,
            "stream_key": s.stream_key,
            "status": s.status.value,
            "scheduled_at": str(s.scheduled_at) if s.scheduled_at else None,
            "started_at": str(s.started_at) if s.started_at else None,
            "ended_at": str(s.ended_at) if s.ended_at else None,
            "multicast_enabled": s.multicast_enabled,
            "created_at": str(s.created_at),
        }
        for s in streams
    ]


@router.post("")
async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db),
                        _: bool = Depends(require_auth)):
    svc = StreamService(db)
    stream = await svc.create_stream(
        title=body.title,
        description=body.description,
        cover_url=body.cover_url,
        scheduled_at=body.scheduled_at,
    )
    rtmp_url = media_service.get_rtmp_url(stream.stream_key)
    return {
        "id": stream.id,
        "title": stream.title,
        "stream_key": stream.stream_key,
        "rtmp_url": rtmp_url,
        "status": stream.status.value,
    }


@router.get("/{stream_id}")
async def get_stream(stream_id: int, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    s = await svc.get_stream(stream_id)
    if not s:
        raise HTTPException(status_code=404, detail="Stream not found")
    return {
        "id": s.id,
        "title": s.title,
        "description": s.description,
        "cover_url": s.cover_url,
        "stream_key": s.stream_key,
        "status": s.status.value,
        "scheduled_at": str(s.scheduled_at) if s.scheduled_at else None,
        "started_at": str(s.started_at) if s.started_at else None,
        "ended_at": str(s.ended_at) if s.ended_at else None,
        "multicast_enabled": s.multicast_enabled,
        "created_at": str(s.created_at),
    }


@router.put("/{stream_id}")
async def update_stream(stream_id: int, body: StreamUpdate, db: AsyncSession = Depends(get_db),
                        _: bool = Depends(require_auth)):
    svc = StreamService(db)
    existing = await svc.get_stream(stream_id)
    if not existing:
        raise HTTPException(status_code=404, detail="Stream not found")
    updates = {k: v for k, v in body.model_dump().items() if v is not None}
    if updates:
        await svc.update_stream(stream_id, **updates)
    return {"ok": True}


@router.delete("/{stream_id}")
async def delete_stream(stream_id: int, db: AsyncSession = Depends(get_db),
                        _: bool = Depends(require_auth)):
    svc = StreamService(db)
    existing = await svc.get_stream(stream_id)
    if not existing:
        raise HTTPException(status_code=404, detail="Stream not found")
    await svc.delete_stream(stream_id)
    return {"ok": True}


@router.get("/{stream_id}/watch")
async def get_watch_urls(stream_id: int, request: Request, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    stream = await svc.get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Stream not found")
    client_ip = _get_client_ip(request)
    urls = await svc.get_watch_urls(stream.stream_key, client_ip)
    return {
        "stream_id": stream.id,
        "title": stream.title,
        "status": stream.status.value,
        "multicast_available": urls["multicast_available"],
        "selected_node": urls.get("selected_node"),
        "lines": urls["lines"],
        "qualities": urls["qualities"],
    }


@router.post("/{stream_id}/go-live")
async def stream_go_live(stream_id: int, db: AsyncSession = Depends(get_db),
                         _: bool = Depends(require_auth)):
    svc = StreamService(db)
    stream = await svc.get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Stream not found")
    await svc.set_stream_live(stream.stream_key)
    return {"ok": True, "status": "live"}


@router.post("/{stream_id}/end")
async def stream_end(stream_id: int, db: AsyncSession = Depends(get_db),
                     _: bool = Depends(require_auth)):
    svc = StreamService(db)
    stream = await svc.get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Stream not found")
    await svc.set_stream_ended(stream.stream_key)
    return {"ok": True, "status": "ended"}


@router.get("/detail/{stream_key}")
async def get_stream_by_key(stream_key: str, request: Request, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    s = await svc.get_stream_by_key(stream_key)
    if not s:
        raise HTTPException(status_code=404, detail="Stream not found")
    client_ip = _get_client_ip(request)
    urls = await svc.get_watch_urls(stream_key, client_ip)
    return {
        "stream_id": s.id,
        "title": s.title,
        "status": s.status.value,
        "multicast_available": urls["multicast_available"],
        "lines": urls["lines"],
        "qualities": urls["qualities"],
    }