From 2e4594fa773794189ef411e13ed80c6ec46dcba4 Mon Sep 17 00:00:00 2001 From: admin Date: Fri, 7 Aug 2026 15:02:07 +0800 Subject: [PATCH] Initial commit: LiveStream Platform 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 --- .gitignore | 10 ++ backend/config.py | 37 ++++ backend/database.py | 21 +++ backend/main.py | 66 ++++++++ backend/models/__init__.py | 3 + backend/models/stream.py | 31 ++++ backend/requirements.txt | 8 + backend/routes/__init__.py | 5 + backend/routes/auth.py | 31 ++++ backend/routes/nodes.py | 62 +++++++ backend/routes/streams.py | 163 ++++++++++++++++++ backend/run.py | 4 + backend/services/__init__.py | 0 backend/services/load_balancer.py | 50 ++++++ backend/services/media_service.py | 48 ++++++ backend/services/multicast_detector.py | 64 +++++++ backend/services/stream_service.py | 120 +++++++++++++ backend/services/transcoder.py | 63 +++++++ frontend/index.html | 15 ++ frontend/package.json | 27 +++ frontend/src/App.tsx | 52 ++++++ frontend/src/admin/AdminLayout.tsx | 93 ++++++++++ frontend/src/admin/LoginPage.tsx | 61 +++++++ frontend/src/admin/pages/Dashboard.tsx | 96 +++++++++++ frontend/src/admin/pages/NodeManager.tsx | 144 ++++++++++++++++ frontend/src/admin/pages/Settings.tsx | 84 +++++++++ frontend/src/admin/pages/StreamForm.tsx | 128 ++++++++++++++ frontend/src/admin/pages/StreamList.tsx | 123 ++++++++++++++ frontend/src/api/auth.ts | 18 ++ frontend/src/api/client.ts | 30 ++++ frontend/src/api/streams.ts | 92 ++++++++++ frontend/src/global.css | 8 + frontend/src/main.tsx | 10 ++ frontend/src/theme/anthropic.ts | 62 +++++++ frontend/src/viewer/pages/WatchPage.tsx | 207 +++++++++++++++++++++++ frontend/tsconfig.json | 21 +++ frontend/vite.config.ts | 19 +++ mediamtx/edge_template.yml | 16 ++ mediamtx/mediamtx.yml | 20 +++ mediamtx/scripts/check_multicast.bat | 23 +++ mediamtx/scripts/start_edge.bat | 23 +++ mediamtx/scripts/start_origin.bat | 3 + start.sh | 35 ++++ 43 files changed, 2196 insertions(+) create mode 100644 .gitignore create mode 100644 backend/config.py create mode 100644 backend/database.py create mode 100644 backend/main.py create mode 100644 backend/models/__init__.py create mode 100644 backend/models/stream.py create mode 100644 backend/requirements.txt create mode 100644 backend/routes/__init__.py create mode 100644 backend/routes/auth.py create mode 100644 backend/routes/nodes.py create mode 100644 backend/routes/streams.py create mode 100644 backend/run.py create mode 100644 backend/services/__init__.py create mode 100644 backend/services/load_balancer.py create mode 100644 backend/services/media_service.py create mode 100644 backend/services/multicast_detector.py create mode 100644 backend/services/stream_service.py create mode 100644 backend/services/transcoder.py create mode 100644 frontend/index.html create mode 100644 frontend/package.json create mode 100644 frontend/src/App.tsx create mode 100644 frontend/src/admin/AdminLayout.tsx create mode 100644 frontend/src/admin/LoginPage.tsx create mode 100644 frontend/src/admin/pages/Dashboard.tsx create mode 100644 frontend/src/admin/pages/NodeManager.tsx create mode 100644 frontend/src/admin/pages/Settings.tsx create mode 100644 frontend/src/admin/pages/StreamForm.tsx create mode 100644 frontend/src/admin/pages/StreamList.tsx create mode 100644 frontend/src/api/auth.ts create mode 100644 frontend/src/api/client.ts create mode 100644 frontend/src/api/streams.ts create mode 100644 frontend/src/global.css create mode 100644 frontend/src/main.tsx create mode 100644 frontend/src/theme/anthropic.ts create mode 100644 frontend/src/viewer/pages/WatchPage.tsx create mode 100644 frontend/tsconfig.json create mode 100644 frontend/vite.config.ts create mode 100644 mediamtx/edge_template.yml create mode 100644 mediamtx/mediamtx.yml create mode 100644 mediamtx/scripts/check_multicast.bat create mode 100644 mediamtx/scripts/start_edge.bat create mode 100644 mediamtx/scripts/start_origin.bat create mode 100755 start.sh diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..80e493e --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +node_modules/ +dist/ +__pycache__/ +*.pyc +*.db +venv/ +*.log +mediamtx/mediamtx.tar.gz +mediamtx/mediamtx +package-lock.json diff --git a/backend/config.py b/backend/config.py new file mode 100644 index 0000000..88c287b --- /dev/null +++ b/backend/config.py @@ -0,0 +1,37 @@ +from pydantic_settings import BaseSettings +from typing import Literal + + +class Settings(BaseSettings): + database_url: str = "sqlite+aiosqlite:///./data.db" + admin_password: str = "admin123" + + media_origin_host: str = "127.0.0.1" + media_origin_rtmp_port: int = 1935 + media_origin_api_port: int = 9997 + media_origin_hls_port: int = 8888 + + edge_nodes: list[dict] = [ + {"id": "edge-1", "name": "Edge Node 1", "host": "127.0.0.1", + "hls_port": 8889, "api_port": 9998, "weight": 100, "enabled": True}, + {"id": "edge-2", "name": "Edge Node 2", "host": "127.0.0.1", + "hls_port": 8890, "api_port": 9999, "weight": 50, "enabled": True}, + ] + + load_balance_strategy: Literal["round-robin", "weighted", "least-conn"] = "weighted" + + qualities: list[dict] = [ + {"label": "1080p", "width": 1920, "height": 1080, "bitrate": "5000k"}, + {"label": "720p", "width": 1280, "height": 720, "bitrate": "2500k"}, + {"label": "480p", "width": 854, "height": 480, "bitrate": "1000k"}, + {"label": "360p", "width": 640, "height": 360, "bitrate": "600k"}, + ] + + multicast_enabled: bool = True + multicast_address_base: str = "239.0.0.1" + multicast_port_base: int = 5000 + + model_config = {"env_prefix": "LS_"} + + +settings = Settings() diff --git a/backend/database.py b/backend/database.py new file mode 100644 index 0000000..1c415e7 --- /dev/null +++ b/backend/database.py @@ -0,0 +1,21 @@ +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession +from sqlalchemy.orm import DeclarativeBase + +from config import settings + +engine = create_async_engine(settings.database_url, echo=False) +session_factory = async_sessionmaker(engine, expire_on_commit=False) + + +class Base(DeclarativeBase): + pass + + +async def get_db() -> AsyncSession: + async with session_factory() as session: + yield session + + +async def init_db(): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) diff --git a/backend/main.py b/backend/main.py new file mode 100644 index 0000000..a83b80b --- /dev/null +++ b/backend/main.py @@ -0,0 +1,66 @@ +import os +from contextlib import asynccontextmanager +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse +from starlette.responses import RedirectResponse +from starlette.background import BackgroundTask +import httpx +from database import init_db +from services.transcoder import transcode_mgr +from routes import streams_router, nodes_router, auth_router +from config import settings + + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + yield + transcode_mgr.stop_all() + + +app = FastAPI(title="LiveStream Platform", lifespan=lifespan) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +app.include_router(streams_router) +app.include_router(nodes_router) +app.include_router(auth_router) + + +@app.get("/api/health") +async def health(): + from services.multicast_detector import multicast_detector + return { + "status": "ok", + "multicast_available": multicast_detector.is_available(), + } + + +@app.get("/hls/{path:path}") +async def serve_hls(path: str): + hls_root = "/tmp/hls" + fs_path = os.path.join(hls_root, path) + if os.path.isfile(fs_path): + return FileResponse(fs_path) + + target = f"http://127.0.0.1:{settings.media_origin_hls_port}/{path}" + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(10)) as client: + req = client.build_request("GET", target) + resp = await client.send(req, stream=True) + from starlette.responses import StreamingResponse + return StreamingResponse( + resp.aiter_bytes(), + status_code=resp.status_code, + headers=dict(resp.headers), + media_type=resp.headers.get("content-type"), + ) + except Exception: + raise HTTPException(status_code=404) diff --git a/backend/models/__init__.py b/backend/models/__init__.py new file mode 100644 index 0000000..1201730 --- /dev/null +++ b/backend/models/__init__.py @@ -0,0 +1,3 @@ +from .stream import Stream + +__all__ = ["Stream"] diff --git a/backend/models/stream.py b/backend/models/stream.py new file mode 100644 index 0000000..836fbfb --- /dev/null +++ b/backend/models/stream.py @@ -0,0 +1,31 @@ +import uuid +from datetime import datetime +from sqlalchemy import String, Text, DateTime, Integer, Boolean, Enum +from sqlalchemy.orm import Mapped, mapped_column +from database import Base +import enum + + +class StreamStatus(str, enum.Enum): + pending = "pending" + live = "live" + ended = "ended" + cancelled = "cancelled" + + +class Stream(Base): + __tablename__ = "streams" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + title: Mapped[str] = mapped_column(String(200), nullable=False) + description: Mapped[str | None] = mapped_column(Text, nullable=True) + cover_url: Mapped[str | None] = mapped_column(String(500), nullable=True) + stream_key: Mapped[str] = mapped_column(String(64), unique=True, default=lambda: uuid.uuid4().hex) + status: Mapped[StreamStatus] = mapped_column(Enum(StreamStatus), default=StreamStatus.pending) + scheduled_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + assigned_nodes: Mapped[str | None] = mapped_column(Text, nullable=True) + multicast_enabled: Mapped[bool] = mapped_column(Boolean, default=False) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) diff --git a/backend/requirements.txt b/backend/requirements.txt new file mode 100644 index 0000000..8a18169 --- /dev/null +++ b/backend/requirements.txt @@ -0,0 +1,8 @@ +fastapi>=0.110.0 +uvicorn[standard]>=0.29.0 +sqlalchemy[asyncio]>=2.0.30 +aiosqlite>=0.20.0 +pydantic>=2.7.0 +pydantic-settings>=2.2.0 +python-multipart>=0.0.9 +httpx>=0.27.0 diff --git a/backend/routes/__init__.py b/backend/routes/__init__.py new file mode 100644 index 0000000..f353480 --- /dev/null +++ b/backend/routes/__init__.py @@ -0,0 +1,5 @@ +from .streams import router as streams_router +from .nodes import router as nodes_router +from .auth import router as auth_router + +__all__ = ["streams_router", "nodes_router", "auth_router"] diff --git a/backend/routes/auth.py b/backend/routes/auth.py new file mode 100644 index 0000000..a1710a7 --- /dev/null +++ b/backend/routes/auth.py @@ -0,0 +1,31 @@ +import hashlib +from fastapi import APIRouter, HTTPException, Depends +from fastapi.responses import JSONResponse +from config import settings + +router = APIRouter(prefix="/api/auth", tags=["auth"]) + +_tokens: set[str] = set() + + +def hash_password(pw: str) -> str: + return hashlib.sha256(f"{pw}:livestream-salt".encode()).hexdigest() + + +ADMIN_HASH = hash_password(settings.admin_password) + + +@router.post("/login") +async def login(password: str): + if hash_password(password) != ADMIN_HASH: + raise HTTPException(status_code=401, detail="Invalid password") + token = hashlib.sha256(f"{password}:token".encode()).hexdigest() + _tokens.add(token) + return {"token": token} + + +@router.post("/verify") +async def verify(token: str): + if token not in _tokens: + raise HTTPException(status_code=401, detail="Invalid token") + return {"valid": True} diff --git a/backend/routes/nodes.py b/backend/routes/nodes.py new file mode 100644 index 0000000..bdf7f12 --- /dev/null +++ b/backend/routes/nodes.py @@ -0,0 +1,62 @@ +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel +from config import settings, Settings +from services.load_balancer import lb +from services.multicast_detector import multicast_detector + +router = APIRouter(prefix="/api/nodes", tags=["nodes"]) + + +class NodeUpdate(BaseModel): + weight: int + enabled: bool | None = None + + +@router.get("") +async def list_nodes(): + nodes = [] + for n in settings.edge_nodes: + nodes.append({ + "id": n["id"], + "name": n["name"], + "host": n["host"], + "hls_port": n["hls_port"], + "api_port": n["api_port"], + "weight": n["weight"], + "enabled": n.get("enabled", True), + }) + return nodes + + +@router.put("/{node_id}") +async def update_node(node_id: str, body: NodeUpdate): + found = False + for n in settings.edge_nodes: + if n["id"] == node_id: + n["weight"] = body.weight + if body.enabled is not None: + n["enabled"] = body.enabled + found = True + break + if not found: + raise HTTPException(status_code=404, detail="Node not found") + return {"ok": True} + + +@router.get("/strategy") +async def get_strategy(): + return { + "strategy": settings.load_balance_strategy, + "multicast_enabled": settings.multicast_enabled, + "multicast_available": multicast_detector.is_available(), + } + + +@router.put("/strategy") +async def update_strategy(strategy: str, multicast_enabled: bool | None = None): + global settings + if strategy in ("round-robin", "weighted", "least-conn"): + settings.load_balance_strategy = strategy + if multicast_enabled is not None: + settings.multicast_enabled = multicast_enabled + return {"ok": True} diff --git a/backend/routes/streams.py b/backend/routes/streams.py new file mode 100644 index 0000000..bc32b82 --- /dev/null +++ b/backend/routes/streams.py @@ -0,0 +1,163 @@ +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"], + } diff --git a/backend/run.py b/backend/run.py new file mode 100644 index 0000000..a2dbc41 --- /dev/null +++ b/backend/run.py @@ -0,0 +1,4 @@ +import uvicorn + +if __name__ == "__main__": + uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=False) diff --git a/backend/services/__init__.py b/backend/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/services/load_balancer.py b/backend/services/load_balancer.py new file mode 100644 index 0000000..411b56d --- /dev/null +++ b/backend/services/load_balancer.py @@ -0,0 +1,50 @@ +import json +import asyncio +import subprocess +from config import settings + + +class LoadBalancer: + def __init__(self): + self._rr_index = 0 + self._conn_counts: dict[str, int] = {n["id"]: 0 for n in settings.edge_nodes} + + def get_enabled_nodes(self): + return [n for n in settings.edge_nodes if n.get("enabled", True)] + + def select_nodes(self, strategy: str | None = None): + nodes = self.get_enabled_nodes() + strategy = strategy or settings.load_balance_strategy + + if strategy == "round-robin": + if not nodes: + return [] + chosen = nodes[self._rr_index % len(nodes)] + self._rr_index += 1 + return [chosen] + elif strategy == "weighted": + total = sum(n["weight"] for n in nodes) + if total == 0: + return nodes + import random + r = random.uniform(0, total) + cumulative = 0 + for n in nodes: + cumulative += n["weight"] + if r <= cumulative: + return [n] + return [nodes[-1]] + elif strategy == "least-conn": + sorted_nodes = sorted(nodes, key=lambda n: self._conn_counts.get(n["id"], 0)) + return [sorted_nodes[0]] + return nodes + + def increment_connection(self, node_id: str): + self._conn_counts[node_id] = self._conn_counts.get(node_id, 0) + 1 + + def decrement_connection(self, node_id: str): + if self._conn_counts.get(node_id, 0) > 0: + self._conn_counts[node_id] -= 1 + + +lb = LoadBalancer() diff --git a/backend/services/media_service.py b/backend/services/media_service.py new file mode 100644 index 0000000..469ab8e --- /dev/null +++ b/backend/services/media_service.py @@ -0,0 +1,48 @@ +import httpx +import asyncio +from config import settings + + +class MediaMTXService: + @staticmethod + def get_rtmp_url(stream_key: str) -> str: + return f"rtmp://{settings.media_origin_host}:{settings.media_origin_rtmp_port}/live/{stream_key}" + + @staticmethod + def get_hls_url(node: dict, stream_key: str) -> str: + return f"http://{node['host']}:{node['hls_port']}/{stream_key}/index.m3u8" + + @staticmethod + async def get_origin_status(stream_key: str) -> dict | None: + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(5)) as client: + resp = await client.get( + f"http://{settings.media_origin_host}:{settings.media_origin_api_port}/v3/paths/list" + ) + if resp.status_code == 200: + data = resp.json() + for item in data.get("items", []): + if item.get("name") == stream_key: + return {"ready": item.get("ready", False), "source_ready": item.get("sourceReady", False)} + except Exception: + return None + return None + + @staticmethod + async def get_edge_status(node: dict, stream_key: str) -> dict | None: + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(3)) as client: + resp = await client.get( + f"http://{node['host']}:{node['api_port']}/v3/paths/list" + ) + if resp.status_code == 200: + data = resp.json() + for item in data.get("items", []): + if item.get("name") == stream_key: + return {"ready": item.get("ready", False), "readers": item.get("readersCount", 0)} + except Exception: + return None + return None + + +media_service = MediaMTXService() diff --git a/backend/services/multicast_detector.py b/backend/services/multicast_detector.py new file mode 100644 index 0000000..a2914ae --- /dev/null +++ b/backend/services/multicast_detector.py @@ -0,0 +1,64 @@ +import socket +import struct +import asyncio +from config import settings + + +def check_multicast_reachable(mcast_addr: str = "239.0.0.1", port: int = 9999, timeout: float = 2.0) -> bool: + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.settimeout(timeout) + ttl = struct.pack("b", 1) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl) + + mreq = struct.pack("4sL", socket.inet_aton(mcast_addr), socket.INADDR_ANY) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + + sock.sendto(b"ping", (mcast_addr, port)) + try: + data, addr = sock.recvfrom(1024) + sock.close() + return True + except socket.timeout: + sock.close() + return True + except Exception: + return False + + +class MulticastDetector: + def __init__(self): + self._multicast_available: bool | None = None + + def is_available(self) -> bool: + if self._multicast_available is not None: + return self._multicast_available + + if not settings.multicast_enabled: + self._multicast_available = False + return False + + self._multicast_available = check_multicast_reachable() + return self._multicast_available + + def get_stream_url(self, stream_key: str, node: dict, quality: dict | None = None): + use_multicast = self.is_available() + host = node["host"] + + if use_multicast: + hash_val = abs(hash(stream_key)) % 255 + mcast_addr = f"{settings.multicast_address_base.rsplit('.', 1)[0]}.{hash_val + 1}" + mcast_port = settings.multicast_port_base + (abs(hash(stream_key)) % 1000) + return { + "type": "multicast", + "url": f"udp://@{mcast_addr}:{mcast_port}", + "unicast_fallback": f"http://{host}:{node['hls_port']}/{stream_key}/index.m3u8", + } + + return { + "type": "hls", + "url": f"http://{host}:{node['hls_port']}/{stream_key}/index.m3u8", + } + + +multicast_detector = MulticastDetector() diff --git a/backend/services/stream_service.py b/backend/services/stream_service.py new file mode 100644 index 0000000..ff7c8cb --- /dev/null +++ b/backend/services/stream_service.py @@ -0,0 +1,120 @@ +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(), + } diff --git a/backend/services/transcoder.py b/backend/services/transcoder.py new file mode 100644 index 0000000..a0791cd --- /dev/null +++ b/backend/services/transcoder.py @@ -0,0 +1,63 @@ +import asyncio +import subprocess +import os +from config import settings + + +class TranscodeManager: + _processes: dict[str, list[subprocess.Popen]] = {} + + def start_transcode(self, stream_key: str, input_url: str): + if stream_key in self._processes: + return + + procs = [] + for quality in settings.qualities: + label = quality["label"] + width = quality["width"] + height = quality["height"] + bitrate = quality["bitrate"] + hls_dir = f"/tmp/hls/{stream_key}/{label}" + os.makedirs(hls_dir, exist_ok=True) + cmd = [ + "ffmpeg", "-y", + "-i", input_url, + "-c:v", "libx264", "-preset", "ultrafast", + "-b:v", bitrate, + "-s", f"{width}x{height}", + "-c:a", "aac", "-b:a", "128k", + "-f", "hls", + "-hls_time", "4", + "-hls_list_size", "6", + "-hls_flags", "delete_segments", + "-hls_segment_filename", f"{hls_dir}/segment_%03d.ts", + f"{hls_dir}/index.m3u8", + ] + try: + proc = subprocess.Popen( + cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ) + procs.append(proc) + except FileNotFoundError: + pass + + self._processes[stream_key] = procs + + def stop_transcode(self, stream_key: str): + procs = self._processes.pop(stream_key, []) + for proc in procs: + try: + proc.terminate() + proc.wait(timeout=5) + except Exception: + try: + proc.kill() + except Exception: + pass + + def stop_all(self): + for key in list(self._processes.keys()): + self.stop_transcode(key) + + +transcode_mgr = TranscodeManager() diff --git a/frontend/index.html b/frontend/index.html new file mode 100644 index 0000000..bdb7cf8 --- /dev/null +++ b/frontend/index.html @@ -0,0 +1,15 @@ + + + + + + LiveStream Platform + + + + + +
+ + + diff --git a/frontend/package.json b/frontend/package.json new file mode 100644 index 0000000..c0716f0 --- /dev/null +++ b/frontend/package.json @@ -0,0 +1,27 @@ +{ + "name": "livestream-frontend", + "private": true, + "version": "1.0.0", + "type": "module", + "scripts": { + "dev": "vite --host 0.0.0.0 --port 5173", + "build": "tsc -b && vite build", + "preview": "vite preview --host 0.0.0.0" + }, + "dependencies": { + "@ant-design/icons": "^5.4.0", + "antd": "^5.21.0", + "axios": "^1.7.7", + "hls.js": "^1.5.14", + "react": "^18.3.1", + "react-dom": "^18.3.1", + "react-router-dom": "^6.26.2" + }, + "devDependencies": { + "@types/react": "^18.3.8", + "@types/react-dom": "^18.3.0", + "@vitejs/plugin-react": "^4.3.1", + "typescript": "^5.6.2", + "vite": "^5.4.7" + } +} diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx new file mode 100644 index 0000000..224a870 --- /dev/null +++ b/frontend/src/App.tsx @@ -0,0 +1,52 @@ +import { useState, useEffect } from 'react' +import { BrowserRouter, Routes, Route, Navigate } from 'react-router-dom' +import { ConfigProvider } from 'antd' +import theme from './theme/anthropic' +import AdminLayout from './admin/AdminLayout' +import LoginPage from './admin/LoginPage' +import Dashboard from './admin/pages/Dashboard' +import StreamList from './admin/pages/StreamList' +import StreamForm from './admin/pages/StreamForm' +import NodeManager from './admin/pages/NodeManager' +import SettingsPage from './admin/pages/Settings' +import WatchPage from './viewer/pages/WatchPage' +import { verifyToken, getToken } from './api/auth' + +function AdminGuard({ children }: { children: React.ReactNode }) { + const [authed, setAuthed] = useState(null) + + useEffect(() => { + const token = getToken() + if (!token) { setAuthed(false); return } + verifyToken().then(setAuthed) + }, []) + + if (authed === null) return null + if (!authed) return + return <>{children} +} + +export default function App() { + return ( + + + + } /> + } /> + + + + }> + } /> + } /> + } /> + } /> + } /> + + } /> + + + + ) +} diff --git a/frontend/src/admin/AdminLayout.tsx b/frontend/src/admin/AdminLayout.tsx new file mode 100644 index 0000000..f861d14 --- /dev/null +++ b/frontend/src/admin/AdminLayout.tsx @@ -0,0 +1,93 @@ +import { useState } from 'react' +import { Outlet, useNavigate, useLocation, Link } from 'react-router-dom' +import { Layout, Menu, Button, Typography } from 'antd' +import { + DashboardOutlined, + VideoCameraOutlined, + PlusOutlined, + CloudServerOutlined, + SettingOutlined, + LogoutOutlined, +} from '@ant-design/icons' +import { clearToken } from '../api/client' + +const { Sider, Content, Header } = Layout +const { Text } = Typography + +const menuItems = [ + { key: '/admin', icon: , label: 'Dashboard' }, + { key: '/admin/streams', icon: , label: 'Streams' }, + { key: '/admin/streams/new', icon: , label: 'New Stream' }, + { key: '/admin/nodes', icon: , label: 'Edge Nodes' }, + { key: '/admin/settings', icon: , label: 'Settings' }, +] + +export default function AdminLayout() { + const [collapsed, setCollapsed] = useState(false) + const navigate = useNavigate() + const location = useLocation() + + const selectedKey = menuItems.find( + (item) => location.pathname === item.key || (location.pathname === '/' && item.key === '/admin') + )?.key || '/admin' + + const handleLogout = () => { + clearToken() + navigate('/admin/login') + } + + return ( + + +
+ + {collapsed ? 'LS' : 'LiveStream'} + +
+ navigate(key)} + style={{ border: 'none', marginTop: 8 }} + /> + + +
+ +
+ + + + + + ) +} diff --git a/frontend/src/admin/LoginPage.tsx b/frontend/src/admin/LoginPage.tsx new file mode 100644 index 0000000..f0a6191 --- /dev/null +++ b/frontend/src/admin/LoginPage.tsx @@ -0,0 +1,61 @@ +import { useState } from 'react' +import { useNavigate } from 'react-router-dom' +import { Card, Form, Input, Button, Typography, message, Space } from 'antd' +import { LockOutlined } from '@ant-design/icons' +import { login } from '../api/auth' + +const { Title, Text } = Typography + +export default function LoginPage() { + const [loading, setLoading] = useState(false) + const navigate = useNavigate() + + const handleSubmit = async (values: { password: string }) => { + setLoading(true) + try { + await login(values.password) + message.success('Welcome back') + navigate('/admin') + } catch { + message.error('Invalid password') + } finally { + setLoading(false) + } + } + + return ( +
+ + +
+ LiveStream + Admin Panel +
+
+ + } + placeholder="Password" + size="large" + /> + + + + +
+ + +
+ ) +} diff --git a/frontend/src/admin/pages/Dashboard.tsx b/frontend/src/admin/pages/Dashboard.tsx new file mode 100644 index 0000000..7c8c99c --- /dev/null +++ b/frontend/src/admin/pages/Dashboard.tsx @@ -0,0 +1,96 @@ +import { useState, useEffect } from 'react' +import { Row, Col, Card, Statistic, Typography, Table, Tag, Space } from 'antd' +import { VideoCameraOutlined, EyeOutlined, CloudServerOutlined, SignalFilled } from '@ant-design/icons' +import { fetchStreams, type StreamInfo } from '../../api/streams' + +const { Title } = Typography + +export default function Dashboard() { + const [streams, setStreams] = useState([]) + const [loading, setLoading] = useState(true) + + useEffect(() => { + loadData() + const timer = setInterval(loadData, 10000) + return () => clearInterval(timer) + }, []) + + async function loadData() { + try { + const data = await fetchStreams() + setStreams(data) + } catch { /* ignore */ } + setLoading(false) + } + + const liveCount = streams.filter((s) => s.status === 'live').length + const pendingCount = streams.filter((s) => s.status === 'pending').length + const endedCount = streams.filter((s) => s.status === 'ended').length + + const statusMap: Record = { + pending: { color: 'blue', label: 'Pending' }, + live: { color: 'green', label: 'Live' }, + ended: { color: 'default', label: 'Ended' }, + cancelled: { color: 'red', label: 'Cancelled' }, + } + + const recentColumns = [ + { title: 'Title', dataIndex: 'title', key: 'title', ellipsis: true }, + { + title: 'Status', dataIndex: 'status', key: 'status', + render: (status: string) => { + const s = statusMap[status] || { color: 'default', label: status } + return {s.label} + }, + }, + { title: 'Created', dataIndex: 'created_at', key: 'created_at', render: (v: string) => v?.slice(0, 16) }, + ] + + return ( +
+ Dashboard + + + + } + loading={loading} + /> + + + + + } + loading={loading} + /> + + + + + } + loading={loading} + /> + + + + + + + + ) +} diff --git a/frontend/src/admin/pages/NodeManager.tsx b/frontend/src/admin/pages/NodeManager.tsx new file mode 100644 index 0000000..c8def56 --- /dev/null +++ b/frontend/src/admin/pages/NodeManager.tsx @@ -0,0 +1,144 @@ +import { useState, useEffect } from 'react' +import { + Card, Table, InputNumber, Switch, Typography, message, Space, Tag, Statistic, Row, Col, Button, +} from 'antd' +import { CheckCircleOutlined, CloseCircleOutlined, ReloadOutlined } from '@ant-design/icons' +import api from '../../api/client' + +const { Title, Text } = Typography + +interface NodeInfo { + id: string + name: string + host: string + hls_port: number + api_port: number + weight: number + enabled: boolean +} + +interface StrategyInfo { + strategy: string + multicast_enabled: boolean + multicast_available: boolean +} + +export default function NodeManager() { + const [nodes, setNodes] = useState([]) + const [strategy, setStrategy] = useState(null) + const [loading, setLoading] = useState(true) + + useEffect(() => { loadData() }, []) + + async function loadData() { + setLoading(true) + try { + const [nodesRes, strategyRes] = await Promise.all([ + api.get('/nodes'), + api.get('/nodes/strategy'), + ]) + setNodes(nodesRes.data) + setStrategy(strategyRes.data) + } catch { message.error('Failed to load') } + setLoading(false) + } + + async function updateWeight(nodeId: string, weight: number) { + try { + await api.put(`/nodes/${nodeId}`, { weight }) + message.success('Updated') + } catch { message.error('Failed') } + } + + async function toggleNode(nodeId: string, enabled: boolean) { + const node = nodes.find((n) => n.id === nodeId) + if (!node) return + try { + await api.put(`/nodes/${nodeId}`, { weight: node.weight, enabled }) + setNodes((prev) => prev.map((n) => (n.id === nodeId ? { ...n, enabled } : n))) + message.success('Updated') + } catch { message.error('Failed') } + } + + const columns = [ + { title: 'Name', dataIndex: 'name', key: 'name' }, + { title: 'Host', dataIndex: 'host', key: 'host' }, + { title: 'HLS Port', dataIndex: 'hls_port', key: 'hls_port', width: 100 }, + { title: 'API Port', dataIndex: 'api_port', key: 'api_port', width: 100 }, + { + title: 'Weight', dataIndex: 'weight', key: 'weight', width: 140, + render: (_: number, record: NodeInfo) => ( + { + const v = val || 0 + setNodes((prev) => prev.map((n) => (n.id === record.id ? { ...n, weight: v } : n))) + updateWeight(record.id, v) + }} + size="small" + style={{ width: 90 }} + /> + ), + }, + { + title: 'Enabled', dataIndex: 'enabled', key: 'enabled', width: 90, + render: (_: any, record: NodeInfo) => ( + toggleNode(record.id, val)} + size="small" + /> + ), + }, + { + title: 'Health', key: 'health', width: 100, + render: () => } color="success">Online, + }, + ] + + return ( +
+ Edge Nodes + + {strategy && ( + +
+ + + + + + + + + + + + + + + + )} + + } onClick={loadData}>Refresh} + > +
+ + + ) +} diff --git a/frontend/src/admin/pages/Settings.tsx b/frontend/src/admin/pages/Settings.tsx new file mode 100644 index 0000000..f8600be --- /dev/null +++ b/frontend/src/admin/pages/Settings.tsx @@ -0,0 +1,84 @@ +import { useState, useEffect } from 'react' +import { + Card, Select, Switch, Typography, message, Space, Descriptions, Button, +} from 'antd' +import api from '../../api/client' + +const { Title } = Typography + +const strategies = [ + { value: 'round-robin', label: 'Round Robin' }, + { value: 'weighted', label: 'Weighted' }, + { value: 'least-conn', label: 'Least Connections' }, +] + +export default function SettingsPage() { + const [strategy, setStrategy] = useState('weighted') + const [multicastEnabled, setMulticastEnabled] = useState(true) + const [multicastAvailable, setMulticastAvailable] = useState(false) + const [saving, setSaving] = useState(false) + + useEffect(() => { + loadSettings() + }, []) + + async function loadSettings() { + try { + const { data } = await api.get('/nodes/strategy') + setStrategy(data.strategy) + setMulticastEnabled(data.multicast_enabled) + setMulticastAvailable(data.multicast_available) + } catch { /* ignore */ } + } + + async function saveStrategy(val: string) { + setSaving(true) + try { + await api.put(`/nodes/strategy?strategy=${val}&multicast_enabled=${multicastEnabled}`) + setStrategy(val) + message.success('Strategy updated') + } catch { message.error('Failed') } + setSaving(false) + } + + async function toggleMulticast(val: boolean) { + setMulticastEnabled(val) + try { + await api.put(`/nodes/strategy?strategy=${strategy}&multicast_enabled=${val}`) + message.success('Multicast setting updated') + } catch { message.error('Failed') } + } + + return ( +
+ Settings + + + + + + + +