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

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()

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()

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()

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(),
+        }

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()