admin
980b91f41f
- 新增 subnet-aware 负载策略: 根据客户端IP匹配最近/16子网节点 - 节点注册端点 (POST /api/nodes/register): 边缘节点自动注册 - 节点心跳端点 (POST /api/nodes/heartbeat): 上报CPU/RAM/带宽/连接数 - 健康汇总 (GET /api/nodes/health): 服务端汇总所有节点状态 - 配置节点新增subnet/is_origin字段 - 前端NodeManager: 实时健康状态/CPU/RAM进度条/带宽/连接数 - X-Forwarded-For/X-Real-IP透传支持客户端IP
117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
from sqlalchemy import select, update, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from models.stream import Stream, StreamStatus
|
|
from services.media_service import media_service
|
|
from services.load_balancer import lb
|
|
from services.multicast_detector import multicast_detector
|
|
from services.transcoder import transcode_mgr
|
|
from config import settings
|
|
import json
|
|
|
|
|
|
class StreamService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def create_stream(self, title: str, description: str | None = None,
|
|
cover_url: str | None = None,
|
|
scheduled_at: str | None = None) -> Stream:
|
|
import uuid
|
|
stream = Stream(
|
|
title=title,
|
|
description=description,
|
|
cover_url=cover_url,
|
|
stream_key=uuid.uuid4().hex,
|
|
status=StreamStatus.pending,
|
|
scheduled_at=scheduled_at,
|
|
)
|
|
self.db.add(stream)
|
|
await self.db.commit()
|
|
await self.db.refresh(stream)
|
|
return stream
|
|
|
|
async def get_all_streams(self):
|
|
result = await self.db.execute(
|
|
select(Stream).order_by(Stream.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def get_stream(self, stream_id: int):
|
|
result = await self.db.execute(
|
|
select(Stream).where(Stream.id == stream_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_stream_by_key(self, stream_key: str):
|
|
result = await self.db.execute(
|
|
select(Stream).where(Stream.stream_key == stream_key)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def update_stream(self, stream_id: int, **kwargs):
|
|
stmt = update(Stream).where(Stream.id == stream_id).values(**kwargs)
|
|
await self.db.execute(stmt)
|
|
await self.db.commit()
|
|
|
|
async def delete_stream(self, stream_id: int):
|
|
stream = await self.get_stream(stream_id)
|
|
if stream:
|
|
transcode_mgr.stop_transcode(stream.stream_key)
|
|
await self.db.delete(stream)
|
|
await self.db.commit()
|
|
|
|
async def set_stream_live(self, stream_key: str):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if stream:
|
|
from datetime import datetime
|
|
await self.update_stream(
|
|
stream.id,
|
|
status=StreamStatus.live,
|
|
started_at=datetime.utcnow(),
|
|
)
|
|
rtmp_url = media_service.get_rtmp_url(stream_key)
|
|
transcode_mgr.start_transcode(stream_key, rtmp_url)
|
|
|
|
async def set_stream_ended(self, stream_key: str):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if stream:
|
|
from datetime import datetime
|
|
await self.update_stream(
|
|
stream.id,
|
|
status=StreamStatus.ended,
|
|
ended_at=datetime.utcnow(),
|
|
)
|
|
transcode_mgr.stop_transcode(stream_key)
|
|
|
|
async def get_watch_urls(self, stream_key: str, client_ip: str | None = None):
|
|
best_node = lb.select_best_node(client_ip)
|
|
if not best_node:
|
|
best_node = lb.get_enabled_nodes()[0] if lb.get_enabled_nodes() else None
|
|
|
|
lines = []
|
|
if best_node:
|
|
base_hls = f"/hls/{stream_key}"
|
|
line_info = {
|
|
"node_id": best_node["id"],
|
|
"node_name": best_node["name"],
|
|
"weight": best_node["weight"],
|
|
"type": "hls",
|
|
"url": f"{base_hls}/index.m3u8",
|
|
"qualities": [],
|
|
}
|
|
for quality in settings.qualities:
|
|
label = quality["label"]
|
|
line_info["qualities"].append({
|
|
"label": label,
|
|
"width": quality["width"],
|
|
"height": quality["height"],
|
|
"url": f"{base_hls}/{label}/index.m3u8",
|
|
})
|
|
lines.append(line_info)
|
|
|
|
return {
|
|
"lines": lines,
|
|
"qualities": [{"label": q["label"], "width": q["width"], "height": q["height"]} for q in settings.qualities],
|
|
"multicast_available": multicast_detector.is_available(),
|
|
}
|