Foxel/domain/video_room/service.py

import asyncio
import json
import secrets
from datetime import datetime, timedelta, timezone
from typing import Optional

from fastapi import HTTPException, WebSocket
from starlette.websockets import WebSocketState

from models.database import VideoRoom


VIDEO_EXTS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".m4v"}


class VideoRoomService:
    _runtime_states: dict[int, dict] = {}
    _room_clients: dict[int, set[WebSocket]] = {}
    _lock = asyncio.Lock()

    @classmethod
    def _now(cls) -> datetime:
        return datetime.now(timezone.utc)

    @classmethod
    def _iso_now(cls) -> str:
        return cls._now().isoformat()

    @classmethod
    def _ensure_video_path(cls, path: str) -> None:
        p = path.lower()
        if not any(p.endswith(ext) for ext in VIDEO_EXTS):
            raise HTTPException(status_code=400, detail="仅支持视频文件创建视频间")

    @classmethod
    def _calc_expires_at(cls, expires_in_days: Optional[int]) -> Optional[datetime]:
        if expires_in_days is None or expires_in_days <= 0:
            return None
        return cls._now() + timedelta(days=expires_in_days)

    @classmethod
    async def create_room(cls, *, user_id: int, path: str, name: Optional[str], expires_in_days: Optional[int], control_mode: str):
        cls._ensure_video_path(path)
        token = secrets.token_urlsafe(18)
        room_name = name or f"{path.split('/')[-1]} 的视频间"
        room = await VideoRoom.create(
            token=token,
            name=room_name,
            path=path,
            owner_id=user_id,
            control_mode=control_mode,
            expires_at=cls._calc_expires_at(expires_in_days),
        )
        cls._runtime_states[room.id] = {
            "position_ms": 0,
            "is_paused": True,
            "playback_rate": 1.0,
            "updated_at": cls._iso_now(),
            "updated_by": f"user:{user_id}",
        }
        return room

    @classmethod
    async def get_room_by_token(cls, token: str) -> VideoRoom:
        room = await VideoRoom.get_or_none(token=token)
        if not room:
            raise HTTPException(status_code=404, detail="视频间不存在")
        if room.expires_at and room.expires_at < cls._now():
            raise HTTPException(status_code=410, detail="视频间已过期")
        return room

    @classmethod
    async def get_state(cls, room_id: int) -> dict:
        return cls._runtime_states.setdefault(
            room_id,
            {
                "position_ms": 0,
                "is_paused": True,
                "playback_rate": 1.0,
                "updated_at": cls._iso_now(),
                "updated_by": "system",
            },
        )

    @classmethod
    async def apply_event(cls, *, room: VideoRoom, actor: str, event_type: str, position_ms: int, playback_rate: float):
        state = await cls.get_state(room.id)
        if room.control_mode == "host_only" and actor != f"user:{room.owner_id}":
            raise HTTPException(status_code=403, detail="仅房主可控制播放")

        if event_type == "play":
            state["is_paused"] = False
        elif event_type == "pause":
            state["is_paused"] = True
        elif event_type == "seek":
            state["position_ms"] = max(position_ms, 0)
        elif event_type == "rate":
            state["playback_rate"] = playback_rate
        state["updated_at"] = cls._iso_now()
        state["updated_by"] = actor
        return state

    @classmethod
    async def ws_connect(cls, room_id: int, websocket: WebSocket):
        await websocket.accept()
        async with cls._lock:
            clients = cls._room_clients.setdefault(room_id, set())
            clients.add(websocket)

    @classmethod
    async def ws_disconnect(cls, room_id: int, websocket: WebSocket):
        async with cls._lock:
            clients = cls._room_clients.get(room_id)
            if not clients:
                return
            clients.discard(websocket)
            if not clients:
                cls._room_clients.pop(room_id, None)

    @classmethod
    async def ws_broadcast(cls, room_id: int, payload: dict):
        clients = list(cls._room_clients.get(room_id, set()))
        if not clients:
            return
        text = json.dumps(payload)
        stale: list[WebSocket] = []
        for ws in clients:
            try:
                if ws.application_state == WebSocketState.CONNECTED:
                    await ws.send_text(text)
                else:
                    stale.append(ws)
            except Exception:
                stale.append(ws)

        if stale:
            async with cls._lock:
                pool = cls._room_clients.get(room_id, set())
                for ws in stale:
                    pool.discard(ws)
                if not pool:
                    cls._room_clients.pop(room_id, None)