From 20cd99efa47f2c2b21328b30f05fbfd29bfef29a Mon Sep 17 00:00:00 2001 From: admin Date: Wed, 12 Aug 2026 21:14:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E7=9B=B4=E6=92=AD=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E7=9B=91=E6=8E=A7=E4=B8=8E=E8=BE=B9=E7=BC=98=E4=BB=A3?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/main.py | 46 ++++++++++++++++ backend/requirements.txt | 1 + backend/routes/nodes.py | 28 ++++++++-- backend/routes/streams.py | 24 ++++++--- backend/services/load_balancer.py | 6 +++ backend/services/node_store.py | 27 ++++++++++ backend/services/stream_service.py | 13 ++++- edge-agent/agent/agent.go | 67 +++++++++++++++++++----- edge-agent/agent/heartbeat.go | 45 ++++++++++++++++ edge-agent/agent/mediamtx.go | 58 +++++++++++++++++++- edge-agent/agent/monitor.go | 31 ++++++++--- edge-agent/install-autostart.ps1 | 14 +++++ edge-agent/install-autostart.sh | 31 +++++++++++ edge-agent/main.go | 17 +++--- edge-agent/tray_windows.go | 7 +++ frontend/src/admin/pages/NodeManager.tsx | 20 +++++-- frontend/src/admin/pages/StreamForm.tsx | 9 +++- frontend/src/api/streams.ts | 1 + frontend/src/viewer/pages/WatchPage.tsx | 3 ++ 19 files changed, 401 insertions(+), 47 deletions(-) create mode 100644 backend/services/node_store.py create mode 100644 edge-agent/install-autostart.ps1 create mode 100755 edge-agent/install-autostart.sh diff --git a/backend/main.py b/backend/main.py index b19c21f..480162c 100644 --- a/backend/main.py +++ b/backend/main.py @@ -6,17 +6,63 @@ from fastapi.responses import FileResponse from starlette.responses import RedirectResponse from starlette.background import BackgroundTask import httpx +import asyncio +import time +import psutil from database import init_db from services.transcoder import transcode_mgr from routes import streams_router, nodes_router, auth_router from routes.settings_route import router as settings_router from config import settings +from services.load_balancer import lb +from services.node_store import load_nodes + + +async def monitor_origin(): + previous = psutil.net_io_counters() + previous_time = time.monotonic() + while True: + await asyncio.sleep(10) + current = psutil.net_io_counters() + current_time = time.monotonic() + elapsed = current_time - previous_time + network_in = ((current.bytes_recv - previous.bytes_recv) * 8 / elapsed / 1e6) if elapsed > 0 else None + network_out = ((current.bytes_sent - previous.bytes_sent) * 8 / elapsed / 1e6) if elapsed > 0 else None + media_available = False + viewer_count = None + try: + async with httpx.AsyncClient(timeout=3) as client: + response = await client.get(f"http://127.0.0.1:{settings.media_origin_api_port}/v3/paths/list") + response.raise_for_status() + payload = response.json() + viewer_count = sum(item.get("readers", 0) for item in payload.get("items", [])) + media_available = True + except Exception: + pass + lb.report_health("default", { + "cpu_percent": psutil.cpu_percent(), + "memory_percent": psutil.virtual_memory().percent, + "bandwidth_mbps": network_in + network_out if network_in is not None and network_out is not None else None, + "network_in_mbps": network_in, + "network_out_mbps": network_out, + "active_connections": len(psutil.net_connections(kind="tcp")), + "hls_readers": viewer_count, + "viewer_count": viewer_count, + "frame_drop_indicators": None, + "packet_loss_indicators": None, + "media_metrics_available": False, + }) + previous = current + previous_time = current_time @asynccontextmanager async def lifespan(app: FastAPI): + load_nodes() await init_db() + origin_monitor = asyncio.create_task(monitor_origin()) yield + origin_monitor.cancel() transcode_mgr.stop_all() diff --git a/backend/requirements.txt b/backend/requirements.txt index 8a18169..75b8188 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -6,3 +6,4 @@ pydantic>=2.7.0 pydantic-settings>=2.2.0 python-multipart>=0.0.9 httpx>=0.27.0 +psutil>=5.9.0 diff --git a/backend/routes/nodes.py b/backend/routes/nodes.py index 7edf5bb..6cbe7b7 100644 --- a/backend/routes/nodes.py +++ b/backend/routes/nodes.py @@ -5,14 +5,16 @@ from config import settings from routes.auth import require_auth from services.load_balancer import lb from services.multicast_detector import multicast_detector +from services.node_store import save_nodes router = APIRouter(prefix="/api/nodes", tags=["nodes"]) class NodeUpdate(BaseModel): - weight: int + weight: int | None = None enabled: bool | None = None subnet: str | None = None + name: str | None = None class StrategyUpdate(BaseModel): @@ -27,6 +29,12 @@ class NodeHeartbeat(BaseModel): bandwidth_mbps: float | None = None active_connections: int | None = None hls_readers: int | None = None + network_in_mbps: float | None = None + network_out_mbps: float | None = None + viewer_count: int | None = None + frame_drop_indicators: int | None = None + packet_loss_indicators: int | None = None + media_metrics_available: bool = False class NodeRegister(BaseModel): @@ -83,15 +91,22 @@ async def update_node(node_id: str, body: NodeUpdate, _: bool = Depends(require_ found = False for n in settings.edge_nodes: if n["id"] == node_id: - n["weight"] = body.weight + if body.weight is not None: + n["weight"] = body.weight if body.enabled is not None: n["enabled"] = body.enabled if body.subnet is not None: n["subnet"] = body.subnet + if body.name is not None: + name = body.name.strip() + if not name: + raise HTTPException(status_code=422, detail="Node name cannot be empty") + n["name"] = name found = True break if not found: raise HTTPException(status_code=404, detail="Node not found") + save_nodes() return {"ok": True} @@ -100,13 +115,13 @@ async def register_node(body: NodeRegister): for n in settings.edge_nodes: if n["id"] == body.node_id: n.update({ - "name": body.name, "host": body.host, "hls_port": body.hls_port, "api_port": body.api_port, "weight": body.weight, "subnet": body.subnet, }) + save_nodes() return {"ok": True, "action": "updated"} settings.edge_nodes.append({ "id": body.node_id, @@ -119,6 +134,7 @@ async def register_node(body: NodeRegister): "subnet": body.subnet, "is_origin": False, }) + save_nodes() return {"ok": True, "action": "registered"} @@ -130,6 +146,12 @@ async def node_heartbeat(body: NodeHeartbeat): "bandwidth_mbps": body.bandwidth_mbps, "active_connections": body.active_connections, "hls_readers": body.hls_readers, + "network_in_mbps": body.network_in_mbps, + "network_out_mbps": body.network_out_mbps, + "viewer_count": body.viewer_count, + "frame_drop_indicators": body.frame_drop_indicators, + "packet_loss_indicators": body.packet_loss_indicators, + "media_metrics_available": body.media_metrics_available, }) return {"ok": True} diff --git a/backend/routes/streams.py b/backend/routes/streams.py index ce761e0..9764e26 100644 --- a/backend/routes/streams.py +++ b/backend/routes/streams.py @@ -62,14 +62,23 @@ async def list_streams(db: AsyncSession = Depends(get_db)): @router.post("") async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db), - _: bool = Depends(require_auth)): + _: bool = Depends(require_auth)): + title = body.title.strip() + if not title: + raise HTTPException(status_code=422, detail="直播标题不能为空") svc = StreamService(db) - stream = await svc.create_stream( - title=body.title, - description=body.description, - cover_url=body.cover_url, - scheduled_at=body.scheduled_at, - ) + try: + stream = await svc.create_stream( + title=title, + description=body.description, + cover_url=body.cover_url, + scheduled_at=body.scheduled_at, + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + except Exception as exc: + await db.rollback() + raise HTTPException(status_code=500, detail="创建直播失败，请检查日期格式和数据库状态") from exc rtmp_url = media_service.get_rtmp_url(stream.stream_key) return { "id": stream.id, @@ -179,6 +188,7 @@ async def get_stream_by_key(stream_key: str, request: Request, db: AsyncSession "title": s.title, "status": s.status.value, "multicast_available": urls["multicast_available"], + "selected_node": urls.get("selected_node"), "lines": urls["lines"], "qualities": urls["qualities"], } diff --git a/backend/services/load_balancer.py b/backend/services/load_balancer.py index c6bfc2f..98fad20 100644 --- a/backend/services/load_balancer.py +++ b/backend/services/load_balancer.py @@ -132,6 +132,12 @@ class LoadBalancer: "bandwidth_mbps": h.get("bandwidth_mbps"), "active_connections": h.get("active_connections"), "hls_readers": h.get("hls_readers"), + "network_in_mbps": h.get("network_in_mbps"), + "network_out_mbps": h.get("network_out_mbps"), + "viewer_count": h.get("viewer_count"), + "frame_drop_indicators": h.get("frame_drop_indicators"), + "packet_loss_indicators": h.get("packet_loss_indicators"), + "media_metrics_available": h.get("media_metrics_available", False), } if h else None, } return result diff --git a/backend/services/node_store.py b/backend/services/node_store.py new file mode 100644 index 0000000..280a811 --- /dev/null +++ b/backend/services/node_store.py @@ -0,0 +1,27 @@ +import json +from pathlib import Path + +from config import settings + + +STORE_PATH = Path(__file__).resolve().parent.parent / "nodes.json" + + +def load_nodes() -> None: + if not STORE_PATH.exists(): + return + try: + nodes = json.loads(STORE_PATH.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return + if isinstance(nodes, list): + settings.edge_nodes = nodes + + +def save_nodes() -> None: + temporary = STORE_PATH.with_suffix(".json.tmp") + temporary.write_text( + json.dumps(settings.edge_nodes, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + temporary.replace(STORE_PATH) diff --git a/backend/services/stream_service.py b/backend/services/stream_service.py index f1607e9..d699034 100644 --- a/backend/services/stream_service.py +++ b/backend/services/stream_service.py @@ -7,6 +7,7 @@ from services.multicast_detector import multicast_detector from services.transcoder import transcode_mgr from config import settings import json +from datetime import datetime class StreamService: @@ -17,13 +18,19 @@ class StreamService: cover_url: str | None = None, scheduled_at: str | None = None) -> Stream: import uuid + scheduled_datetime = None + if scheduled_at: + try: + scheduled_datetime = datetime.fromisoformat(scheduled_at.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("预定开播时间格式无效") from exc stream = Stream( title=title, description=description, cover_url=cover_url, stream_key=uuid.uuid4().hex, status=StreamStatus.pending, - scheduled_at=scheduled_at, + scheduled_at=scheduled_datetime, ) self.db.add(stream) await self.db.commit() @@ -113,4 +120,8 @@ class StreamService: "lines": lines, "qualities": [{"label": q["label"], "width": q["width"], "height": q["height"]} for q in settings.qualities], "multicast_available": multicast_detector.is_available(), + "selected_node": { + "id": best_node["id"], + "name": best_node["name"], + } if best_node else None, } diff --git a/edge-agent/agent/agent.go b/edge-agent/agent/agent.go index cafafe3..ebe0400 100644 --- a/edge-agent/agent/agent.go +++ b/edge-agent/agent/agent.go @@ -5,17 +5,19 @@ import ( "fmt" "log" "os" + "sync" "time" ) type Config struct { - NodeName string - OriginURL string - HlsPort int - ApiPort int - Subnet string - DataDir string - Weight int + NodeName string + OriginURL string + DiscoveryURL string + HlsPort int + ApiPort int + Subnet string + DataDir string + Weight int } type Agent struct { @@ -26,6 +28,8 @@ type Agent struct { monitor *Monitor nodeID string stopCh chan struct{} + connected bool + mu sync.RWMutex } func New(cfg Config, mtxFS embed.FS) (*Agent, error) { @@ -42,6 +46,13 @@ func (a *Agent) Start() error { return fmt.Errorf("create data dir: %w", err) } + a.hb = NewHeartbeat(a.cfg, a.nodeID) + if err := a.hb.Discover(); err != nil { + log.Printf("Warning: initial discovery failed: %v (will retry every 10 seconds)", err) + } else { + a.cfg.OriginURL = a.hb.OriginURL() + } + a.mtx = NewMediaMTXManager(a.cfg, a.mediamtxFS) if err := a.mtx.Extract(); err != nil { return fmt.Errorf("extract mediamtx: %w", err) @@ -56,12 +67,10 @@ func (a *Agent) Start() error { } a.monitor = NewMonitor() - a.hb = NewHeartbeat(a.cfg, a.nodeID) - time.Sleep(2 * time.Second) - if err := a.hb.Register(); err != nil { - log.Printf("Warning: registration failed: %v (will retry)", err) + if err := a.connect(); err != nil { + log.Printf("Warning: connection failed: %v (will retry every 10 seconds)", err) } go a.heartbeatLoop() @@ -87,8 +96,15 @@ func (a *Agent) heartbeatLoop() { case <-a.stopCh: return case <-ticker.C: + if !a.Connected() { + if err := a.connect(); err != nil { + log.Printf("Connection retry failed: %v", err) + continue + } + } metrics := a.monitor.Snapshot() if err := a.hb.Send(metrics); err != nil { + a.setConnected(false) log.Printf("Heartbeat error: %v", err) } } @@ -104,12 +120,37 @@ func (a *Agent) monitorLoop() { return case <-ticker.C: a.monitor.Tick() + if a.mtx != nil { + a.monitor.SetMediaMetrics(a.mtx.Metrics()) + } } } } -func (a *Agent) NodeID() string { return a.nodeID } -func (a *Agent) Config() Config { return a.cfg } +func (a *Agent) NodeID() string { return a.nodeID } +func (a *Agent) Config() Config { return a.cfg } +func (a *Agent) Connected() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.connected +} + +func (a *Agent) setConnected(connected bool) { + a.mu.Lock() + a.connected = connected + a.mu.Unlock() +} + +func (a *Agent) connect() error { + if err := a.hb.Discover(); err != nil { + return err + } + if err := a.hb.Register(); err != nil { + return err + } + a.setConnected(true) + return nil +} func (a *Agent) Metrics() Metrics { if a.monitor != nil { return a.monitor.Snapshot() diff --git a/edge-agent/agent/heartbeat.go b/edge-agent/agent/heartbeat.go index c84c102..7478a8c 100644 --- a/edge-agent/agent/heartbeat.go +++ b/edge-agent/agent/heartbeat.go @@ -4,7 +4,9 @@ import ( "bytes" "encoding/json" "fmt" + "io" "net/http" + "strings" "time" ) @@ -21,6 +23,12 @@ type HeartbeatPayload struct { BandwidthMbps float64 `json:"bandwidth_mbps"` ActiveConnections int `json:"active_connections"` HlsReaders int `json:"hls_readers"` + NetworkInMbps float64 `json:"network_in_mbps"` + NetworkOutMbps float64 `json:"network_out_mbps"` + ViewerCount int `json:"viewer_count"` + FrameDrops int `json:"frame_drop_indicators"` + PacketLoss int `json:"packet_loss_indicators"` + MediaAvailable bool `json:"media_metrics_available"` } type RegisterPayload struct { @@ -54,6 +62,37 @@ func (h *Heartbeat) Register() error { return h.post("/api/nodes/register", payload) } +func (h *Heartbeat) Discover() error { + if h.cfg.DiscoveryURL == "" { + return nil + } + resp, err := h.client.Get(h.cfg.DiscoveryURL) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return fmt.Errorf("discovery status %d", resp.StatusCode) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, 4096)) + if err != nil { + return err + } + host := strings.TrimSpace(string(body)) + if host == "" { + return fmt.Errorf("discovery returned an empty address") + } + if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") { + host = "http://" + host + } + h.cfg.OriginURL = strings.TrimRight(host, "/") + return nil +} + +func (h *Heartbeat) OriginURL() string { + return h.cfg.OriginURL +} + func (h *Heartbeat) Send(m Metrics) error { payload := HeartbeatPayload{ NodeID: h.nodeID, @@ -62,6 +101,12 @@ func (h *Heartbeat) Send(m Metrics) error { BandwidthMbps: m.Bandwidth, ActiveConnections: m.ActiveConns, HlsReaders: m.HlsReaders, + NetworkInMbps: m.NetworkIn, + NetworkOutMbps: m.NetworkOut, + ViewerCount: m.ViewerCount, + FrameDrops: m.FrameDrops, + PacketLoss: m.PacketLoss, + MediaAvailable: m.MediaAvailable, } return h.post("/api/nodes/heartbeat", payload) } diff --git a/edge-agent/agent/mediamtx.go b/edge-agent/agent/mediamtx.go index 3b750f5..2db34a8 100644 --- a/edge-agent/agent/mediamtx.go +++ b/edge-agent/agent/mediamtx.go @@ -1,6 +1,7 @@ package agent import ( + "bufio" "embed" "fmt" "io" @@ -9,6 +10,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" ) type MediaMTXManager struct { @@ -17,6 +19,15 @@ type MediaMTXManager struct { binPath string configPath string cmd *exec.Cmd + mu sync.RWMutex + metrics MediaMetrics +} + +type MediaMetrics struct { + ViewerCount int + FrameDrops int + PacketLoss int + Available bool } func NewMediaMTXManager(cfg Config, fs embed.FS) *MediaMTXManager { @@ -116,14 +127,57 @@ paths: func (m *MediaMTXManager) Start() error { m.cmd = exec.Command(m.binPath, m.configPath) - m.cmd.Stdout = os.Stdout - m.cmd.Stderr = os.Stderr + stdout, err := m.cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("capture mediamtx stdout: %w", err) + } + stderr, err := m.cmd.StderrPipe() + if err != nil { + return fmt.Errorf("capture mediamtx stderr: %w", err) + } if err := m.cmd.Start(); err != nil { return fmt.Errorf("start mediamtx: %w", err) } + go m.parseLogs(stdout) + go m.parseLogs(stderr) return nil } +func (m *MediaMTXManager) parseLogs(reader io.Reader) { + scanner := bufio.NewScanner(reader) + for scanner.Scan() { + line := scanner.Text() + fmt.Fprintln(os.Stdout, line) + lower := strings.ToLower(line) + m.mu.Lock() + if strings.Contains(lower, "reader opened") || strings.Contains(lower, "hls client connected") { + m.metrics.Available = true + m.metrics.ViewerCount++ + } + if strings.Contains(lower, "reader closed") || strings.Contains(lower, "hls client disconnected") { + m.metrics.Available = true + if m.metrics.ViewerCount > 0 { + m.metrics.ViewerCount-- + } + } + if strings.Contains(lower, "frame drop") || strings.Contains(lower, "dropped frame") { + m.metrics.Available = true + m.metrics.FrameDrops++ + } + if strings.Contains(lower, "packet loss") || strings.Contains(lower, "lost packet") { + m.metrics.Available = true + m.metrics.PacketLoss++ + } + m.mu.Unlock() + } +} + +func (m *MediaMTXManager) Metrics() MediaMetrics { + m.mu.RLock() + defer m.mu.RUnlock() + return m.metrics +} + func (m *MediaMTXManager) Stop() { if m.cmd != nil && m.cmd.Process != nil { m.cmd.Process.Kill() diff --git a/edge-agent/agent/monitor.go b/edge-agent/agent/monitor.go index a4b0000..595ebdb 100644 --- a/edge-agent/agent/monitor.go +++ b/edge-agent/agent/monitor.go @@ -11,11 +11,17 @@ import ( ) type Metrics struct { - CPU float64 - Memory float64 - Bandwidth float64 - ActiveConns int - HlsReaders int + CPU float64 + Memory float64 + Bandwidth float64 + NetworkIn float64 + NetworkOut float64 + ActiveConns int + HlsReaders int + ViewerCount int + FrameDrops int + PacketLoss int + MediaAvailable bool } type Monitor struct { @@ -54,8 +60,9 @@ func (m *Monitor) Tick() { } else { elapsed := now.Sub(m.prevNetTime).Seconds() if elapsed > 0 { - bytesTotal := (curIn - m.prevNetIn) + (curOut - m.prevNetOut) - metrics.Bandwidth = float64(bytesTotal) * 8 / elapsed / 1e6 + metrics.NetworkIn = float64(curIn-m.prevNetIn) * 8 / elapsed / 1e6 + metrics.NetworkOut = float64(curOut-m.prevNetOut) * 8 / elapsed / 1e6 + metrics.Bandwidth = metrics.NetworkIn + metrics.NetworkOut } m.prevNetIn = curIn m.prevNetOut = curOut @@ -81,3 +88,13 @@ func (m *Monitor) Snapshot() Metrics { defer m.mu.RUnlock() return m.current } + +func (m *Monitor) SetMediaMetrics(media MediaMetrics) { + m.mu.Lock() + m.current.ViewerCount = media.ViewerCount + m.current.HlsReaders = media.ViewerCount + m.current.FrameDrops = media.FrameDrops + m.current.PacketLoss = media.PacketLoss + m.current.MediaAvailable = media.Available + m.mu.Unlock() +} diff --git a/edge-agent/install-autostart.ps1 b/edge-agent/install-autostart.ps1 new file mode 100644 index 0000000..cc599b3 --- /dev/null +++ b/edge-agent/install-autostart.ps1 @@ -0,0 +1,14 @@ +param( + [string]$Executable = "$PSScriptRoot\dist\streamcore-edge-windows-amd64.exe" +) + +$resolved = (Resolve-Path $Executable).Path +$taskName = "StreamCore Edge Agent" +$action = New-ScheduledTaskAction -Execute $resolved +$trigger = New-ScheduledTaskTrigger -AtStartup +$settings = New-ScheduledTaskSettingsSet -RestartCount 999 -RestartInterval (New-TimeSpan -Seconds 10) +$principal = New-ScheduledTaskPrincipal -UserId "SYSTEM" -LogonType ServiceAccount -RunLevel Highest + +Register-ScheduledTask -TaskName $taskName -Action $action -Trigger $trigger -Settings $settings -Principal $principal -Force | Out-Null +Start-ScheduledTask -TaskName $taskName +Write-Host "Installed and started scheduled task: $taskName" diff --git a/edge-agent/install-autostart.sh b/edge-agent/install-autostart.sh new file mode 100755 index 0000000..01fc714 --- /dev/null +++ b/edge-agent/install-autostart.sh @@ -0,0 +1,31 @@ +#!/bin/sh +set -eu + +BIN="$(readlink -f "${1:-./dist/streamcore-edge-linux-amd64}")" +SERVICE="/etc/systemd/system/streamcore-edge.service" + +if [ ! -x "$BIN" ]; then + echo "Edge agent executable not found: $BIN" >&2 + exit 1 +fi + +cat > "$SERVICE" < n.id === nid); if (!node) return try { await api.put(`/nodes/${nid}`, { weight: node.weight, enabled: en }); setNodes(p => p.map(n => n.id === nid ? { ...n, enabled: en } : n)) } catch { message.error('更新失败') } } + async function updateName(nid: string, name: string) { + const trimmed = name.trim() + if (!trimmed) { message.error('节点名称不能为空'); loadData(); return } + try { await api.put(`/nodes/${nid}`, { name: trimmed }); setNodes(p => p.map(n => n.id === nid ? { ...n, name: trimmed } : n)); message.success('名称已更新') } + catch { message.error('名称更新失败'); loadData() } + } const onlineCount = nodes.filter(n => n.online).length @@ -42,8 +48,8 @@ export default function NodeManager() { boxShadow: `0 0 6px ${v ? 'rgba(91,140,90,0.3)' : 'rgba(181,74,74,0.3)'}` }} /> ), }, - { title: '名称', dataIndex: 'name', key: 'name', - render: (v: string, r: NodeInfo) => {v}{r.is_origin && Origin} }, + { title: '名称', dataIndex: 'name', key: 'name', width: 190, + render: (v: string, r: NodeInfo) => updateName(r.id, e.currentTarget.value)} onBlur={e => { if (e.currentTarget.value !== v) updateName(r.id, e.currentTarget.value) }} />{r.is_origin && 主节点} }, { title: '地址', dataIndex: 'host', key: 'host', width: 130 }, { title: '子网', dataIndex: 'subnet', key: 'subnet', width: 130, render: (v: string) => {v} }, { title: 'CPU', key: 'cpu', width: 80, @@ -58,6 +64,12 @@ export default function NodeManager() { : -- }, { title: '带宽', key: 'bw', width: 85, render: (_: any, r: NodeInfo) => r.metrics?.bandwidth_mbps != null ? {r.metrics.bandwidth_mbps.toFixed(1)} Mbps : -- }, + { title: '网络入/出', key: 'network', width: 125, + render: (_: any, r: NodeInfo) => r.metrics?.network_in_mbps != null && r.metrics?.network_out_mbps != null ? {r.metrics.network_in_mbps.toFixed(1)} / {r.metrics.network_out_mbps.toFixed(1)} Mbps : 不可用 }, + { title: '观众', key: 'viewers', width: 65, + render: (_: any, r: NodeInfo) => r.metrics?.viewer_count != null ? r.metrics.viewer_count : 不可用 }, + { title: '媒体异常', key: 'media', width: 120, + render: (_: any, r: NodeInfo) => r.metrics?.media_metrics_available ? 丢帧 {r.metrics.frame_drop_indicators ?? 0} / 丢包 {r.metrics.packet_loss_indicators ?? 0} : 不可用 }, { title: '连接', key: 'conn', width: 60, render: (_: any, r: NodeInfo) => r.metrics?.active_connections != null ? r.metrics.active_connections : -- }, { title: '权重', dataIndex: 'weight', key: 'weight', width: 120, diff --git a/frontend/src/admin/pages/StreamForm.tsx b/frontend/src/admin/pages/StreamForm.tsx index 51541d5..9af38ae 100644 --- a/frontend/src/admin/pages/StreamForm.tsx +++ b/frontend/src/admin/pages/StreamForm.tsx @@ -3,6 +3,7 @@ import { useNavigate } from 'react-router-dom' import { Form, Input, Button, DatePicker, Typography, message, Space, Alert } from 'antd' import { CopyOutlined, ArrowLeftOutlined, CheckCircleOutlined } from '@ant-design/icons' import { createStream } from '../../api/streams' +import axios from 'axios' const { Title, Text } = Typography const { TextArea } = Input @@ -18,8 +19,12 @@ export default function StreamForm() { const res = await createStream({ title: values.title, description: values.description, cover_url: values.cover_url, scheduled_at: values.scheduled_at?.toISOString() }) setResult({ rtmp_url: res.rtmp_url, stream_key: res.stream_key }) message.success('直播已创建') - } catch { message.error('创建失败') } - setLoading(false) + } catch (error) { + const detail = axios.isAxiosError(error) ? error.response?.data?.detail : undefined + message.error(typeof detail === 'string' ? detail : '创建失败，请稍后重试') + } finally { + setLoading(false) + } } if (result) { diff --git a/frontend/src/api/streams.ts b/frontend/src/api/streams.ts index 11a1c4e..ae95e69 100644 --- a/frontend/src/api/streams.ts +++ b/frontend/src/api/streams.ts @@ -26,6 +26,7 @@ export interface WatchInfo { title: string status: string multicast_available: boolean + selected_node?: { id: string; name: string } | null lines: LineInfo[] qualities: QualityInfo[] } diff --git a/frontend/src/viewer/pages/WatchPage.tsx b/frontend/src/viewer/pages/WatchPage.tsx index 5096741..22f7cc9 100644 --- a/frontend/src/viewer/pages/WatchPage.tsx +++ b/frontend/src/viewer/pages/WatchPage.tsx @@ -19,6 +19,7 @@ export default function WatchPage() { const [activeLine, setActiveLine] = useState('') const [activeQuality, setActiveQuality] = useState('') const hlsRef = useRef(null) + const currentLine = info?.lines.find(line => line.node_id === activeLine) useEffect(() => { if (!id) return @@ -102,6 +103,8 @@ export default function WatchPage() { + 节点：{currentLine?.node_name || info.selected_node?.name || '不可用'} + 清晰度：{activeQuality || '不可用'} {info.multicast_available ? '组播' : '单播'}