完善节目管理与回放播放体验

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)):