Source code for bigbluebutton.api.meeting

"""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.logout_url = self.logout_url or 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.meta.setdefault("bbb-origin-server-name","Creating meeting {self.meeting_id} on server {}") 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 """"Updating information on meeting {self.meeting_id} on server {}") 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. """"Ending meeting {self.meeting_id} on server {}") 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))