admin
2e4594fa77
Backend: FastAPI + SQLAlchemy + SQLite Frontend: React + Vite + Ant Design (serif Anthropic theme) Media server: MediaMTX with edge node support Load balancing: round-robin / weighted / least-conn Multicast detection with automatic fallback Transcoding support via FFmpeg (configurable quality) hls.js player with line & quality switching Admin panel with password auth Windows compatible scripts
121 lines
4.3 KiB
Python
121 lines
4.3 KiB
Python
from sqlalchemy import select, update, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from models.stream import Stream, StreamStatus
|
|
from services.media_service import media_service
|
|
from services.load_balancer import lb
|
|
from services.multicast_detector import multicast_detector
|
|
from services.transcoder import transcode_mgr
|
|
from config import settings
|
|
import json
|
|
|
|
|
|
class StreamService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def create_stream(self, title: str, description: str | None = None,
|
|
cover_url: str | None = None,
|
|
scheduled_at: str | None = None) -> Stream:
|
|
import uuid
|
|
stream = Stream(
|
|
title=title,
|
|
description=description,
|
|
cover_url=cover_url,
|
|
stream_key=uuid.uuid4().hex,
|
|
status=StreamStatus.pending,
|
|
scheduled_at=scheduled_at,
|
|
)
|
|
self.db.add(stream)
|
|
await self.db.commit()
|
|
await self.db.refresh(stream)
|
|
return stream
|
|
|
|
async def get_all_streams(self):
|
|
result = await self.db.execute(
|
|
select(Stream).order_by(Stream.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def get_stream(self, stream_id: int):
|
|
result = await self.db.execute(
|
|
select(Stream).where(Stream.id == stream_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_stream_by_key(self, stream_key: str):
|
|
result = await self.db.execute(
|
|
select(Stream).where(Stream.stream_key == stream_key)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def update_stream(self, stream_id: int, **kwargs):
|
|
stmt = update(Stream).where(Stream.id == stream_id).values(**kwargs)
|
|
await self.db.execute(stmt)
|
|
await self.db.commit()
|
|
|
|
async def delete_stream(self, stream_id: int):
|
|
stream = await self.get_stream(stream_id)
|
|
if stream:
|
|
transcode_mgr.stop_transcode(stream.stream_key)
|
|
await self.db.delete(stream)
|
|
await self.db.commit()
|
|
|
|
async def set_stream_live(self, stream_key: str):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if stream:
|
|
from datetime import datetime
|
|
await self.update_stream(
|
|
stream.id,
|
|
status=StreamStatus.live,
|
|
started_at=datetime.utcnow(),
|
|
)
|
|
assigned = [n["id"] for n in lb.get_enabled_nodes()]
|
|
await self.update_stream(
|
|
stream.id,
|
|
assigned_nodes=json.dumps(assigned),
|
|
multicast_enabled=multicast_detector.is_available(),
|
|
)
|
|
rtmp_url = media_service.get_rtmp_url(stream_key)
|
|
transcode_mgr.start_transcode(stream_key, rtmp_url)
|
|
|
|
async def set_stream_ended(self, stream_key: str):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if stream:
|
|
from datetime import datetime
|
|
await self.update_stream(
|
|
stream.id,
|
|
status=StreamStatus.ended,
|
|
ended_at=datetime.utcnow(),
|
|
)
|
|
transcode_mgr.stop_transcode(stream_key)
|
|
|
|
async def get_watch_urls(self, stream_key: str):
|
|
nodes = lb.get_enabled_nodes()
|
|
lines = []
|
|
for node in nodes:
|
|
line_info = {
|
|
"node_id": node["id"],
|
|
"node_name": node["name"],
|
|
"weight": node["weight"],
|
|
"qualities": [],
|
|
}
|
|
base_hls = f"/hls/{stream_key}"
|
|
line_info["type"] = "hls"
|
|
line_info["url"] = f"{base_hls}/index.m3u8"
|
|
|
|
for quality in settings.qualities:
|
|
label = quality["label"]
|
|
line_info["qualities"].append({
|
|
"label": label,
|
|
"width": quality["width"],
|
|
"height": quality["height"],
|
|
"url": f"{base_hls}/{label}/index.m3u8",
|
|
})
|
|
lines.append(line_info)
|
|
|
|
return {
|
|
"lines": lines,
|
|
"qualities": [{"label": q["label"], "width": q["width"], "height": q["height"]} for q in settings.qualities],
|
|
"multicast_available": multicast_detector.is_available(),
|
|
}
|