""" Сервис для работы с LiveKit. Использует официальный livekit-api SDK (https://github.com/livekit/livekit) для генерации токенов. """ import uuid import logging from datetime import timedelta from django.conf import settings logger = logging.getLogger(__name__) try: from livekit import api LIVEKIT_API_AVAILABLE = True except ImportError: LIVEKIT_API_AVAILABLE = False logger.warning("livekit-api не установлен. Установите: pip install livekit-api") class LiveKitService: """Сервис для работы с LiveKit (официальный Go-сервер).""" @staticmethod def generate_room_name() -> str: """Генерация уникального имени комнаты.""" return str(uuid.uuid4()) @staticmethod def get_server_url(request=None) -> str: """ Публичный URL LiveKit для фронтенда. Приоритет: LIVEKIT_PUBLIC_URL > по request (порт 80/443 = nginx) > dev fallback. """ url = getattr(settings, 'LIVEKIT_PUBLIC_URL', '').strip() if url: return url if request: scheme = 'wss' if request.is_secure() else 'ws' host = request.get_host().split(':')[0] port = request.get_port() or (443 if request.is_secure() else 80) # Только если запрос через nginx (порт 80/443) if (request.is_secure() and port == 443) or (not request.is_secure() and port == 80): return f"{scheme}://{host}/livekit" return 'ws://127.0.0.1:7880' @staticmethod def get_ice_servers() -> list: """Список ICE серверов для WebRTC.""" default_ice_servers = [ {'urls': 'stun:stun.l.google.com:19302'}, ] custom = getattr(settings, 'LIVEKIT_ICE_SERVERS', None) if custom: import json return json.loads(custom) if isinstance(custom, str) else custom return default_ice_servers @staticmethod def generate_access_token( room_name: str, participant_name: str, participant_identity: str, is_admin: bool = False, expires_in_minutes: int = 180, metadata: str | None = None, ) -> str: """ Генерация JWT токена через официальный livekit-api. metadata — JSON-строка (например, {"board_id": "uuid"}) для передачи board_id. """ if not LIVEKIT_API_AVAILABLE: raise ValueError('livekit-api не установлен. Установите: pip install livekit-api') api_key = getattr(settings, 'LIVEKIT_API_KEY', '') api_secret = getattr(settings, 'LIVEKIT_API_SECRET', '') if not api_key or not api_secret: raise ValueError('LIVEKIT_API_KEY и LIVEKIT_API_SECRET должны быть установлены') identity_str = str(participant_identity).strip() if not identity_str or identity_str == 'None': identity_str = f"user_{uuid.uuid4().hex[:8]}" logger.warning('participant_identity пустой, сгенерирован fallback') name_str = str(participant_name).strip() if participant_name else identity_str if not name_str: name_str = identity_str grants = api.VideoGrants( room_join=True, room=room_name, can_publish=True, can_subscribe=True, can_publish_data=True, ) if is_admin: grants.room_admin = True grants.room_create = True grants.room_list = True token_builder = ( api.AccessToken(api_key=api_key, api_secret=api_secret) .with_identity(identity_str) .with_name(name_str) .with_grants(grants) .with_ttl(timedelta(minutes=expires_in_minutes)) ) if metadata: token_builder = token_builder.with_metadata(metadata) token = token_builder.to_jwt() logger.info(f'LiveKit token: identity={identity_str}, room={room_name}') return token