diff --git a/backend/config.py b/backend/config.py
index 13b010c..27af34f 100644
--- a/backend/config.py
+++ b/backend/config.py
@@ -10,6 +10,8 @@ class Settings(BaseSettings):
media_origin_rtmp_port: int = 1935
media_origin_api_port: int = 9997
media_origin_hls_port: int = 8888
+ replay_media_dir: str = "/workspace/media"
+ hls_segment_seconds: int = 4
edge_nodes: list[dict] = [
{
@@ -34,6 +36,15 @@ class Settings(BaseSettings):
{"label": "360p", "width": 640, "height": 360, "bitrate": "600k"},
]
+ transcode_profiles: dict[str, list[dict]] = {
+ "standard": qualities,
+ "bandwidth-saver": [
+ {"label": "720p", "width": 1280, "height": 720, "bitrate": "1800k"},
+ {"label": "480p", "width": 854, "height": 480, "bitrate": "800k"},
+ {"label": "360p", "width": 640, "height": 360, "bitrate": "450k"},
+ ],
+ }
+
multicast_enabled: bool = True
multicast_address_base: str = "239.0.0.1"
multicast_port_base: int = 5000
diff --git a/backend/database.py b/backend/database.py
index 1c415e7..50a442c 100644
--- a/backend/database.py
+++ b/backend/database.py
@@ -1,5 +1,6 @@
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy import text
from config import settings
@@ -19,3 +20,16 @@ async def get_db() -> AsyncSession:
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
+ if settings.database_url.startswith("sqlite"):
+ columns = {
+ "replay_file": "VARCHAR(500)",
+ "chapters_json": "TEXT",
+ "transcode_profile": "VARCHAR(64) NOT NULL DEFAULT 'standard'",
+ "homepage_visible": "BOOLEAN NOT NULL DEFAULT 1",
+ "remote_control_enabled": "BOOLEAN NOT NULL DEFAULT 0",
+ }
+ rows = await conn.execute(text("PRAGMA table_info(streams)"))
+ existing = {row[1] for row in rows}
+ for name, definition in columns.items():
+ if name not in existing:
+ await conn.execute(text(f"ALTER TABLE streams ADD COLUMN {name} {definition}"))
diff --git a/backend/main.py b/backend/main.py
index 480162c..30e7d0c 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -66,7 +66,7 @@ async def lifespan(app: FastAPI):
transcode_mgr.stop_all()
-app = FastAPI(title="StreamCore", lifespan=lifespan)
+app = FastAPI(title="ProgramCore", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
diff --git a/backend/models/stream.py b/backend/models/stream.py
index 836fbfb..14c5ad7 100644
--- a/backend/models/stream.py
+++ b/backend/models/stream.py
@@ -20,6 +20,11 @@ class Stream(Base):
title: Mapped[str] = mapped_column(String(200), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
cover_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
+ 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")
+ 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)
status: Mapped[StreamStatus] = mapped_column(Enum(StreamStatus), default=StreamStatus.pending)
scheduled_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
diff --git a/backend/routes/streams.py b/backend/routes/streams.py
index 9764e26..3e16d51 100644
--- a/backend/routes/streams.py
+++ b/backend/routes/streams.py
@@ -1,6 +1,12 @@
-from fastapi import APIRouter, HTTPException, Depends, Query, Request
+import json
+import secrets
+import time
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException, Depends, Request
+from fastapi.responses import FileResponse
from sqlalchemy.ext.asyncio import AsyncSession
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
from typing import Optional
from database import get_db
from routes.auth import require_auth
@@ -17,6 +23,11 @@ class StreamCreate(BaseModel):
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"
+ homepage_visible: bool = True
+ remote_control_enabled: bool = False
class StreamUpdate(BaseModel):
@@ -24,6 +35,55 @@ class StreamUpdate(BaseModel):
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
+ homepage_visible: Optional[bool] = None
+ remote_control_enabled: Optional[bool] = None
+
+
+class RemoteSessionCreate(BaseModel):
+ expires_in: int = 300
+
+
+class RemoteCommand(BaseModel):
+ token: str
+ action: str
+ value: float | None = None
+
+
+_remote_sessions: dict[str, dict] = {}
+_remote_commands: dict[int, list[dict]] = {}
+
+
+def _media_path(filename: str | None) -> Path | None:
+ if not filename:
+ return None
+ root = Path(settings.replay_media_dir).resolve()
+ candidate = (root / filename).resolve()
+ if root not in candidate.parents or not candidate.is_file():
+ raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
+ return candidate
+
+
+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,
+ "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:
@@ -39,25 +99,18 @@ def _get_client_ip(request: Request) -> str | None:
@router.get("")
-async def list_streams(db: AsyncSession = Depends(get_db)):
+async def list_streams(admin: bool = False, db: AsyncSession = Depends(get_db)):
svc = StreamService(db)
- streams = await svc.get_all_streams()
- return [
- {
- "id": s.id,
- "title": s.title,
- "description": s.description,
- "cover_url": s.cover_url,
- "stream_key": s.stream_key,
- "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,
- "created_at": str(s.created_at),
- }
- for s in streams
- ]
+ 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("/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("")
@@ -65,7 +118,10 @@ 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="直播标题不能为空")
+ 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="转码配置不存在")
svc = StreamService(db)
try:
stream = await svc.create_stream(
@@ -73,12 +129,17 @@ async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db),
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,
+ 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
+ raise HTTPException(status_code=500, detail="创建节目失败,请检查日期格式和数据库状态") from exc
rtmp_url = media_service.get_rtmp_url(stream.stream_key)
return {
"id": stream.id,
@@ -95,19 +156,7 @@ async def get_stream(stream_id: int, db: AsyncSession = Depends(get_db)):
s = await svc.get_stream(stream_id)
if not s:
raise HTTPException(status_code=404, detail="Stream not found")
- return {
- "id": s.id,
- "title": s.title,
- "description": s.description,
- "cover_url": s.cover_url,
- "stream_key": s.stream_key,
- "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,
- "created_at": str(s.created_at),
- }
+ return _serialize(s)
@router.put("/{stream_id}")
@@ -118,6 +167,12 @@ async def update_stream(stream_id: int, body: StreamUpdate, db: AsyncSession = D
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:
+ _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 "chapters" in updates:
+ updates["chapters_json"] = json.dumps(updates.pop("chapters"), ensure_ascii=False)
if updates:
await svc.update_stream(stream_id, **updates)
return {"ok": True}
@@ -150,9 +205,62 @@ async def get_watch_urls(stream_id: int, request: Request, db: AsyncSession = De
"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}/replay")
+async def serve_replay(stream_id: int, 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(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="节目未启用远程控制")
+ 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)}
+
+
+@router.post("/{stream_id}/remote-command")
+async def send_remote_command(stream_id: int, body: RemoteCommand,
+ _: 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():
+ raise HTTPException(status_code=403, detail="远程控制会话无效或已过期")
+ if body.action not in {"play", "pause", "seek"}:
+ 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}
+
+
+@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
+
+
@router.post("/{stream_id}/go-live")
async def stream_go_live(stream_id: int, db: AsyncSession = Depends(get_db),
_: bool = Depends(require_auth)):
diff --git a/backend/services/stream_service.py b/backend/services/stream_service.py
index d699034..88b8e7d 100644
--- a/backend/services/stream_service.py
+++ b/backend/services/stream_service.py
@@ -16,7 +16,12 @@ class StreamService:
async def create_stream(self, title: str, description: str | None = None,
cover_url: str | None = None,
- scheduled_at: str | None = None) -> Stream:
+ scheduled_at: str | None = None,
+ replay_file: str | None = None,
+ chapters: list[dict] | None = None,
+ transcode_profile: str = "standard",
+ homepage_visible: bool = True,
+ remote_control_enabled: bool = False) -> Stream:
import uuid
scheduled_datetime = None
if scheduled_at:
@@ -28,6 +33,11 @@ class StreamService:
title=title,
description=description,
cover_url=cover_url,
+ replay_file=replay_file,
+ chapters_json=json.dumps(chapters or [], ensure_ascii=False),
+ transcode_profile=transcode_profile,
+ homepage_visible=homepage_visible,
+ remote_control_enabled=remote_control_enabled,
stream_key=uuid.uuid4().hex,
status=StreamStatus.pending,
scheduled_at=scheduled_datetime,
@@ -43,6 +53,14 @@ class StreamService:
)
return result.scalars().all()
+ async def get_public_streams(self):
+ result = await self.db.execute(
+ select(Stream)
+ .where(Stream.homepage_visible.is_(True))
+ .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)
@@ -77,7 +95,7 @@ class StreamService:
started_at=datetime.utcnow(),
)
rtmp_url = media_service.get_rtmp_url(stream_key)
- transcode_mgr.start_transcode(stream_key, rtmp_url)
+ transcode_mgr.start_transcode(stream_key, rtmp_url, stream.transcode_profile)
async def set_stream_ended(self, stream_key: str):
stream = await self.get_stream_by_key(stream_key)
@@ -91,6 +109,10 @@ class StreamService:
transcode_mgr.stop_transcode(stream_key)
async def get_watch_urls(self, stream_key: str, client_ip: str | None = None):
+ 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)
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
@@ -106,7 +128,7 @@ class StreamService:
"url": f"{base_hls}/index.m3u8",
"qualities": [],
}
- for quality in settings.qualities:
+ for quality in qualities:
label = quality["label"]
line_info["qualities"].append({
"label": label,
@@ -118,7 +140,7 @@ class StreamService:
return {
"lines": lines,
- "qualities": [{"label": q["label"], "width": q["width"], "height": q["height"]} for q in settings.qualities],
+ "qualities": [{"label": q["label"], "width": q["width"], "height": q["height"]} for q in qualities],
"multicast_available": multicast_detector.is_available(),
"selected_node": {
"id": best_node["id"],
diff --git a/backend/services/transcoder.py b/backend/services/transcoder.py
index a0791cd..f668d29 100644
--- a/backend/services/transcoder.py
+++ b/backend/services/transcoder.py
@@ -1,4 +1,3 @@
-import asyncio
import subprocess
import os
from config import settings
@@ -7,12 +6,15 @@ from config import settings
class TranscodeManager:
_processes: dict[str, list[subprocess.Popen]] = {}
- def start_transcode(self, stream_key: str, input_url: str):
+ def start_transcode(self, stream_key: str, input_url: str, profile: str = "standard"):
if stream_key in self._processes:
return
procs = []
- for quality in settings.qualities:
+ 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"]
@@ -25,11 +27,16 @@ class TranscodeManager:
"-c:v", "libx264", "-preset", "ultrafast",
"-b:v", bitrate,
"-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})",
+ "-copyts", "-start_at_zero",
"-c:a", "aac", "-b:a", "128k",
"-f", "hls",
- "-hls_time", "4",
+ "-hls_time", str(segment_seconds),
"-hls_list_size", "6",
- "-hls_flags", "delete_segments",
+ "-hls_flags", "delete_segments+independent_segments+program_date_time",
"-hls_segment_filename", f"{hls_dir}/segment_%03d.ts",
f"{hls_dir}/index.m3u8",
]
diff --git a/frontend/src/admin/AdminLayout.tsx b/frontend/src/admin/AdminLayout.tsx
index 05c8254..4a64d2d 100644
--- a/frontend/src/admin/AdminLayout.tsx
+++ b/frontend/src/admin/AdminLayout.tsx
@@ -13,8 +13,8 @@ const { Text } = Typography
const menuItems = [
{ key: '/admin', icon: