feat: add video room feature with API, database model, and UI integration

Author: DrizzleTime

api/routers.py

@@ -11,6 +11,7 @@
 from domain.processors import api as processors
 from domain.share import api as share
 from domain.tasks import api as tasks
+from domain.video_rooms import api as video_rooms
 from domain.ai import api as ai
 from domain.agent import api as agent
 from domain.virtual_fs import api as virtual_fs
@@ -32,6 +33,7 @@ def include_routers(app: FastAPI):
     app.include_router(config.router)
     app.include_router(processors.router)
     app.include_router(tasks.router)
+    app.include_router(video_rooms.router)
     app.include_router(share.router)
     app.include_router(share.public_router)
     app.include_router(backup.router)

domain/adapters/providers/quark.py

@@ -360,25 +360,14 @@ async def stream_file(self, root: str, rel: str, range_header: str | None):
             if tr:
                 url = tr
         dl_headers = self._download_headers()
-
-        # 预获取大小/是否支持范围
-        total_size: Optional[int] = None
-        async with httpx.AsyncClient(timeout=self._timeout, follow_redirects=True) as client:
-            try:
-                head_resp = await client.head(url, headers=dl_headers)
-                if head_resp.status_code == 200:
-                    cl = head_resp.headers.get("Content-Length")
-                    if cl and cl.isdigit():
-                        total_size = int(cl)
-            except Exception:
-                pass
+        file_size = int(it.get("size") or 0)
 
         mime, _ = mimetypes.guess_type(rel)
         content_type = mime or "application/octet-stream"
 
         # 解析 Range
         start = 0
-        end: Optional[int] = None
+        end: Optional[int] = file_size - 1 if file_size > 0 else None
         status_code = 200
         if range_header and range_header.startswith("bytes="):
             status_code = 206
@@ -388,35 +377,65 @@ async def stream_file(self, root: str, rel: str, range_header: str | None):
                 start = int(s)
             if e.strip():
                 end = int(e)
-
-        if total_size is not None and end is None and status_code == 206:
-            end = total_size - 1
-        if end is not None and total_size is not None and end >= total_size:
-            end = total_size - 1
-        if total_size is not None and start >= total_size:
+            elif file_size > 0:
+                end = file_size - 1
+            if file_size > 0:
+                if start >= file_size:
+                    raise HTTPException(416, detail="Requested Range Not Satisfiable")
+                if end is None or end >= file_size:
+                    end = file_size - 1
+                if start > end:
+                    raise HTTPException(416, detail="Requested Range Not Satisfiable")
+        headers = dict(dl_headers)
+        if status_code == 206:
+            headers["Range"] = f"bytes={start}-" if end is None else f"bytes={start}-{end}"
+
+        client = httpx.AsyncClient(timeout=None, follow_redirects=True)
+        req = client.build_request("GET", url, headers=headers)
+        resp = await client.send(req, stream=True)
+        if resp.status_code == 404:
+            await resp.aclose()
+            await client.aclose()
+            raise FileNotFoundError(rel)
+        if resp.status_code == 416:
+            await resp.aclose()
+            await client.aclose()
             raise HTTPException(416, detail="Requested Range Not Satisfiable")
+        try:
+            resp.raise_for_status()
+        except Exception:
+            await resp.aclose()
+            await client.aclose()
+            raise
 
-        resp_headers: Dict[str, str] = {"Accept-Ranges": "bytes", "Content-Type": content_type}
-        if status_code == 206 and total_size is not None and end is not None:
-            resp_headers["Content-Range"] = f"bytes {start}-{end}/{total_size}"
-            resp_headers["Content-Length"] = str(end - start + 1)
-        elif total_size is not None:
-            resp_headers["Content-Length"] = str(total_size)
+        resp_headers: Dict[str, str] = {
+            "Accept-Ranges": resp.headers.get("Accept-Ranges", "bytes"),
+            "Content-Type": resp.headers.get("Content-Type", content_type),
+        }
+        content_range = resp.headers.get("Content-Range")
+        content_length = resp.headers.get("Content-Length")
+        if content_range:
+            resp_headers["Content-Range"] = content_range
+        elif status_code == 206 and file_size > 0 and end is not None:
+            resp_headers["Content-Range"] = f"bytes {start}-{end}/{file_size}"
+        if content_length:
+            resp_headers["Content-Length"] = content_length
+        elif file_size > 0:
+            if status_code == 206 and end is not None:
+                resp_headers["Content-Length"] = str(end - start + 1)
+            elif resp.status_code == 200:
+                resp_headers["Content-Length"] = str(file_size)
 
         async def iterator():
-            headers = dict(dl_headers)
-            if status_code == 206 and end is not None:
-                headers["Range"] = f"bytes={start}-{end}"
-            async with httpx.AsyncClient(timeout=None, follow_redirects=True) as client:
-                async with client.stream("GET", url, headers=headers) as resp:
-                    if resp.status_code in (404, 416):
-                        await resp.aclose()
-                        raise HTTPException(resp.status_code, detail="Upstream not available")
-                    async for chunk in resp.aiter_bytes():
-                        if chunk:
-                            yield chunk
-
-        return StreamingResponse(iterator(), status_code=status_code, headers=resp_headers, media_type=content_type)
+            try:
+                async for chunk in resp.aiter_bytes():
+                    if chunk:
+                        yield chunk
+            finally:
+                await resp.aclose()
+                await client.aclose()
+
+        return StreamingResponse(iterator(), status_code=resp.status_code, headers=resp_headers, media_type=content_type)
 
     # -----------------
     # 上传（大文件分片）

domain/video_rooms/__init__.py

@@ -0,0 +1,9 @@
+from .service import VideoRoomService
+from .types import VideoRoomCreate, VideoRoomInfo, VideoRoomState
+
+__all__ = [
+    "VideoRoomService",
+    "VideoRoomCreate",
+    "VideoRoomInfo",
+    "VideoRoomState",
+]

domain/video_rooms/api.py

@@ -0,0 +1,72 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request, WebSocket, WebSocketDisconnect
+
+from api.response import success
+from domain.audit import AuditAction, audit
+from domain.auth import User, get_current_active_user
+from domain.permission import require_path_permission
+from domain.permission.types import PathAction
+from models.database import UserAccount
+from .service import VideoRoomService
+from .types import VideoRoomCreate, VideoRoomInfo
+from .ws import video_room_ws_manager
+
+router = APIRouter(prefix="/api/video-rooms", tags=["Video Rooms"])
+
+
+@router.post("", response_model=VideoRoomInfo)
+@audit(action=AuditAction.SHARE, description="创建视频房", body_fields=["name", "path"])
+@require_path_permission(PathAction.SHARE, "payload.path")
+async def create_video_room(
+    request: Request,
+    payload: VideoRoomCreate,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    user_account = await UserAccount.get(id=current_user.id)
+    room = await VideoRoomService.create_room(
+        user=user_account,
+        name=payload.name,
+        path=payload.path,
+    )
+    return VideoRoomInfo.from_orm(room, VideoRoomService.get_effective_state(room))
+
+
+@router.get("/{token}", response_model=VideoRoomInfo)
+@audit(action=AuditAction.SHARE, description="获取视频房信息")
+async def get_video_room(request: Request, token: str):
+    room = await VideoRoomService.get_room_by_token(token)
+    return VideoRoomInfo.from_orm(room, VideoRoomService.get_effective_state(room))
+
+
+@router.get("/{token}/stream")
+@audit(action=AuditAction.DOWNLOAD, description="播放视频房文件")
+async def stream_video_room(token: str, request: Request):
+    return await VideoRoomService.stream_room_file(token, request.headers.get("Range"))
+
+
+@router.websocket("/{token}/ws")
+async def video_room_ws(websocket: WebSocket, token: str):
+    room = await VideoRoomService.get_room_by_token(token)
+    await video_room_ws_manager.connect(token, websocket)
+    try:
+        state = VideoRoomService.get_effective_state(room)
+        await websocket.send_json({"type": "state", "state": state.model_dump()})
+        while True:
+            data = await websocket.receive_json()
+            if data.get("type") != "state":
+                continue
+            state = await VideoRoomService.update_state(
+                room,
+                current_time=float(data.get("current_time") or 0),
+                paused=bool(data.get("paused")),
+            )
+            await video_room_ws_manager.broadcast(
+                token,
+                {"type": "state", "state": state.model_dump()},
+                exclude=websocket,
+            )
+    except WebSocketDisconnect:
+        pass
+    finally:
+        video_room_ws_manager.disconnect(token, websocket)

domain/video_rooms/service.py

@@ -0,0 +1,82 @@
+import secrets
+from datetime import datetime, timezone
+from urllib.parse import quote
+
+from fastapi import HTTPException
+from fastapi.responses import Response
+
+from domain.virtual_fs import VirtualFSService
+from models.database import UserAccount, VideoRoom
+from .types import VideoRoomState
+
+
+VIDEO_EXTENSIONS = {".mp4", ".webm", ".ogg", ".m4v", ".mov", ".mkv", ".avi", ".flv"}
+
+
+class VideoRoomService:
+    @classmethod
+    def _is_video_path(cls, path: str) -> bool:
+        lower = path.lower()
+        return any(lower.endswith(ext) for ext in VIDEO_EXTENSIONS)
+
+    @classmethod
+    async def create_room(cls, user: UserAccount, name: str, path: str) -> VideoRoom:
+        if not path or path == "/" or ".." in path.split("/"):
+            raise HTTPException(status_code=400, detail="无效的视频路径")
+        if not cls._is_video_path(path):
+            raise HTTPException(status_code=400, detail="仅支持视频文件创建视频房")
+
+        stat = await VirtualFSService.stat_file(path)
+        if stat.get("is_dir"):
+            raise HTTPException(status_code=400, detail="目录不能创建视频房")
+
+        token = secrets.token_urlsafe(16)
+        return await VideoRoom.create(
+            token=token,
+            name=name or path.rsplit("/", 1)[-1],
+            path=path,
+            user=user,
+            state_updated_at=datetime.now(timezone.utc),
+        )
+
+    @classmethod
+    async def get_room_by_token(cls, token: str) -> VideoRoom:
+        room = await VideoRoom.get_or_none(token=token).prefetch_related("user")
+        if not room:
+            raise HTTPException(status_code=404, detail="视频房不存在")
+        return room
+
+    @classmethod
+    def get_effective_state(cls, room: VideoRoom) -> VideoRoomState:
+        current_time = float(room.current_time or 0)
+        updated_at = room.state_updated_at
+        if not room.paused and updated_at:
+            if updated_at.tzinfo is None:
+                updated_at = updated_at.replace(tzinfo=timezone.utc)
+            current_time += max(0, (datetime.now(timezone.utc) - updated_at).total_seconds())
+        return VideoRoomState(
+            current_time=max(0, current_time),
+            paused=bool(room.paused),
+            updated_at=updated_at.isoformat() if updated_at else None,
+        )
+
+    @classmethod
+    async def update_state(cls, room: VideoRoom, current_time: float, paused: bool) -> VideoRoomState:
+        now = datetime.now(timezone.utc)
+        room.current_time = max(0, float(current_time or 0))
+        room.paused = bool(paused)
+        room.state_updated_at = now
+        await room.save(update_fields=["current_time", "paused", "state_updated_at"])
+        return VideoRoomState(
+            current_time=room.current_time,
+            paused=room.paused,
+            updated_at=now.isoformat(),
+        )
+
+    @classmethod
+    async def stream_room_file(cls, token: str, range_header: str | None) -> Response:
+        room = await cls.get_room_by_token(token)
+        response = await VirtualFSService.stream_file(room.path, range_header)
+        filename = room.path.rsplit("/", 1)[-1]
+        response.headers["Content-Disposition"] = f"inline; filename*=UTF-8''{quote(filename)}"
+        return response

domain/video_rooms/types.py

@@ -0,0 +1,39 @@
+from pydantic import BaseModel
+
+from models.database import VideoRoom
+
+
+class VideoRoomCreate(BaseModel):
+    name: str
+    path: str
+
+
+class VideoRoomState(BaseModel):
+    current_time: float
+    paused: bool
+    updated_at: str | None = None
+
+
+class VideoRoomInfo(BaseModel):
+    id: int
+    token: str
+    name: str
+    path: str
+    created_at: str
+    state: VideoRoomState
+
+    @classmethod
+    def from_orm(cls, obj: VideoRoom, state: VideoRoomState | None = None):
+        return cls(
+            id=obj.id,
+            token=obj.token,
+            name=obj.name,
+            path=obj.path,
+            created_at=obj.created_at.isoformat(),
+            state=state
+            or VideoRoomState(
+                current_time=obj.current_time,
+                paused=obj.paused,
+                updated_at=obj.state_updated_at.isoformat() if obj.state_updated_at else None,
+            ),
+        )

domain/video_rooms/ws.py

@@ -0,0 +1,31 @@
+from fastapi import WebSocket
+
+
+class VideoRoomWebSocketManager:
+    def __init__(self):
+        self.rooms: dict[str, set[WebSocket]] = {}
+
+    async def connect(self, token: str, websocket: WebSocket):
+        await websocket.accept()
+        self.rooms.setdefault(token, set()).add(websocket)
+
+    def disconnect(self, token: str, websocket: WebSocket):
+        sockets = self.rooms.get(token)
+        if not sockets:
+            return
+        sockets.discard(websocket)
+        if not sockets:
+            self.rooms.pop(token, None)
+
+    async def broadcast(self, token: str, message: dict, exclude: WebSocket | None = None):
+        sockets = list(self.rooms.get(token, set()))
+        for socket in sockets:
+            if socket is exclude:
+                continue
+            try:
+                await socket.send_json(message)
+            except Exception:
+                self.disconnect(token, socket)
+
+
+video_room_ws_manager = VideoRoomWebSocketManager()

models/database.py

@@ -234,6 +234,23 @@ class Meta:
         table = "share_links"
 
 
+class VideoRoom(Model):
+    id = fields.IntField(pk=True)
+    token = fields.CharField(max_length=100, unique=True, index=True)
+    name = fields.CharField(max_length=255)
+    path = fields.CharField(max_length=4096)
+    user: fields.ForeignKeyRelation[UserAccount] = fields.ForeignKeyField(
+        "models.UserAccount", related_name="video_rooms", on_delete=fields.CASCADE
+    )
+    current_time = fields.FloatField(default=0)
+    paused = fields.BooleanField(default=True)
+    state_updated_at = fields.DatetimeField(null=True)
+    created_at = fields.DatetimeField(auto_now_add=True)
+
+    class Meta:
+        table = "video_rooms"
+
+
 class RecentFile(Model):
     id = fields.IntField(pk=True)
     user: fields.ForeignKeyRelation[UserAccount] = fields.ForeignKeyField(