完善节目平台与直播监控体验

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