"""Data structures for BigBLueButton meetings and handling code"""
import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Dict, Optional
from ._caching import cache
from .attendee import Attendee
from .util import camel_to_snake, snake_to_camel, to_field_type
if TYPE_CHECKING: # pragma: no cover
from .bigbluebutton import BigBlueButton
logger = logging.getLogger(__name__)
[docs]@dataclass
class Meeting:
"""Define the meta-data of a meeting and its state.
The meeting is always linked to one :class:`~bigbluebutton.api.bigbluebutton.BigBlueButton`,
on which it executes any API calls.
"""
api: "BigBlueButton" # noqa: F821
meeting_id: Optional[str] = None
meeting_name: Optional[str] = field(default=None, compare=False)
attendee_pw: Optional[str] = field(default=None, compare=False)
moderator_pw: Optional[str] = field(default=None, compare=False)
welcome: Optional[str] = field(default=None, compare=False)
dial_number: Optional[str] = field(default=None, compare=False)
voice_bridge: Optional[str] = None
max_participants: Optional[int] = field(default=None, compare=False)
logout_url: Optional[str] = field(default=None, compare=False)
record: Optional[bool] = field(default=None, compare=False)
duration: Optional[int] = field(default=None, compare=False)
meta: Dict[str, str] = field(default_factory=dict)
moderator_only_message: Optional[str] = field(default=None, compare=False)
auto_start_recording: Optional[bool] = field(default=None, compare=False)
allow_start_stop_recording: Optional[bool] = field(default=None, compare=False)
webcams_only_for_moderator: Optional[bool] = field(default=None, compare=False)
logo: Optional[str] = field(default=None, compare=False)
banner_text: Optional[str] = field(default=None, compare=False)
banner_color: Optional[str] = field(default=None, compare=False)
copyright: Optional[str] = field(default=None, compare=False) # noqa: A003
mute_on_start: Optional[bool] = field(default=None, compare=False)
allow_mods_to_unmute_users: Optional[bool] = field(default=None, compare=False)
lock_settings_disable_cam: Optional[bool] = field(default=None, compare=False)
lock_settings_disable_mic: Optional[bool] = field(default=None, compare=False)
lock_settings_disable_private_chat: Optional[bool] = field(default=None, compare=False)
lock_settings_disable_public_chat: Optional[bool] = field(default=None, compare=False)
lock_settings_disable_note: Optional[bool] = field(default=None, compare=False)
lock_settings_locked_layout: Optional[bool] = field(default=None, compare=False)
lock_settings_lock_on_join: Optional[bool] = field(default=None, compare=False)
lock_settings_lock_on_join_configurable: Optional[bool] = field(default=None, compare=False)
guest_policy: Optional[str] = field(default=None, compare=False)
create_time: int = 0
running: bool = field(default=False, init=False, compare=False)
has_user_joined: bool = field(default=False, init=False, compare=False)
recording: bool = field(default=False, init=False, compare=False)
has_been_forcibly_ended: bool = field(default=False, init=False, compare=False)
start_time: int = field(default=0, init=False, compare=False)
end_time: int = field(default=0, init=False, compare=False)
participant_count: int = field(default=0, init=False, compare=False)
listener_count: int = field(default=0, init=False, compare=False)
voice_participant_count: int = field(default=0, init=False, compare=False)
video_count: int = field(default=0, init=False, compare=False)
max_users: int = field(default=0, init=False, compare=False)
moderator_count: int = field(default=0, init=False, compare=False)
@property
def origin(self) -> str:
"""Origin of the meeting
Derived from the meta-data values bbb-origin and bbb-origin-server-name,
which are de facto standard fields.
"""
origin = self.meta.get("bbb-origin", "unknown")
server_name = self.meta.get("bbb-origin-server-name", "")
return f"{origin} ({server_name})"
def __post_init__(self):
"""Set defaults for all unset, but mandatory, attributes.
meeting_id is generated by the
:meth:`~bigbluebutton.api.bigbluebutton.BigBlueButtonGroup.generate_meeting_id`
method, which can be overridden by any client using the library.
"""
self.meeting_id = self.meeting_id or self.api.group.generate_meeting_id()
self.logout_url = self.logout_url or self.api.group.logout_url
self.attendees = {}
self.api.meetings[self.meeting_id] = self
[docs] def create(self) -> None:
"""Create the meeting on the linked BigBlueButton server."""
# Set origin metadata from API defaults if not defined
self.meta.setdefault("bbb-origin", self.api.group.origin)
self.meta.setdefault("bbb-origin-server-name", self.api.group.origin_server_name)
logger.info(f"Creating meeting {self.meeting_id} on server {self.api.name}")
res = self.api._request("create", self._get_url_args(meeting_name="name"))
self._update_from_response(res)
# We need to refresh all data after the create call
# FIXME Check whether this can be done less time-consuming
self.get_meeting_info()
[docs] def is_meeting_running(self) -> bool:
"""Call the isMeetingRunning API method.
The result is cached in the `running` attribute and immediately returned.
:return: True if the server reports the meeting as running, False if not
"""
res = self.api._request("isMeetingRunning", self._get_url_args("meeting_id"))
self._update_from_response(res)
return self.running
[docs] def get_meeting_info(self) -> "Meeting":
"""Update the meeting information from the server.
The results are cached in the corresponding attributes.
:return: the Meeting object itself, after being updated
"""
logger.info(f"Updating information on meeting {self.meeting_id} on server {self.api.name}")
res = self.api._request("getMeetingInfo", self._get_url_args("meeting_id"))
self._update_from_response(res)
return self
[docs] def end(self) -> None:
"""Ask the server to forcefully end the meeting.
This only sends the end API call to the server. According to the
BigBlueButton documentation, the call imemdiately returns, but
it can take several seconds for all componentes to pick up the
request and actually end the meeting. Accordingly, it is is up to
the developer to track the progress by calling :meth:`is_meeting_running`
or :meth:`get_meeting_info` again later on.
"""
logger.info(f"Ending meeting {self.meeting_id} on server {self.api.name}")
res = self.api._request("end", self._get_url_args("meeting_id", moderator_pw="password"))
self._update_from_response(res)
[docs] def to_dict(self, *args: str, **kwargs: str) -> Dict[str, Any]:
"""Return relevant data of this meeting as a dictionary.
The dictionary can be used to build an XML document compatible
with BigBlueButton API clients.
If names of attributes are passed as positional arguments, only
these attributes are returned in the dictionary.
If attribute names are passed as names of keyword arguments,
they are renamed to the string passed as value in the dictionary.
"""
res: Dict[str, Any] = {}
for name, value in self.__dict__.items():
if args and name not in args and name not in kwargs:
continue
if name == "api":
continue
elif name == "meta":
res["meta"] = self.meta
elif name == "attendees":
res["attendees"] = {
"attendee": [
attendee.to_dict() for attendee in self.attendees.values()
]
}
elif value is not None:
if name in kwargs:
camel_name = kwargs[name]
else:
camel_name = snake_to_camel(name)
if isinstance(value, bool):
str_value = "true" if value else "false"
else:
str_value = str(value)
res[camel_name] = str_value
return res
def _get_url_args(self, *args: str, **kwargs: str) -> Dict[str, str]:
url_args = self.to_dict(*args, **kwargs)
if "meta" in url_args:
# Unpack meta dictionary as values are passed one by one in URL
for name, value in url_args["meta"].items():
url_args["meta_" + name] = value
del url_args["meta"]
return url_args
[docs] @classmethod
def get_kwargs_from_url_args(cls, urlargs: Dict[str, str]) -> Dict[str, Any]:
"""Construct a dictionary suitable for passing as kwargs to the constructor.
The passed urlargs are expected to be a dictionary of URL arguments following
the BigBlueButton HTTP API schema.
This is useful to generate a meeting object from a URL call from a foreign
BBB client, an API reply, or comparable things.
"""
kwargs: Dict[str, Any] = {}
for name, value in urlargs.items():
if name.startswith("meta_"):
kwargs.setdefault("meta", {})
kwargs["meta"][name[5:]] = value
else:
snake_name = camel_to_snake(name)
kwargs[snake_name] = to_field_type(cls, snake_name, value)
return kwargs
def _update_from_response(self, res: Dict[str, Any]) -> None:
for name, value in res.items():
if name == "attendees":
if not value or not value["attendee"]:
self.attendees.clear()
else:
if not isinstance(value["attendee"], list):
value["attendee"] = [value["attendee"]]
for attendee_dict in value["attendee"]:
full_name = attendee_dict["fullName"]
if full_name in self.attendees:
attendee = self.attendees[full_name]
else:
attendee = Attendee(self, full_name)
self.attendees[full_name] = attendee
attendee._update_from_response(attendee_dict)
elif name == "metadata":
if value:
self.meta = dict(value)
else:
self.meta.clear()
else:
snake_name = camel_to_snake(name)
if hasattr(self, snake_name):
setattr(self, snake_name, to_field_type(self, snake_name, value))