adminandClaude Opus 4.8
cdf591f933
Replay
- Hand the replay file to nginx with X-Accel-Redirect instead of streaming it
through the backend. A replay is tens of gigabytes, so Python was copying the
whole file and holding a worker for the entire transfer: measured 8 MB in
8.46s (~1 MB/s). nginx serves it from disk with sendfile at ~400 MB/s (1 GB in
2.68s) and the backend process does not appear in CPU usage during the transfer.
Authentication and the path-traversal check still run in Python; only the bytes
moved. The internal location is what keeps the path from being reachable
directly and bypassing that check.
- nginx reloads and the media directory is aliased internally.
Settings
- GET /settings/transcode-profiles returned the bare mapping while the settings
page reads `data.profiles`, so it always saw undefined, fell back to an empty
object, and every save posted `{"profiles": {}}` — which the PUT rejects as
"at least one profile". Saving settings could never succeed. The response now
matches the shape PUT accepts, verified by round-tripping it.
Auth
- Derive the admin token instead of storing it. The token is
sha256(password + ":token"), but it was kept in a module-level set, which only
works with one process: with uvicorn --workers N, logging in on one worker left
the others rejecting the token and the admin appeared logged out at random.
Viewer
- Show the real session id in the stats panel; the previous code assigned it
from a local variable that was never populated, so it always read "不可用".
- Send the first heartbeat before the first command poll. Both fired together on
mount, so the poll reached the backend before the session existed and 404'd on
every page load.
Housekeeping
- Ignore backend/nodes.json; the admin node manager writes it at runtime.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
685 lines
29 KiB
Python
685 lines
29 KiB
Python
import json
|
|
import secrets
|
|
import asyncio
|
|
import mimetypes
|
|
import time
|
|
import ntpath
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Request
|
|
from fastapi.responses import Response, StreamingResponse
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from pydantic import BaseModel, Field
|
|
from typing import Optional
|
|
from database import get_db, session_factory
|
|
from routes.auth import require_auth
|
|
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"])
|
|
|
|
|
|
class RenditionInput(BaseModel):
|
|
label: str
|
|
enabled: bool = True
|
|
width: int
|
|
height: int
|
|
fps: int = 30
|
|
codec: str = "libx264"
|
|
bitrate: str = "2500k"
|
|
crf: int = 23
|
|
preset: str = "veryfast"
|
|
audio_bitrate: str = "128k"
|
|
keyframe_seconds: int = 2
|
|
|
|
|
|
class StreamCreate(BaseModel):
|
|
title: str
|
|
description: Optional[str] = None
|
|
cover_url: Optional[str] = None
|
|
scheduled_at: Optional[str] = None
|
|
replay_file: Optional[str] = None
|
|
replay_url: Optional[str] = None
|
|
chapters: list[dict] = Field(default_factory=list)
|
|
transcode_profile: str = "standard"
|
|
renditions: Optional[list[RenditionInput]] = None
|
|
homepage_visible: bool = True
|
|
remote_control_enabled: bool = False
|
|
|
|
|
|
class StreamUpdate(BaseModel):
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
cover_url: Optional[str] = None
|
|
scheduled_at: Optional[str] = None
|
|
replay_file: Optional[str] = None
|
|
replay_url: Optional[str] = None
|
|
chapters: Optional[list[dict]] = None
|
|
transcode_profile: Optional[str] = None
|
|
renditions: Optional[list[RenditionInput]] = None
|
|
homepage_visible: Optional[bool] = None
|
|
remote_control_enabled: Optional[bool] = None
|
|
|
|
|
|
class RemoteSessionCreate(BaseModel):
|
|
expires_in: int = 300
|
|
operator: str = "administrator"
|
|
|
|
|
|
class RemoteCommand(BaseModel):
|
|
token: str
|
|
action: str
|
|
value: str | None = None
|
|
targets: list[str] = Field(default_factory=list)
|
|
|
|
|
|
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:
|
|
if not filename:
|
|
return None
|
|
root = Path(settings.replay_media_dir).resolve()
|
|
supplied = Path(filename)
|
|
windows_drive, _ = ntpath.splitdrive(filename)
|
|
if supplied.is_absolute() or supplied.drive or windows_drive or ".." in supplied.parts:
|
|
raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
|
|
candidate = (root / supplied).resolve()
|
|
if candidate != root and root not in candidate.parents:
|
|
raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
|
|
if not candidate.is_file():
|
|
raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
|
|
return candidate
|
|
|
|
|
|
def _valid_replay_url(url: str | None) -> str | None:
|
|
if not url:
|
|
return None
|
|
url = url.strip()
|
|
if not (url.startswith("http://") or url.startswith("https://")):
|
|
raise HTTPException(status_code=422, detail="回放 URL 必须是 http(s) 地址")
|
|
if " " in url or "\n" in url or "\r" in url:
|
|
raise HTTPException(status_code=422, detail="回放 URL 格式无效")
|
|
return url
|
|
|
|
|
|
def _utc_iso(moment: datetime) -> str:
|
|
"""Render a stored timestamp as an explicit UTC instant for JSON.
|
|
|
|
Rows are written with datetime.utcnow(), so the value is UTC but carries no
|
|
tzinfo. `isoformat()` therefore emits "2026-09-22T07:47:11" with no offset,
|
|
and JavaScript's Date() reads that as *local* time — in UTC+8 every heartbeat
|
|
looked eight hours old and the live board's "active viewer" filter discarded
|
|
every session, so the count was permanently zero. The trailing Z is what makes
|
|
the client parse it as UTC.
|
|
"""
|
|
if moment is None:
|
|
return None
|
|
return moment.replace(microsecond=0).isoformat() + "Z"
|
|
|
|
|
|
def _status_payload(stream) -> dict:
|
|
return {
|
|
"stream_id": stream.id,
|
|
"status": stream.status.value,
|
|
"replay_local_url": f"/api/streams/{stream.id}/replay" if stream.replay_file else None,
|
|
"replay_url": stream.replay_url,
|
|
"chapters": json.loads(stream.chapters_json or "[]"),
|
|
}
|
|
|
|
|
|
def _normalize_chapters(chapters: list[dict]) -> list:
|
|
normalized = []
|
|
for chapter in chapters:
|
|
title = str(chapter.get("title", "")).strip()
|
|
try:
|
|
chapter_time = float(chapter.get("time", 0))
|
|
except (TypeError, ValueError) as exc:
|
|
raise HTTPException(status_code=422, detail="章节时间必须是非负数字") from exc
|
|
if not title or chapter_time < 0:
|
|
raise HTTPException(status_code=422, detail="章节标题不能为空，时间必须是非负数字")
|
|
normalized.append({"title": title, "time": chapter_time})
|
|
return sorted(normalized, key=lambda chapter: chapter["time"])
|
|
|
|
|
|
def _normalize_renditions(renditions: list[dict]) -> list:
|
|
codecs = {"libx264", "libx265"}
|
|
presets = {"ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow"}
|
|
normalized = []
|
|
labels = set()
|
|
for rendition in renditions:
|
|
label = str(rendition.get("label", "")).strip()
|
|
if not label or not label.replace("-", "").replace("_", "").isalnum() or label in labels:
|
|
raise HTTPException(status_code=422, detail="清晰度标签无效或重复")
|
|
labels.add(label)
|
|
if not rendition.get("enabled", True):
|
|
continue
|
|
width, height = int(rendition["width"]), int(rendition["height"])
|
|
fps, crf = int(rendition.get("fps", 30)), int(rendition.get("crf", 23))
|
|
keyframe_seconds = int(rendition.get("keyframe_seconds", 2))
|
|
codec, preset = rendition.get("codec", "libx264"), rendition.get("preset", "veryfast")
|
|
bitrate, audio_bitrate = str(rendition.get("bitrate", "")), str(rendition.get("audio_bitrate", ""))
|
|
if not (160 <= width <= 7680 and 90 <= height <= 4320 and 1 <= fps <= 120 and 0 <= crf <= 51 and 1 <= keyframe_seconds <= 10):
|
|
raise HTTPException(status_code=422, detail="清晰度参数超出允许范围")
|
|
if codec not in codecs or preset not in presets:
|
|
raise HTTPException(status_code=422, detail="编码器或预设不受支持")
|
|
if not bitrate[:-1].isdigit() or bitrate[-1:] not in {"k", "M"} or not audio_bitrate[:-1].isdigit() or audio_bitrate[-1:] not in {"k", "M"}:
|
|
raise HTTPException(status_code=422, detail="码率格式必须类似 2500k 或 5M")
|
|
normalized.append({"label": label, "enabled": True, "width": width, "height": height, "fps": fps, "codec": codec, "bitrate": bitrate, "crf": crf, "preset": preset, "audio_bitrate": audio_bitrate, "keyframe_seconds": keyframe_seconds})
|
|
if not normalized:
|
|
raise HTTPException(status_code=422, detail="至少启用一个清晰度")
|
|
return normalized
|
|
|
|
|
|
def _serialize(s, include_private: bool = True):
|
|
data = {
|
|
"id": s.id, "title": s.title, "description": s.description,
|
|
"cover_url": s.cover_url, "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,
|
|
"replay_file": s.replay_file,
|
|
"replay_url": s.replay_url,
|
|
"chapters": json.loads(s.chapters_json or "[]"),
|
|
"transcode_profile": s.transcode_profile,
|
|
"renditions": json.loads(getattr(s, "renditions_json", None)) if getattr(s, "renditions_json", None) else None,
|
|
"homepage_visible": s.homepage_visible,
|
|
"remote_control_enabled": s.remote_control_enabled,
|
|
"created_at": str(s.created_at),
|
|
}
|
|
if include_private:
|
|
data["stream_key"] = s.stream_key
|
|
return data
|
|
|
|
|
|
def _get_client_ip(request: Request) -> str | None:
|
|
forwarded = request.headers.get("X-Forwarded-For")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
client = request.headers.get("X-Real-IP")
|
|
if client:
|
|
return client
|
|
if request.client:
|
|
return request.client.host
|
|
return None
|
|
|
|
|
|
@router.get("")
|
|
async def list_streams(admin: bool = False, db: AsyncSession = Depends(get_db)):
|
|
svc = StreamService(db)
|
|
streams = await svc.get_all_streams() if admin else await svc.get_public_streams()
|
|
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():
|
|
db.expire_all()
|
|
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:
|
|
# A session id lives in the viewer's browser, not per programme, so anyone
|
|
# who switches streams arrives still bound to the one they started on.
|
|
# Re-bind them rather than 409: "this session is watching something else"
|
|
# is not a conflict a viewer can act on, and leaving it stuck meant every
|
|
# later heartbeat and command poll for this viewer failed for good.
|
|
viewer.stream_id = stream_id
|
|
viewer.remote_adjusted = False
|
|
if not viewer:
|
|
viewer = ViewerSession(session_id=session_id, stream_id=stream_id)
|
|
db.add(viewer)
|
|
try:
|
|
await db.flush()
|
|
except IntegrityError:
|
|
# A first heartbeat can be sent twice in quick succession (React
|
|
# StrictMode runs mount effects twice in development), and both
|
|
# requests can miss the lookup above before either commits. The loser
|
|
# adopts the winner's row instead of returning a 500.
|
|
#
|
|
# Recovery runs on its own session: this one is now in a failed
|
|
# transaction, and a session that is still dirty at teardown makes
|
|
# get_db commit on exit and re-raise the very IntegrityError we just
|
|
# handled.
|
|
async with session_factory() as recovery:
|
|
viewer = await recovery.get(ViewerSession, session_id)
|
|
if not viewer:
|
|
raise HTTPException(status_code=409, detail="观众会话创建失败，请刷新页面")
|
|
return {"ok": True, "remote_adjusted": viewer.remote_adjusted}
|
|
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": _utc_iso(viewer.last_heartbeat),
|
|
} 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": _utc_iso(command.created_at),
|
|
"claimed_at": _utc_iso(command.claimed_at),
|
|
} 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)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
extensions = {".mp4", ".m4v", ".mov", ".webm", ".mkv"}
|
|
return [p.name for p in root.iterdir() if p.is_file() and p.suffix.lower() in extensions]
|
|
|
|
|
|
@router.post("")
|
|
async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db),
|
|
_: bool = Depends(require_auth)):
|
|
title = body.title.strip()
|
|
if not title:
|
|
raise HTTPException(status_code=422, detail="节目标题不能为空")
|
|
_media_path(body.replay_file)
|
|
_valid_replay_url(body.replay_url)
|
|
if body.transcode_profile not in settings.transcode_profiles:
|
|
raise HTTPException(status_code=422, detail="转码配置不存在")
|
|
renditions = _normalize_renditions([item.model_dump() for item in body.renditions]) if body.renditions is not None else None
|
|
svc = StreamService(db)
|
|
try:
|
|
stream = await svc.create_stream(
|
|
title=title,
|
|
description=body.description,
|
|
cover_url=body.cover_url,
|
|
scheduled_at=body.scheduled_at,
|
|
replay_file=body.replay_file,
|
|
replay_url=body.replay_url,
|
|
chapters=body.chapters,
|
|
transcode_profile=body.transcode_profile,
|
|
renditions_json=json.dumps(renditions, ensure_ascii=False) if renditions is not None else None,
|
|
homepage_visible=body.homepage_visible,
|
|
remote_control_enabled=body.remote_control_enabled,
|
|
)
|
|
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,
|
|
"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 _serialize(s)
|
|
|
|
|
|
@router.put("/{stream_id}")
|
|
async def update_stream(stream_id: int, body: StreamUpdate, db: AsyncSession = Depends(get_db),
|
|
_: bool = Depends(require_auth)):
|
|
svc = StreamService(db)
|
|
existing = await svc.get_stream(stream_id)
|
|
if not existing:
|
|
raise HTTPException(status_code=404, detail="Stream not found")
|
|
updates = body.model_dump(exclude_unset=True)
|
|
replay_fields = {"replay_file", "replay_url", "chapters"}
|
|
if replay_fields.intersection(updates) and existing.status.value != "ended":
|
|
raise HTTPException(status_code=409, detail="仅已结束的节目可以配置回放和章节")
|
|
|
|
if updates.get("replay_file"):
|
|
_media_path(updates["replay_file"])
|
|
if "replay_url" in updates and updates["replay_url"] is not None:
|
|
updates["replay_url"] = _valid_replay_url(updates["replay_url"])
|
|
|
|
final_replay_file = updates.get("replay_file", existing.replay_file)
|
|
final_replay_url = updates.get("replay_url", existing.replay_url)
|
|
if final_replay_file and final_replay_url:
|
|
raise HTTPException(status_code=409, detail="本地回放文件和远端回放地址只能选择一种")
|
|
|
|
if "transcode_profile" in updates and updates["transcode_profile"] not in settings.transcode_profiles:
|
|
raise HTTPException(status_code=422, detail="转码配置不存在")
|
|
if "renditions" in updates:
|
|
renditions = updates.pop("renditions")
|
|
updates["renditions_json"] = (
|
|
json.dumps(_normalize_renditions(renditions), ensure_ascii=False)
|
|
if renditions is not None else None
|
|
)
|
|
if "chapters" in updates:
|
|
chapters = updates.pop("chapters")
|
|
if chapters and not (final_replay_file or final_replay_url):
|
|
raise HTTPException(status_code=409, detail="请先配置回放来源再编辑章节")
|
|
updates["chapters_json"] = json.dumps(_normalize_chapters(chapters or []), ensure_ascii=False)
|
|
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),
|
|
_: bool = Depends(require_auth)):
|
|
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, request: Request, 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")
|
|
client_ip = _get_client_ip(request)
|
|
urls = await svc.get_watch_urls(stream.stream_key, client_ip)
|
|
return {
|
|
"stream_id": stream.id,
|
|
"title": stream.title,
|
|
"status": stream.status.value,
|
|
"multicast_available": urls["multicast_available"],
|
|
"selected_node": urls.get("selected_node"),
|
|
"lines": urls["lines"],
|
|
"qualities": urls["qualities"],
|
|
"description": stream.description,
|
|
"scheduled_at": str(stream.scheduled_at) if stream.scheduled_at else None,
|
|
"replay_local_url": f"/api/streams/{stream.id}/replay" if stream.replay_file else None,
|
|
"replay_url": stream.replay_url,
|
|
"chapters": json.loads(stream.chapters_json or "[]"),
|
|
"remote_control_enabled": stream.remote_control_enabled,
|
|
}
|
|
|
|
|
|
@router.get("/{stream_id}/events")
|
|
async def stream_events(stream_id: int, request: Request, db: AsyncSession = Depends(get_db)):
|
|
if not await StreamService(db).get_stream(stream_id):
|
|
raise HTTPException(status_code=404, detail="Stream not found")
|
|
|
|
async def events():
|
|
previous = None
|
|
while not await request.is_disconnected():
|
|
stream = await StreamService(db).get_stream(stream_id)
|
|
if not stream:
|
|
yield "event: deleted\ndata: {}\n\n"
|
|
return
|
|
payload = json.dumps(_status_payload(stream), ensure_ascii=False)
|
|
if payload != previous:
|
|
yield f"event: status\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.get("/{stream_id}/replay")
|
|
async def serve_replay(stream_id: int, preview: str | None = None,
|
|
db: AsyncSession = Depends(get_db)):
|
|
stream = await StreamService(db).get_stream(stream_id)
|
|
if not stream:
|
|
raise HTTPException(status_code=404, detail="Program not found")
|
|
filename = preview if preview is not None else stream.replay_file
|
|
path = _media_path(filename)
|
|
if not path:
|
|
raise HTTPException(status_code=404, detail="Replay not available")
|
|
|
|
# Authentication and the path-traversal check happen above, in Python; the
|
|
# bytes are handed to nginx. Returning FileResponse here would stream the
|
|
# whole file through this worker, which for a multi-gigabyte replay means the
|
|
# file is copied twice and one worker is occupied for the entire transfer.
|
|
#
|
|
# nginx answers `X-Accel-Redirect` on its own response by serving the named
|
|
# internal location instead of relaying the body, so the client still sees a
|
|
# single normal response.
|
|
root = Path(settings.replay_media_dir).resolve()
|
|
relative = path.relative_to(root).as_posix()
|
|
# Quote each segment: replay names can contain spaces or non-ASCII characters,
|
|
# and nginx interprets the header as a URI.
|
|
location = "/media-internal/" + "/".join(quote(part) for part in relative.split("/"))
|
|
|
|
response = Response(status_code=200)
|
|
response.headers["X-Accel-Redirect"] = location
|
|
# Let the client buffer ahead and seek; nginx serves the range requests.
|
|
response.headers["Accept-Ranges"] = "bytes"
|
|
response.headers["Content-Type"] = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
|
|
# A direct hit on the backend (no nginx in front) must not look like a valid
|
|
# empty file, or a misconfigured deployment would silently play nothing.
|
|
response.headers["X-Replay-File"] = relative
|
|
return response
|
|
|
|
|
|
@router.post("/{stream_id}/remote-session")
|
|
async def create_remote_session(stream_id: int, body: RemoteSessionCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: bool = Depends(require_auth)):
|
|
stream = await StreamService(db).get_stream(stream_id)
|
|
if not stream or not stream.remote_control_enabled:
|
|
raise HTTPException(status_code=409, detail="节目未启用远程控制")
|
|
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 = (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="远程控制会话无效或已过期")
|
|
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="不支持的远程命令")
|
|
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, session_id: str,
|
|
db: AsyncSession = Depends(get_db)):
|
|
viewer = await db.get(ViewerSession, session_id)
|
|
if not viewer:
|
|
raise HTTPException(status_code=404, detail="观众会话不存在")
|
|
if viewer.stream_id != stream_id:
|
|
# A session id is stored per browser, not per programme, so a viewer who
|
|
# switches streams arrives with the id they registered on the old one.
|
|
# Remote control is meaningless if that 404s forever, so re-bind them
|
|
# instead — they are watching this stream now, which is the only fact the
|
|
# admin panel needs. Commands aimed at the old stream are not carried over.
|
|
viewer.stream_id = stream_id
|
|
viewer.remote_adjusted = False
|
|
await db.commit()
|
|
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": _utc_iso(command.created_at)}
|
|
|
|
|
|
@router.post("/{stream_id}/go-live")
|
|
async def stream_go_live(stream_id: int, db: AsyncSession = Depends(get_db),
|
|
_: bool = Depends(require_auth)):
|
|
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),
|
|
_: bool = Depends(require_auth)):
|
|
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, request: Request, 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")
|
|
client_ip = _get_client_ip(request)
|
|
urls = await svc.get_watch_urls(stream_key, client_ip)
|
|
return {
|
|
"stream_id": s.id,
|
|
"title": s.title,
|
|
"status": s.status.value,
|
|
"multicast_available": urls["multicast_available"],
|
|
"selected_node": urls.get("selected_node"),
|
|
"lines": urls["lines"],
|
|
"qualities": urls["qualities"],
|
|
}
|