ejabberd-tools/calls.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import logging
import re

from packaging import version

from api import EjabberdApi

log = logging.getLogger(__name__)


class EjabberdApiCalls(EjabberdApi):
    @property
    def nodename(self):
        if self._login is not None:
            status = self.cmd("status", {})
            # "The node ejabberd@localhost is started with status: startedejabberd 20.07 is running in that node"

            try:
                tmp = status.split()[2]

            except AttributeError:
                # emtpy response or None obj
                log.warning("nodename not found on split")
                raise SystemExit(17)

            except IndexError:
                # status string differs from what we expect
                log.warning("status string is different then expected")
                tmp = "ejabberd@status-string-split-error"
                pass

            # strip double quotations
            if tmp.startswith("'"):
                tmp = tmp.strip("'")

            log.debug(f"fetched node string: {tmp}")
            return tmp

        return None

    @property
    def verstring(self):
        if self._login is not None:
            ver_str = re.compile("([1-9][0-9.]+(?![.a-z]))\\b")
            status = self.cmd("status", {})

            # matches
            try:
                tmp = ver_str.findall(status)[0]
            # raise SystemExit code 17 if no status message is received
            except TypeError:
                log.warning("ver sting not parsed")
                raise SystemExit(17)

            # return parsed version string
            log.debug(f"fetched version string: {tmp}")
            return version.parse(tmp)

        return None

    def fetch_onlineuser(self):
        tmp = self.cmd("connected_users_info", {})
        if "connected_users_info" not in tmp:
            return tmp
        data = []
        for c in tmp["connected_users_info"]:
            if "session" not in c:
                continue
            user = {}
            for attrs in c["session"]:
                for k, v in attrs.items():
                    user[k] = v
            data.append(user)
        return data

    def fetch_nodes(self):
        result = self.cmd("list_cluster", {})
        if "nodes" not in result:
            return result
        data = []
        for node in result["nodes"]:
            data.append(node["node"])
        return data

    def fetch_vhosts(self):
        result = self.cmd("registered_vhosts", {})
        if "vhosts" not in result:
            return result
        data = []
        for vhost in result["vhosts"]:
            data.append(vhost["vhost"])
        return data

    def fetch_s2s_in(self):
        result = self.cmd("incoming_s2s_number", {})
        if isinstance(result, dict):
            if "s2s_incoming" in result:
                return result["s2s_incoming"]
            log.warning("fetch_s2s_in: error empty result " + str(result))
            return 0
        return result

    def fetch_s2s_out(self):
        result = self.cmd("outgoing_s2s_number", {})
        if isinstance(result, dict):
            if "s2s_outgoing" in result:
                return result["s2s_outgoing"]
            log.warning("fetch_s2s_out: error empty result " + str(result))
            return 0
        return result

    def fetch_uptime(self):
        result = self.cmd("stats", {"name": "uptimeseconds"})
        if isinstance(result, dict):
            if "stat" in result:
                return result["stat"]
            log.warning("uptime: error empty result " + str(result))
            return 0
        return result

    def fetch_processes(self):
        result = self.cmd("stats", {"name": "processes"})
        if isinstance(result, dict):
            if "stat" in result:
                return result["stat"]
            log.warning("processes: error empty result " + str(result))
            return 0
        return result

    def fetch_registered_count(self, vhost=None):
        if vhost is None:
            result = self.cmd("stats", {"name": "registeredusers"})
            if "stat" in result:
                return result["stat"]
        else:
            result = self.cmd("stats_host", {"name": "registeredusers", "host": vhost})
            if "stat" in result:
                return result["stat"]

    def fetch_muc_count(self, vhost=None, muc_host="conference"):
        host = "global"
        if vhost is not None:
            if self.verstring.major >= 19:
                host = ".".join([muc_host, vhost])
            else:
                host = vhost
        result = self.cmd("muc_online_rooms", {"host": host})
        if "rooms" in result:
            return len(result["rooms"])
        return len(result)