Admin tool to send chat messags to running BBB meetings

bbb-broadcast.py

#!/usr/bin/env python3
import sys, json, subprocess, time, argparse
import logging
from typing import List
import urllib.request
import urllib.parse
import hashlib

"""
Admin tool to send chat messages to running BBB meetings.
This can be used for example to broadcast a maintenance warning message before shutting down a server.

Tested with BBB 3.0.3
Copyright 2022 Marcel Hellkamp (https://gist.github.com/defnull/8231ec8633781f621c59e8945256f028)
Modified 2025 Marc Kohaupt (https://gist.github.com/debuglevel/6a28f71ad2fff56df7471fa236f68db5)
License: MIT
"""

_logger = logging.getLogger(__name__)

def setup_logging(loglevel: int) -> None:
    """Setup basic logging

    Args:
      loglevel (int): minimum log level for emitting messages
    """
    _logger.debug(f"Setting up logging with loglevel={loglevel}...")

    log_format = "%(asctime)s %(levelname)-5s %(name)s - %(message)s"
    logging.basicConfig(
        level=loglevel,
        stream=sys.stdout,
        format=log_format,
        datefmt="%Y-%m-%d %H:%M:%S",
        # Overwrite existing logger configurations;
        # this is needed as basicConfig() only works if _logger was not used before.
        force=True,
    )

    _logger.debug(f"Set up logging with loglevel={loglevel}.")


def main() -> None:
  setup_logging(logging.DEBUG)

  parser = argparse.ArgumentParser(description='Send system messages to public chats of existing meetings.')
  parser.add_argument('--endpoint', required=True, help='BBB Endpoint')
  parser.add_argument('--secret', required=True, help='BBB Secret')
  parser.add_argument('--username', default="SYSTEM", help='User name (default: SYSTEM)')
  parser.add_argument('--meeting', default="ALL", help='Meeting ID (default: all meetings)')
  parser.add_argument('text', help='Message to send')
  args = parser.parse_args()

  if args.meeting == 'ALL':
    meetings_ids = get_all_meetings_ids(args.endpoint, args.secret)
  else:
    meetings_ids = [args.meeting]

  for meeting_id in meetings_ids:
    send_message(meeting_id, args.text, args.username, args.endpoint, args.secret)


def get_checksum(query: str, secret: str) -> str:
  _logger.debug(f"Getting checksum for query={query}...")
  
  query_plus_secret = query+secret
  _logger.debug(f"Generating checksum for query_plus_secret={query_plus_secret}...")
  checksum = hashlib.sha1(query_plus_secret.encode('utf8')).hexdigest()

  _logger.debug(f"Got checksum={checksum} for query={query}.")
  return checksum


def get_url(endpoint: str, secret: str, command: str, query: str) -> str:
  _logger.debug(f"Getting url with checksum for endpoint={endpoint} command={command} query={query}...")
  
  checksum = get_checksum(f"{command}{query}", secret)
  url = f"{endpoint}{command}?{query}&checksum={checksum}"

  _logger.debug(f"Got checksummed url={url} for endpoint={endpoint} command={command} query={query}.")
  return url


def get_all_meetings_ids(endpoint: str, secret: str):
  _logger.debug("Getting all meetings...")
  import xml.etree.ElementTree as ET

  url = get_url(endpoint, secret, "getMeetings", "")

  _logger.debug(f"Requesting url={url}...")
  with urllib.request.urlopen(url) as f:
    response = f.read().decode('utf8')
  _logger.debug(f"Requested url={url}.")
  # _logger.debug(f"Got response={response}.")

  _logger.debug(f"Parsing xml...")
  root = ET.fromstring(response)
  _logger.debug(f"Parsed xml.")

  _logger.debug(f"Extracting meetings...")
  meetings = root.findall('./meetings/meeting/internalMeetingID')
  _logger.debug(f"Extracted {len(meetings)} meetings.")
  for meeting in meetings:
    _logger.debug(f"Extracted meetingId={meeting.text}.")
    yield meeting.text


def send_message(meeting_id: str, text: str, username: str, endpoint: str, secret: str) -> None:
  _logger.debug(f"Sending message to meeting={meeting_id}...")

  encoded_message = urllib.parse.quote(text)
  encoded_username = urllib.parse.quote(username)

  query = f"meetingID={meeting_id}&message={encoded_message}&userName={encoded_username}"
  url = get_url(endpoint, secret, "sendChatMessage", query)

  _logger.debug(f"Requesting url={url}...")
  with urllib.request.urlopen(url) as f:
    response = f.read().decode('utf8')
  _logger.debug(f"Requested url={url}.")
  # _logger.debug(f"Got response={response}.")

  _logger.debug(f"Sent message to meeting={meeting_id}.")


if __name__ == '__main__':
  main()