from fastapi import APIRouter, HTTPException, Depends, Query from sqlalchemy.ext.asyncio import AsyncSession from pydantic import BaseModel from typing import Optional from database import get_db 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 @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)): 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)): 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)): 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, 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") urls = await svc.get_watch_urls(stream.stream_key) return { "stream_id": stream.id, "title": stream.title, "status": stream.status.value, "multicast_available": urls["multicast_available"], "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)): 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)): 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, 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") urls = await svc.get_watch_urls(stream_key) return { "stream_id": s.id, "title": s.title, "status": s.status.value, "multicast_available": urls["multicast_available"], "lines": urls["lines"], "qualities": urls["qualities"], }