579 lines
24 KiB
Python
579 lines
24 KiB
Python
import json
|
|
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
|
|
from database import get_db
|
|
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
|
|
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
|
|
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 _status_payload(stream) -> dict:
|
|
return {
|
|
"stream_id": stream.id,
|
|
"status": stream.status.value,
|
|
"replay_url": f"/api/streams/{stream.id}/replay" if stream.replay_file else None,
|
|
"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,
|
|
"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:
|
|
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)
|
|
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)
|
|
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,
|
|
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 = {k: v for k, v in body.model_dump().items() if v is not None}
|
|
if ("replay_file" in updates or "chapters" in updates) and existing.status.value != "ended":
|
|
raise HTTPException(status_code=409, detail="仅已结束的节目可以配置回放和章节")
|
|
if "replay_file" in updates:
|
|
_media_path(updates["replay_file"])
|
|
if "transcode_profile" in updates and updates["transcode_profile"] not in settings.transcode_profiles:
|
|
raise HTTPException(status_code=422, detail="转码配置不存在")
|
|
if "renditions" in updates:
|
|
updates["renditions_json"] = json.dumps(_normalize_renditions(updates.pop("renditions")), ensure_ascii=False)
|
|
if "chapters" in updates:
|
|
if not updates.get("replay_file") and not existing.replay_file:
|
|
raise HTTPException(status_code=409, detail="请先选择回放文件再编辑章节")
|
|
chapters = _normalize_chapters(updates.pop("chapters"))
|
|
updates["chapters_json"] = json.dumps(chapters, 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_url": f"/api/streams/{stream.id}/replay" if stream.replay_file else None,
|
|
"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")
|
|
path = _media_path(preview if preview is not None else stream.replay_file)
|
|
if not path:
|
|
raise HTTPException(status_code=404, detail="Replay not available")
|
|
return FileResponse(path)
|
|
|
|
|
|
@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 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")
|
|
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"],
|
|
}
|