diff --git a/backend/models/__init__.py b/backend/models/__init__.py index 1201730..757fcaa 100644 --- a/backend/models/__init__.py +++ b/backend/models/__init__.py @@ -1,3 +1,4 @@ from .stream import Stream +from .audit import ControlCommand, ControlSession, ViewerSession -__all__ = ["Stream"] +__all__ = ["Stream", "ViewerSession", "ControlSession", "ControlCommand"] diff --git a/backend/models/audit.py b/backend/models/audit.py new file mode 100644 index 0000000..485f9b3 --- /dev/null +++ b/backend/models/audit.py @@ -0,0 +1,50 @@ +import uuid +from datetime import datetime + +from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from database import Base + + +class ViewerSession(Base): + __tablename__ = "viewer_sessions" + + session_id: Mapped[str] = mapped_column(String(64), primary_key=True) + stream_id: Mapped[int] = mapped_column(ForeignKey("streams.id"), index=True) + node_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + node_name: Mapped[str | None] = mapped_column(String(200), nullable=True) + quality: Mapped[str | None] = mapped_column(String(64), nullable=True) + playback_state: Mapped[str] = mapped_column(String(32), default="idle") + buffered_seconds: Mapped[float | None] = mapped_column(Float, nullable=True) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + remote_adjusted: Mapped[bool] = mapped_column(Boolean, default=False) + last_heartbeat: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + + +class ControlSession(Base): + __tablename__ = "control_sessions" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(128), unique=True, default=lambda: uuid.uuid4().hex) + stream_id: Mapped[int] = mapped_column(ForeignKey("streams.id"), index=True) + operator: Mapped[str] = mapped_column(String(128), default="administrator") + expires_at: Mapped[datetime] = mapped_column(DateTime, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + + +class ControlCommand(Base): + __tablename__ = "control_commands" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + control_session_id: Mapped[int] = mapped_column(ForeignKey("control_sessions.id"), index=True) + stream_id: Mapped[int] = mapped_column(ForeignKey("streams.id"), index=True) + target_session_id: Mapped[str] = mapped_column(ForeignKey("viewer_sessions.session_id"), index=True) + operator: Mapped[str] = mapped_column(String(128), default="administrator") + action: Mapped[str] = mapped_column(String(32)) + value: Mapped[str | None] = mapped_column(String(256), nullable=True) + status: Mapped[str] = mapped_column(String(32), default="pending") + result: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + claimed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) diff --git a/backend/routes/streams.py b/backend/routes/streams.py index 16562c9..ab40775 100644 --- a/backend/routes/streams.py +++ b/backend/routes/streams.py @@ -3,10 +3,12 @@ import secrets import asyncio import time import ntpath +from datetime import datetime, timedelta from pathlib import Path from fastapi import APIRouter, HTTPException, Depends, Request from fastapi.responses import FileResponse, StreamingResponse +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from pydantic import BaseModel, Field from typing import Optional @@ -16,6 +18,7 @@ from services.stream_service import StreamService from services.media_service import media_service from services.load_balancer import lb from config import settings +from models.audit import ControlCommand, ControlSession, ViewerSession router = APIRouter(prefix="/api/streams", tags=["streams"]) @@ -46,16 +49,31 @@ class StreamUpdate(BaseModel): class RemoteSessionCreate(BaseModel): expires_in: int = 300 + operator: str = "administrator" class RemoteCommand(BaseModel): token: str action: str - value: float | None = None + value: str | None = None + targets: list[str] = Field(default_factory=list) -_remote_sessions: dict[str, dict] = {} -_remote_commands: dict[int, list[dict]] = {} +class ViewerHeartbeat(BaseModel): + session_id: str + node_id: str | None = None + node_name: str | None = None + quality: str | None = None + playback_state: str = "idle" + buffered_seconds: float | None = None + error: str | None = None + + +class CommandResult(BaseModel): + result: str = "applied" + + +VIEWER_ONLINE_SECONDS = 45 def _media_path(filename: str | None) -> Path | None: @@ -136,6 +154,107 @@ async def list_streams(admin: bool = False, db: AsyncSession = Depends(get_db)): return [_serialize(s, include_private=admin) for s in streams] +@router.get("/catalog/events") +async def catalog_events(request: Request, db: AsyncSession = Depends(get_db)): + async def events(): + previous = None + while not await request.is_disconnected(): + streams = await StreamService(db).get_public_streams() + payload = json.dumps( + [_serialize(stream, include_private=False) for stream in streams], + ensure_ascii=False, + ) + if payload != previous: + yield f"event: catalog\ndata: {payload}\n\n" + previous = payload + else: + yield ": keep-alive\n\n" + await asyncio.sleep(2) + + return StreamingResponse(events(), media_type="text/event-stream", headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }) + + +@router.post("/{stream_id}/viewer-heartbeat") +async def viewer_heartbeat(stream_id: int, body: ViewerHeartbeat, + db: AsyncSession = Depends(get_db)): + stream = await StreamService(db).get_stream(stream_id) + if not stream: + raise HTTPException(status_code=404, detail="节目不存在") + session_id = body.session_id.strip() + if not session_id or len(session_id) > 64: + raise HTTPException(status_code=422, detail="观众会话 ID 无效") + viewer = await db.get(ViewerSession, session_id) + if viewer and viewer.stream_id != stream_id: + raise HTTPException(status_code=409, detail="观众会话已用于其他节目") + if not viewer: + viewer = ViewerSession(session_id=session_id, stream_id=stream_id) + db.add(viewer) + viewer.node_id = body.node_id + viewer.node_name = body.node_name + viewer.quality = body.quality + viewer.playback_state = body.playback_state[:32] + viewer.buffered_seconds = body.buffered_seconds + viewer.error = body.error[:1000] if body.error else None + viewer.last_heartbeat = datetime.utcnow() + await db.commit() + return {"ok": True, "remote_adjusted": viewer.remote_adjusted} + + +@router.get("/{stream_id}/viewer-sessions") +async def list_viewer_sessions(stream_id: int, db: AsyncSession = Depends(get_db), + _: bool = Depends(require_auth)): + cutoff = datetime.utcnow() - timedelta(seconds=VIEWER_ONLINE_SECONDS) + viewers = (await db.execute(select(ViewerSession).where( + ViewerSession.stream_id == stream_id, + ViewerSession.last_heartbeat >= cutoff, + ).order_by(ViewerSession.last_heartbeat.desc()))).scalars().all() + return [{ + "session_id": viewer.session_id, + "node_id": viewer.node_id, + "node_name": viewer.node_name, + "quality": viewer.quality, + "playback_state": viewer.playback_state, + "buffered_seconds": viewer.buffered_seconds, + "error": viewer.error, + "remote_adjusted": viewer.remote_adjusted, + "last_heartbeat": viewer.last_heartbeat.isoformat(), + } for viewer in viewers] + + +@router.get("/{stream_id}/control-audit") +async def control_audit(stream_id: int, db: AsyncSession = Depends(get_db), + _: bool = Depends(require_auth)): + commands = (await db.execute(select(ControlCommand).where( + ControlCommand.stream_id == stream_id, + ).order_by(ControlCommand.created_at.desc()).limit(100))).scalars().all() + return [{ + "id": command.id, + "operator": command.operator, + "target_session_id": command.target_session_id, + "action": command.action, + "value": command.value, + "status": command.status, + "result": command.result, + "created_at": command.created_at.isoformat(), + "claimed_at": command.claimed_at.isoformat() if command.claimed_at else None, + } for command in commands] + + +@router.post("/{stream_id}/remote-command/{command_id}/result") +async def report_command_result(stream_id: int, command_id: int, body: CommandResult, + session_id: str, db: AsyncSession = Depends(get_db)): + command = await db.get(ControlCommand, command_id) + if not command or command.stream_id != stream_id or command.target_session_id != session_id: + raise HTTPException(status_code=404, detail="控制命令不存在") + command.status = "completed" + command.result = body.result[:1000] + await db.commit() + return {"ok": True} + + @router.get("/media-files") async def list_media_files(_: bool = Depends(require_auth)): root = Path(settings.replay_media_dir) @@ -294,34 +413,77 @@ async def create_remote_session(stream_id: int, body: RemoteSessionCreate, stream = await StreamService(db).get_stream(stream_id) if not stream or not stream.remote_control_enabled: raise HTTPException(status_code=409, detail="节目未启用远程控制") - token = secrets.token_urlsafe(24) - _remote_sessions[token] = { - "stream_id": stream_id, - "expires_at": time.time() + min(max(body.expires_in, 30), 3600), - } - return {"token": token, "expires_in": min(max(body.expires_in, 30), 3600)} + expires_in = min(max(body.expires_in, 30), 3600) + session = ControlSession( + token=secrets.token_urlsafe(24), + stream_id=stream_id, + operator=body.operator.strip() or "administrator", + expires_at=datetime.utcnow() + timedelta(seconds=expires_in), + ) + db.add(session) + await db.commit() + return {"token": session.token, "expires_in": expires_in} @router.post("/{stream_id}/remote-command") async def send_remote_command(stream_id: int, body: RemoteCommand, + db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)): - session = _remote_sessions.get(body.token) - if not session or session["stream_id"] != stream_id or session["expires_at"] < time.time(): + session = (await db.execute(select(ControlSession).where( + ControlSession.token == body.token, + ControlSession.stream_id == stream_id, + ))).scalar_one_or_none() + if not session or session.expires_at < datetime.utcnow(): raise HTTPException(status_code=403, detail="远程控制会话无效或已过期") - if body.action not in {"play", "pause", "seek"}: + stream = await StreamService(db).get_stream(stream_id) + if not stream or not stream.remote_control_enabled: + raise HTTPException(status_code=409, detail="节目未启用远程控制") + if body.action not in {"switch_node", "switch_quality", "restore_auto"}: raise HTTPException(status_code=422, detail="不支持的远程命令") - command = {"action": body.action, "value": body.value, "created_at": time.time()} - _remote_commands.setdefault(stream_id, []).append(command) - return {"ok": True, "command": command} + if not body.targets: + raise HTTPException(status_code=422, detail="至少选择一个观众会话") + viewers = (await db.execute(select(ViewerSession).where( + ViewerSession.stream_id == stream_id, + ViewerSession.session_id.in_(body.targets), + ))).scalars().all() + found = {viewer.session_id for viewer in viewers} + if found != set(body.targets): + raise HTTPException(status_code=404, detail="部分观众会话不存在") + commands = [] + for target in body.targets: + command = ControlCommand( + control_session_id=session.id, + stream_id=stream_id, + target_session_id=target, + operator=session.operator, + action=body.action, + value=body.value, + ) + db.add(command) + commands.append(command) + await db.commit() + return {"ok": True, "count": len(commands), "command_ids": [command.id for command in commands]} @router.get("/{stream_id}/remote-command") -async def poll_remote_command(stream_id: int, token: str): - session = _remote_sessions.get(token) - if not session or session["stream_id"] != stream_id or session["expires_at"] < time.time(): - raise HTTPException(status_code=403, detail="远程控制会话无效或已过期") - commands = _remote_commands.get(stream_id, []) - return commands[-1] if commands else None +async def poll_remote_command(stream_id: int, session_id: str, + db: AsyncSession = Depends(get_db)): + viewer = await db.get(ViewerSession, session_id) + if not viewer or viewer.stream_id != stream_id: + raise HTTPException(status_code=404, detail="观众会话不存在") + command = (await db.execute(select(ControlCommand).where( + ControlCommand.stream_id == stream_id, + ControlCommand.target_session_id == session_id, + ControlCommand.status == "pending", + ).order_by(ControlCommand.created_at).limit(1))).scalar_one_or_none() + if not command: + return None + command.status = "claimed" + command.claimed_at = datetime.utcnow() + viewer.remote_adjusted = command.action != "restore_auto" + await db.commit() + return {"id": command.id, "action": command.action, "value": command.value, + "created_at": command.created_at.isoformat()} @router.post("/{stream_id}/go-live") diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 7a30e8d..40279b4 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -12,6 +12,7 @@ const WatchPage = lazy(() => import('./viewer/pages/WatchPage')) const AdminLayout = lazy(() => import('./admin/AdminLayout')) const LoginPage = lazy(() => import('./admin/LoginPage')) const Dashboard = lazy(() => import('./admin/pages/Dashboard')) +const LiveBoard = lazy(() => import('./admin/pages/LiveBoard')) const StreamList = lazy(() => import('./admin/pages/StreamList')) const StreamForm = lazy(() => import('./admin/pages/StreamForm')) const ReplayEditor = lazy(() => import('./admin/pages/ReplayEditor')) @@ -74,6 +75,7 @@ function AppRoutes() { }> } /> + } /> } /> } /> } /> diff --git a/frontend/src/admin/AdminLayout.tsx b/frontend/src/admin/AdminLayout.tsx index 4a64d2d..ed0db86 100644 --- a/frontend/src/admin/AdminLayout.tsx +++ b/frontend/src/admin/AdminLayout.tsx @@ -2,7 +2,7 @@ import { useState } from 'react' import { Outlet, useNavigate, useLocation } from 'react-router-dom' import { Layout, Menu, Button, Typography } from 'antd' import { - DashboardOutlined, VideoCameraOutlined, PlusOutlined, + DashboardOutlined, VideoCameraOutlined, PlusOutlined, MonitorOutlined, CloudServerOutlined, SettingOutlined, LogoutOutlined, } from '@ant-design/icons' import { clearToken } from '../api/client' @@ -13,6 +13,7 @@ const { Text } = Typography const menuItems = [ { key: '/admin', icon: , label: '概览' }, + { key: '/admin/board', icon: , label: '直播看板' }, { key: '/admin/streams', icon: , label: '节目列表' }, { key: '/admin/streams/new', icon: , label: '创建节目' }, { key: '/admin/nodes', icon: , label: '边缘节点' }, diff --git a/frontend/src/admin/pages/LiveBoard.tsx b/frontend/src/admin/pages/LiveBoard.tsx new file mode 100644 index 0000000..c1b0da8 --- /dev/null +++ b/frontend/src/admin/pages/LiveBoard.tsx @@ -0,0 +1,127 @@ +import { useEffect, useState } from 'react' +import { Button, Card, Col, Empty, Row, Select, Space, Statistic, Table, Tag, Typography, message } from 'antd' +import { ReloadOutlined, SignalFilled, TeamOutlined, WarningOutlined } from '@ant-design/icons' +import { + createRemoteSession, + fetchStreams, + fetchViewerSessions, + sendRemoteCommand, + type StreamInfo, + type ViewerSessionInfo, +} from '../../api/streams' + +const { Title, Text } = Typography + +export default function LiveBoard() { + const [streams, setStreams] = useState([]) + const [streamId, setStreamId] = useState() + const [sessions, setSessions] = useState([]) + const [selected, setSelected] = useState([]) + const [loading, setLoading] = useState(true) + const [quality, setQuality] = useState('auto') + + async function load() { + try { + const programs = await fetchStreams(true) + setStreams(programs) + const nextId = streamId || programs.find(program => program.status === 'live')?.id + if (nextId) { + setStreamId(nextId) + setSessions(await fetchViewerSessions(nextId)) + } else { + setSessions([]) + } + } catch { + message.error('看板数据加载失败') + } finally { + setLoading(false) + } + } + + useEffect(() => { + void load() + const timer = window.setInterval(load, 5000) + return () => window.clearInterval(timer) + }, [streamId]) + + async function controlViewers() { + if (!streamId || selected.length === 0) return + try { + const session = await createRemoteSession(streamId) + const action = quality === 'auto' ? 'restore_auto' : 'switch_quality' + await sendRemoteCommand(streamId, session.token, action, selected.map(String), quality === 'auto' ? undefined : quality) + message.success(`已向 ${selected.length} 个观众下发播放策略`) + } catch { + message.error('控制命令下发失败,请确认节目已启用审计式播放控制') + } + } + + const activeSessions = sessions.filter(item => Date.now() - new Date(item.last_heartbeat).getTime() < 30000) + const buffering = activeSessions.filter(item => item.playback_state === 'buffering' || Number(item.buffered_seconds || 0) < 1) + + return ( +
+ +
+
+ 直播看板 + 实时检查节目、观众会话、线路与播放健康度 +
+ + + + + } + > + value || '未知' }, + { title: '清晰度', dataIndex: 'quality', render: value => value || '自动' }, + { title: '状态', dataIndex: 'playback_state', render: value => {value} }, + { title: '缓冲', dataIndex: 'buffered_seconds', render: value => `${Number(value || 0).toFixed(1)}s` }, + { title: '远程调整', dataIndex: 'remote_adjusted', render: value => value ? 已调整 : '否' }, + { title: '最后心跳', dataIndex: 'last_heartbeat', render: value => new Date(value).toLocaleTimeString() }, + { title: '错误', dataIndex: 'error', ellipsis: true, render: value => value || '-' }, + ]} + /> + + )} + + + ) +} diff --git a/frontend/src/admin/pages/StreamForm.tsx b/frontend/src/admin/pages/StreamForm.tsx index d91a795..5f46b98 100644 --- a/frontend/src/admin/pages/StreamForm.tsx +++ b/frontend/src/admin/pages/StreamForm.tsx @@ -78,15 +78,15 @@ export default function StreamForm() { style={{ marginBottom: 20, color: '#5E5A52' }}>返回列表
- 创建直播 + 创建节目 - 填写直播信息,系统将生成推流码与 RTMP 地址 + 填写节目信息并预约播出,系统将生成推流码与 RTMP 地址
- - + +