Skip to content
Snippets Groups Projects
Select Git revision
  • 6bca4fc1745bdf46c00430060c0009132a549732
  • master default protected
  • v0.1.24
  • v0.1.23
  • v0.1.22
  • v0.1.21
  • v0.1.19
  • v0.1.18
  • v0.1.17
  • v0.1.16
  • v0.1.15
  • v0.1.14
  • v1.1
  • v.0.1.12
  • v.0.1.11
  • v.0.1.10
  • v.0.1.9
  • v.0.1.8
  • v.0.1.7
  • v.0.1.6
  • v0.1.5
21 results

meetings.py

Blame
  • meetings.py 10.60 KiB
    #!/usr/bin/env python3
    # -*- coding: utf-8 -*-
    
    import time
    import subprocess
    import hashlib
    from datetime import datetime, timedelta
    import requests
    from xml.etree import cElementTree as ElementTree
    from typing import NewType, Optional, Tuple, Iterable, List
    import click
    
    from bbbmon.xmldict import XmlListConfig, XmlDictConfig
    from bbbmon.configuration import Config, Endpoint, SERVER_PROPERTIES_FILE, Url, Secret, get_user_config_path, init_config, new_config
    import bbbmon.printing as printing
    
    
    
    
    
    
    def generate_checksum(call_name: str, query_string: str, secret: Secret) -> str:
        """
        Generate Checksum for the request header (passed as value for `?checksum=`)
        """
        m = hashlib.sha1()
        m.update(call_name.encode('utf-8'))
        m.update(query_string.encode('utf-8'))
        m.update(secret.encode('utf-8'))
        return m.hexdigest()
    
    
    def request_meetings(secret: Secret, bbb_url: Url, user_config_path: str) -> XmlDictConfig:
        """
        Make a getMeetings-API Call to the bbb instance and return a XmlDictConfig
        with the servers response
        """
        call_name = "getMeetings"
        checksum = generate_checksum(call_name, "", secret)
        url = "{}/api/{}?checksum={}".format(bbb_url, call_name, checksum)
    
        try:
            r = requests.get(url)
        except:
            click.echo("{} The URL \"{}\" is unreachable.\n       Check your network connection, and the URL and Secret of the endpoint.".format(click.style('Error:', fg='red', bold=True), url))
            print()
            time.sleep(1)
            if click.confirm(click.style('Do you want to open the config file at {} with your default editor?'.format(user_config_path), fg="yellow"), abort=True):
                click.edit(filename=user_config_path)
                exit()
    
        root = ElementTree.XML(r.text)
        xmldict = XmlDictConfig(root)
        if "returncode" in xmldict.keys():
            if xmldict['returncode'] == "FAILED":
                print(xmldict)
                exit()
        else:
            print(r.text)
            exit()
        return xmldict
    
    
    def get_meetings(secret: Secret, bbb_url: Url, user_config_path: str) -> Iterable[XmlDictConfig]:
        """
        Request meetings and return a list of them. Sorted by biggest first
        """
        meetings = []
        d = request_meetings(secret, bbb_url, user_config_path)
    
        if d["meetings"] is None:
            return []
    
        if type(d["meetings"]["meeting"]) is XmlListConfig:
            meetings = sorted([m for m in d["meetings"]["meeting"] if m["running"] == "true"], key=lambda x:int(x['participantCount']), reverse=True)
        elif type(d["meetings"]["meeting"]) is XmlDictConfig:
            meetings = [d["meetings"]["meeting"]]
        return meetings
    
    
    def get_presenter(meeting: XmlDictConfig) -> Optional[XmlDictConfig]:
        """
        Get the presenter of a meeting (return None if there is none)
        """
        presenters = []
        if type(meeting["attendees"]["attendee"]) is XmlListConfig:
            presenters = [a for a in meeting["attendees"]["attendee"] if a["isPresenter"] == "true"]
        elif type(meeting["attendees"]["attendee"]) is XmlDictConfig:
            presenters = [meeting["attendees"]["attendee"]]
        
        if len(presenters) > 0:
            return presenters[0]
        else:
            return None
    
    
    def get_duration(meeting: XmlDictConfig) -> timedelta:
        """
        Return the duration of a meeting
        """
        timestamp = int(meeting["startTime"][:-3])
        start_time = datetime.fromtimestamp(timestamp)
        duration = datetime.now() - start_time
        return duration
    
    
    def list_meetings(config: Config, leaderboards: bool, n: int, participants: bool, presenter: bool, presenter_id: bool, show_meetings: bool, watch: int, fancy: bool, compact: bool):
        """
        For each endpoint in the configuration get the active meetings and print 
        out an overview of the current bbb-usage
        """
    
        # Request Meetings from API
        meetings = [get_meetings(e.secret, e.url, config.path) for e in config.endpoints]
    
        # Clear screen after request is done, and before printing new data to keep
        # blinking to a minimum
        if watch is not None:
            click.clear()
    
    
        for i, endpoint in enumerate(config.endpoints):
            meeting = meetings[i]
    
            # Print divider if there is more than one endpoint
            if i > 0:
                print()
                print("="*click.get_terminal_size()[0])
                print()
    
            # If there are no meetings, skip to next endpoint
            if len(meeting) == 0:
                if show_meetings:
                    printing.print_header(endpoint.name, "MEETINGS", fancy)
                    print("   └─── Currently no active meetings.")
                continue
    
            n_running = len(meeting)
            n_recording = len([m for m in meeting if m["recording"] == "true"])
            n_participants = sum([int(m["participantCount"]) for m in meeting])
            n_listeners = sum([int(m["listenerCount"]) for m in meeting])
            n_voice = sum([int(m["voiceParticipantCount"]) for m in meeting])
            n_video = sum([int(m["videoCount"]) for m in meeting])
            n_moderator = sum([int(m["moderatorCount"]) for m in meeting])
    
            if show_meetings and not compact:
                printing.print_header(endpoint.name, "MEETINGS", fancy)
                print("   ├─── {:>4} running".format(n_running))
                print("   └─── {:>4} recording".format(n_recording))
                print()
    
            if participants and not compact:
                printing.print_header(endpoint.name, "PARTICIPANTS across all {} rooms".format(n_running), fancy)
                print("   └─┬─ {:>4} total".format(n_participants))
                print("     ├─ {:>4} listening only".format(n_listeners))
                print("     ├─ {:>4} mic on".format(n_voice))
                print("     ├─ {:>4} video on".format(n_video))
                print("     └─ {:>4} moderators".format(n_moderator))
    
            if compact: 
                h1 = printing.format_header(endpoint.name, "MEETINGS", fancy) 
                h2 = printing.format_header("", "PARTICIPANTS across all {} rooms".format(n_running), fancy)
                s1 ="   ├─── {:>4} running  ".format(n_running)
                s2 ="   └─── {:>4} recording".format(n_recording)
                p1 ="   └─┬─ {:>4} total".format(n_participants)
                p2 ="     ├─ {:>4} listening only".format(n_listeners)
                p3 ="     ├─ {:>4} mic on".format(n_voice)
                p4 ="     ├─ {:>4} video on".format(n_video)
                p5 ="     └─ {:>4} moderators".format(n_moderator)
                print("{:<20}  {}".format(h1.rstrip(), h2))
                print("{:<22} {}".format(s1, p1))
                print("{:<22} {}".format(s2, p2))
                print("{:<22} {}".format("", p3))
                print("{:<22} {}".format("", p4))
                print("{:<22} {}".format("", p5))
    
            if leaderboards:
                print()
                printing.print_leaderboard(meeting, n, "participantCount", endpoint.name, presenter, presenter_id, fancy)
                printing.print_leaderboard(meeting, n, "videoCount", endpoint.name, presenter, presenter_id, fancy)
                printing.print_leaderboard(meeting, n, "voiceParticipantCount", endpoint.name, presenter, presenter_id, fancy)
                printing.print_duration_leaderboard(meeting, n, endpoint.name, presenter, presenter_id, fancy)
    
    
    
    def meetings_twolines(config: Config, watch: int, fancy: bool):
        """
        For each endpoint in the configuration get the active meetings and print 
        out an overview of the current bbb-usage. This is guaranteed to fit within
        60 characters and two lines
        """
    
        # Request Meetings from API
        meetings = [get_meetings(e.secret, e.url, config.path) for e in config.endpoints]
    
        # Clear screen after request is done, and before printing new data to keep
        # blinking to a minimum
        if watch is not None:
            click.clear()
    
        for i, endpoint in enumerate(config.endpoints):
            meeting = meetings[i]
    
            # Print divider if there is more than one endpoint
            if i > 0:
                print("="*click.get_terminal_size()[0])
    
            # If there are no meetings, skip to next endpoint
            if len(meeting) == 0:
                lines = [
                    "{:^60}".format("{}   there   are   currently  no   sessions.").format(endpoint.name[:3]),
                    ""
                ]
                # Cut above 60 characters fill empty
                lines = ["{:<60}".format(l[:61]) for l in lines]
                lines = "\n".join(lines)
                print(lines)
                continue
    
            n_running = len(meeting)
            n_recording = len([m for m in meeting if m["recording"] == "true"])
            n_participants = sum([int(m["participantCount"]) for m in meeting])
            n_listeners = sum([int(m["listenerCount"]) for m in meeting])
            n_voice = sum([int(m["voiceParticipantCount"]) for m in meeting])
            n_video = sum([int(m["videoCount"]) for m in meeting])
            n_moderator = sum([int(m["moderatorCount"]) for m in meeting])
            avg_s = sum([int(get_duration(m).total_seconds()) for m in meeting])/float(n_running)
            avg_s = avg_s/60./60.
    
            if not fancy:
                if config.on_server:
                    # If on server get load
                    w = subprocess.run(["w | head -1"], shell=True, stdout=subprocess.PIPE)
                    w = w.stdout.decode('utf-8').strip().split("load average:")
                    if len(w) != 2:
                        w = 999.0
                    else:
                        w = w[1].strip().split(" ")[0].rstrip(",")
                        try:
                            w = float(w.replace(",", "."))
                        except:
                            w =  666.0
                    lines = [
                        "{}    rec / ses    ppl    mod   vid   mic  ear  lod".format(endpoint.name[:3]),
                        "stats   {:>2} /  {:<2}    {:>3}    {:>3}   {:>3}   {:>3}  {:>3}  {:.1f}"\
                        .format(n_recording, n_running, n_participants, n_moderator, n_video, n_voice, n_listeners, w)
                    ]
                else:
                    lines = [
                        "{}    rec / ses    ppl    mod   vid   mic  ear".format(endpoint.name[:3]),
                        "stats   {:>2} /  {:<2}    {:>3}    {:>3}   {:>3}   {:>3}  {:>3}"\
                        .format(n_recording, n_running, n_participants, n_moderator, n_video, n_voice, n_listeners)
                    ]
            else:
                lines = [
                    "{}: There are {} people in {} meetings for {:.1f}h on average"\
                    .format(endpoint.name, n_participants, n_running, avg_s),
                    "{} webcams and {} microphones are on. {} just listen. {} mods"\
                    .format(n_video, n_voice, n_listeners, n_moderator)
                ]
    
            # Cut above 60 characters fill empty
            lines = ["{:^60}".format(l[:61]) for l in lines]
            lines = "\n".join(lines)
            print(lines)