feat: expand livestream management platform
This commit is contained in:
1 parent
383c066d86
commit
ed62d42a63
12 files changed
+223
-24
No files matched your search
@@ -25,6 +25,7 @@ async def init_db():
|
||||
"replay_file": "VARCHAR(500)",
|
||||
"chapters_json": "TEXT",
|
||||
"transcode_profile": "VARCHAR(64) NOT NULL DEFAULT 'standard'",
|
||||
"renditions_json": "TEXT",
|
||||
"homepage_visible": "BOOLEAN NOT NULL DEFAULT 1",
|
||||
"remote_control_enabled": "BOOLEAN NOT NULL DEFAULT 0",
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ class Stream(Base):
|
||||
replay_file: Mapped[str | None] = mapped_column(String(500), nullable=True)
|
||||
chapters_json: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||||
transcode_profile: Mapped[str] = mapped_column(String(64), default="standard")
|
||||
renditions_json: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||||
homepage_visible: Mapped[bool] = mapped_column(Boolean, default=True)
|
||||
remote_control_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
|
||||
stream_key: Mapped[str] = mapped_column(String(64), unique=True, default=lambda: uuid.uuid4().hex)
|
||||
|
||||
@@ -105,6 +105,17 @@ async def get_system_metrics(_: bool = Depends(require_auth)):
|
||||
disk_path = metrics_path if metrics_path.exists() else Path.cwd()
|
||||
disk = shutil.disk_usage(disk_path)
|
||||
cpu_count = os.cpu_count() or 1
|
||||
network_in_bytes = 0
|
||||
network_out_bytes = 0
|
||||
netdev = Path("/proc/net/dev")
|
||||
if netdev.exists():
|
||||
for line in netdev.read_text(encoding="ascii").splitlines()[2:]:
|
||||
_, _, counters = line.partition(":")
|
||||
fields = counters.split()
|
||||
if len(fields) >= 9:
|
||||
network_in_bytes += int(fields[0])
|
||||
network_out_bytes += int(fields[8])
|
||||
|
||||
return {
|
||||
"cpu": {
|
||||
"count": cpu_count,
|
||||
@@ -123,4 +134,8 @@ async def get_system_metrics(_: bool = Depends(require_auth)):
|
||||
"free": disk.free,
|
||||
"used_percent": round(disk.used / disk.total * 100, 1) if disk.total else 0,
|
||||
},
|
||||
"network": {
|
||||
"bytes_recv": network_in_bytes,
|
||||
"bytes_sent": network_out_bytes,
|
||||
},
|
||||
}
|
||||
@@ -23,6 +23,20 @@ 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
|
||||
@@ -31,6 +45,7 @@ class StreamCreate(BaseModel):
|
||||
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
|
||||
|
||||
@@ -43,6 +58,7 @@ class StreamUpdate(BaseModel):
|
||||
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
|
||||
|
||||
@@ -115,6 +131,35 @@ def _normalize_chapters(chapters: list[dict]) -> list:
|
||||
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,
|
||||
@@ -126,6 +171,7 @@ def _serialize(s, include_private: bool = True):
|
||||
"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),
|
||||
@@ -272,6 +318,7 @@ async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db),
|
||||
_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(
|
||||
@@ -282,6 +329,7 @@ async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db),
|
||||
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,
|
||||
)
|
||||
@@ -323,6 +371,8 @@ async def update_stream(stream_id: int, body: StreamUpdate, db: AsyncSession = D
|
||||
_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="请先选择回放文件再编辑章节")
|
||||
|
||||
@@ -14,6 +14,17 @@ class StreamService:
|
||||
def __init__(self, db: AsyncSession):
|
||||
self.db = db
|
||||
|
||||
@staticmethod
|
||||
def effective_renditions(stream: Stream) -> list:
|
||||
if stream.renditions_json:
|
||||
try:
|
||||
renditions = json.loads(stream.renditions_json)
|
||||
if isinstance(renditions, list) and renditions:
|
||||
return renditions
|
||||
except (TypeError, ValueError, json.JSONDecodeError):
|
||||
pass
|
||||
return settings.transcode_profiles.get(stream.transcode_profile, settings.qualities)
|
||||
|
||||
async def create_stream(self, title: str, description: str | None = None,
|
||||
cover_url: str | None = None,
|
||||
scheduled_at: str | None = None,
|
||||
@@ -95,7 +106,7 @@ class StreamService:
|
||||
started_at=datetime.utcnow(),
|
||||
)
|
||||
rtmp_url = media_service.get_rtmp_url(stream_key)
|
||||
transcode_mgr.start_transcode(stream_key, rtmp_url, stream.transcode_profile)
|
||||
transcode_mgr.start_transcode(stream_key, rtmp_url, self.effective_renditions(stream))
|
||||
|
||||
async def set_stream_ended(self, stream_key: str):
|
||||
stream = await self.get_stream_by_key(stream_key)
|
||||
@@ -112,7 +123,7 @@ class StreamService:
|
||||
stream = await self.get_stream_by_key(stream_key)
|
||||
if not stream:
|
||||
return {"lines": [], "qualities": [], "multicast_available": False, "selected_node": None}
|
||||
qualities = settings.transcode_profiles.get(stream.transcode_profile, settings.qualities)
|
||||
qualities = self.effective_renditions(stream)
|
||||
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
|
||||
|
||||
@@ -6,33 +6,38 @@ from config import settings
|
||||
class TranscodeManager:
|
||||
_processes: dict[str, list[subprocess.Popen]] = {}
|
||||
|
||||
def start_transcode(self, stream_key: str, input_url: str, profile: str = "standard"):
|
||||
def start_transcode(self, stream_key: str, input_url: str, qualities: list[dict]):
|
||||
if stream_key in self._processes:
|
||||
return
|
||||
|
||||
procs = []
|
||||
qualities = settings.transcode_profiles.get(profile, settings.qualities)
|
||||
segment_seconds = max(1, settings.hls_segment_seconds)
|
||||
keyframe_frames = segment_seconds * 30
|
||||
for quality in qualities:
|
||||
label = quality["label"]
|
||||
width = quality["width"]
|
||||
height = quality["height"]
|
||||
fps = quality.get("fps", 30)
|
||||
codec = quality.get("codec", "libx264")
|
||||
bitrate = quality["bitrate"]
|
||||
crf = quality.get("crf", 23)
|
||||
preset = quality.get("preset", "ultrafast")
|
||||
audio_bitrate = quality.get("audio_bitrate", "128k")
|
||||
keyframe_seconds = quality.get("keyframe_seconds", segment_seconds)
|
||||
keyframe_frames = keyframe_seconds * fps
|
||||
hls_dir = f"/tmp/hls/{stream_key}/{label}"
|
||||
os.makedirs(hls_dir, exist_ok=True)
|
||||
cmd = [
|
||||
"ffmpeg", "-y",
|
||||
"-i", input_url,
|
||||
"-c:v", "libx264", "-preset", "ultrafast",
|
||||
"-b:v", bitrate,
|
||||
"-c:v", codec, "-preset", preset,
|
||||
"-r", str(fps), "-b:v", bitrate, "-crf", str(crf),
|
||||
"-s", f"{width}x{height}",
|
||||
"-sc_threshold", "0",
|
||||
"-g", str(keyframe_frames),
|
||||
"-keyint_min", str(keyframe_frames),
|
||||
"-force_key_frames", f"expr:gte(t,n_forced*{segment_seconds})",
|
||||
"-force_key_frames", f"expr:gte(t,n_forced*{keyframe_seconds})",
|
||||
"-copyts", "-start_at_zero",
|
||||
"-c:a", "aac", "-b:a", "128k",
|
||||
"-c:a", "aac", "-b:a", audio_bitrate,
|
||||
"-f", "hls",
|
||||
"-hls_time", str(segment_seconds),
|
||||
"-hls_list_size", "6",
|
||||
|
||||
Reference in new issue
Block a user