xbotlib/xbotlib.py

"""XMPP bots for humans."""

from configparser import ConfigParser
from getpass import getpass
from os.path import exists
from pathlib import Path

from slixmpp import ClientXMPP


class EasyMessage:
    """A simple message interface."""

    def __init__(self, message):
        self.message = message

    @property
    def body(self):
        return self.message["body"]

    @property
    def sender(self):
        return self.message["from"].bare

    @property
    def receiver(self):
        return self.message["to"]

    @property
    def nickname(self):
        return self.message["mucnick"]

    @property
    def type(self):
        return self.message["type"]


class Bot(ClientXMPP):
    CONFIG_FILE = "bot.conf"

    def __init__(self):
        self.read_config()
        self.init_bot()
        self.register_xmpp_event_handlers()
        self.register_xmpp_plugins()
        self.run()

    def read_config(self):
        """Read configuration for running bot."""
        config_file_path = Path(self.CONFIG_FILE).absolute()

        if not exists(config_file_path):
            self.generate_config()

        self.config = ConfigParser()
        self.config.read(config_file_path)

    def generate_config(self):
        """Generate bot configuration."""
        jid = input("XMPP address (e.g. foo@bar.com): ") or "foo@bar.com"
        password = (
            getpass("Password (e.g. my-cool-password): ") or "my-cool-password"
        )
        room = input("XMPP room (e.g. foo@muc.bar.com): ")
        nick = input("Nickname (e.g. lurkbot): ")

        config = ConfigParser()
        config["bot"] = {"jid": jid, "password": password}

        if room:
            config["bot"]["room"] = room
        if nick:
            config["bot"]["nick"] = nick

        with open("bot.conf", "w") as file_handle:
            config.write(file_handle)

    def init_bot(self):
        """Initialise bot with connection details."""
        jid = self.config["bot"]["jid"]
        password = self.config["bot"]["password"]
        ClientXMPP.__init__(self, jid, password)

    def register_xmpp_event_handlers(self):
        """Register functions against specific XMPP event handlers."""
        self.add_event_handler("session_start", self.session_start)
        self.add_event_handler("message", self.message)
        self.add_event_handler("groupchat_message", self.groupchat_message)

    def message(self, message):
        """Handle message event."""
        if message["type"] in ("chat", "normal"):
            self.react(EasyMessage(message))

    def session_start(self, event):
        """Handle session_start event."""
        self.send_presence()
        self.get_roster()

        room = self.config["bot"].get("room")
        nick = self.config["bot"].get("nick")

        if room and nick:
            self.plugin["xep_0045"].join_muc(room, nick)

    def groupchat_message(self, message):
        """Handle groupchat_message event."""
        if message["type"] in ("groupchat", "normal"):
            if message["mucnick"] != self.config["bot"]["nick"]:
                self.react(EasyMessage(message))

    def register_xmpp_plugins(self):
        """Register XMPP plugins that the bot supports."""
        self.register_plugin("xep_0030")  # Service Discovery
        self.register_plugin("xep_0045")  # Multi-User Chat
        self.register_plugin("xep_0199")  # XMPP Ping

    def run(self):
        """Run the bot."""
        self.connect()

        try:
            self.process()
        except KeyboardInterrupt:
            pass

    def reply(self, to=None, room=None, body=None):
        """Send back a reply."""
        if to is None and room is None:
            message = "`to` or `room` arguments required for `reply`"
            raise RuntimeError(message)

        if to is not None and room is not None:
            message = "Cannot send to both `to` and `room` for `reply`"
            raise RuntimeError(message)

        kwargs = {"mbody": body}
        if to is not None:
            kwargs["mto"] = to
            kwargs["mtype"] = "chat"
        else:
            kwargs["mto"] = room
            kwargs["mtype"] = "groupchat"

        self.send_message(**kwargs)