You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
74 lines
2.9 KiB
74 lines
2.9 KiB
import json
|
|
import hmac
|
|
import hashlib
|
|
from typing import Any, Dict, Optional
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.core.exceptions import ImproperlyConfigured
|
|
|
|
|
|
class PlugNMeetError(Exception):
|
|
def __init__(self, message: str, *, status_code: Optional[int] = None, response_data: Optional[Dict[str, Any]] = None):
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.response_data = response_data or {}
|
|
|
|
|
|
class PlugNMeetClient:
|
|
def __init__(self, *, base_url: Optional[str] = None, api_key: Optional[str] = None, api_secret: Optional[str] = None, timeout: Optional[float] = None):
|
|
self.base_url = (base_url or getattr(settings, "PLUGNMEET_SERVER_URL", "")).rstrip("/")
|
|
self.api_key = api_key or getattr(settings, "PLUGNMEET_API_KEY", "")
|
|
self.api_secret = api_secret or getattr(settings, "PLUGNMEET_API_SECRET", "")
|
|
self.timeout = timeout or getattr(settings, "PLUGNMEET_TIMEOUT", 10.0)
|
|
|
|
if not self.base_url or not self.api_key or not self.api_secret:
|
|
raise ImproperlyConfigured("PlugNMeet integration settings are incomplete.")
|
|
|
|
def create_room(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
return self._post("/auth/room/create", payload)
|
|
|
|
def get_join_token(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
return self._post("/auth/room/getJoinToken", payload)
|
|
|
|
def _post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
url = urljoin(f"{self.base_url}/", path.lstrip("/"))
|
|
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"API-KEY": self.api_key,
|
|
"HASH-SIGNATURE": self._build_signature(body),
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, data=body.encode("utf-8"), headers=headers, timeout=self.timeout)
|
|
except requests.RequestException as exc:
|
|
raise PlugNMeetError("Failed to reach PlugNMeet server.") from exc
|
|
|
|
if response.status_code >= 400:
|
|
response_data = self._safe_json(response)
|
|
raise PlugNMeetError(
|
|
"PlugNMeet server returned an error.",
|
|
status_code=response.status_code,
|
|
response_data=response_data,
|
|
)
|
|
|
|
data = self._safe_json(response)
|
|
if data is None:
|
|
raise PlugNMeetError("PlugNMeet server returned an invalid response format.")
|
|
return data
|
|
|
|
def _build_signature(self, body: str) -> str:
|
|
digest = hmac.new(self.api_secret.encode("utf-8"), body.encode("utf-8"), hashlib.sha256)
|
|
return digest.hexdigest()
|
|
|
|
@staticmethod
|
|
def _safe_json(response: requests.Response) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
__all__ = ["PlugNMeetClient", "PlugNMeetError"]
|