Second commit:

* Adding an 8ball game
* Adding an admin manager
* Creating the bot shell

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2017 Elijah Lazkani
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,18 @@
+# boots
+
+## Description
+
+This project is an attempt to write a fully featured IRC bot. I am learning to write code in python and get better at it.
+If you are reading this then you are welcome to contribute code, reviews are a part of learning programming =)
+
+# Requirements
+This code only requires one dependency; [bottom](https://github.com/numberoverzero/bottom).
+
+## 8ball
+
+The `Eightball` class gives the bot the ability to support the iconic 8Ball game.
+It will only answer to the `keyword` provided to the `Eightball` object.
+
+# Admin
+
+The `Admin` module provides the ability to add bot administrators who can request some destructive actions that might put the bot itself being abused like `join`, `part` and `quit`.

boots/admin.py

@@ -0,0 +1,122 @@
+import types
+import logging
+import re
+import asyncio
+import functools
+import robot
+
+
+class Admin:
+    """
+    Admin class that will keep user access to the bot
+    """
+    LOGIN = re.compile(r'^LOGIN\s+([^\s]+)\s+([^\s]+)\s*$')
+
+    def __init__(self, client: robot.Bot):
+        self.client = client
+        self.admins = {}
+        self.services = {}
+        self.logger = logging.getLogger()
+        self.logger.debug("Initializing...")
+        client.on("privmsg")(self._handle)
+
+    def add_admin(self, user: str, password: str) -> None:
+        """
+        Method to add an admin to the list of admins
+
+        :param user str: the user of the admin to be added
+        :param password str: the password to be added
+        :return: None
+        """
+        self.admins[user] = {}
+        self.admins[user]['password'] = password
+        self.admins[user]['logged_in'] = False
+        self.admins[user]['logged_in_hostname'] = None
+        self.admins[user]['logged_in_nick'] = None
+        self.admins[user]['LOGIN'] = re.compile(r'^LOGIN\s+({})\s+({})\s*$'.format(user, password))
+
+    def _handle(self, nick: str, target: str, message: str, **kwargs: dict) -> None:
+        """
+        client callback on event trigger
+
+        :param nick str: nickname of the caller
+        :param target str: nickname to reply to
+        :param message str: event hooks
+        :param kwargs dict: for API compatibility
+        :return: None
+        """
+        self.logger.debug("nick={} target={} message={}".format(nick, target, message))
+        self.logger.debug("{}".format(kwargs))
+        _user = kwargs['user']
+        _host = kwargs['host']
+        match = self.LOGIN.match(message)
+        self.logger.debug("match={}".format(match))
+        if match and target == self.client.nick:
+            user = self.admins.get(match.group(1), None)
+            if user:
+                _login = user.get('LOGIN', None)
+                if _login:
+                    login_match = _login.match(message)
+                    if login_match:
+                        self.admins[match.group(1)]['logged_in'] = True
+                        self.admins[match.group(1)]['logged_in_nick'] = str(_user)
+                        self.admins[match.group(1)]['logged_in_hostname'] = str(_host)
+                        kwargs['target'] = nick
+                        kwargs['message'] = "{}, you are logged in successfully".format(nick)
+                        self.logger.debug("kwargs={}".format(kwargs))
+                        self.client.send("PRIVMSG", **kwargs)
+                        return
+
+        for regex, (func, command) in self.services.items():
+            match = regex.match(message)
+            if match:
+                if self.is_admin(_user, _host):
+                    kwargs['channel'] = match.group(1)
+                    self.logger.debug("command={} kwargs={}".format(command, kwargs))
+                    self.client.loop.create_task(
+                        func(command, **kwargs))
+
+    def on_prvmsg(self,
+                  special_char: chr,
+                  command: str,
+                  func: types.FunctionType = None,
+                  **kwargs):
+        """
+        Decorator function for the admin plugin
+
+        :param command str: the command to answer to
+        :param func function: the function being decorated
+        :param kwargs dict: for API compatibility
+        :return: function the function that called it
+        """
+        self.logger.debug("sepcial_char={} command={} kwargs={}".format(special_char, command, kwargs))
+        if func is None:
+            return functools.partial(self.on_prvmsg, special_char, command)
+
+        wrapped = func
+        if not asyncio.iscoroutinefunction(wrapped):
+            wrapped = asyncio.coroutine(wrapped)
+
+        commands = command.split("|")
+
+        for command in commands:
+            compiled = re.compile(r'^{}{}\s*(.*)$'.format(re.escape(special_char), command))
+            self.services[compiled] = (wrapped, command)
+
+        return func
+
+    def is_admin(self, user: str, host: str):
+        """
+        Method to check if the current user is an admin
+
+        :param user str: the user to check against
+        :param host str: the host to check against
+        :return bool: if the user is an admin
+        """
+        for nick, value in self.admins.items():
+            if value.get('logged_in', None) and \
+               value.get('logged_in_nick', None) == user and \
+               value.get('logged_in_hostname', None) == host:
+                self.logger.debug("{} is an admin".format(nick))
+                return True
+        return False

boots/boots.py

@@ -0,0 +1,43 @@
+import logging
+import robot
+import eightball
+import admin
+
+logging.basicConfig()
+logger = logging.getLogger()
+logger.setLevel(logging.DEBUG)
+
+
+def main():
+    host = 'eu.undernet.org'
+    port = 6667
+    ssl = False
+
+    nick = "LuckyBoots"
+    channel = "#msaytbeh"
+
+    bot = robot.Bot(host=host, port=port, ssl=ssl, nick=nick, channels=[channel])
+    eightball = eightball.EightBall(bot)
+
+    @eightball.on_keyword
+    async def ball(target, message, **kwargs):
+        bot.send("PRIVMSG", target=target, message=message)
+
+    admin = admin.Admin(bot)
+    admin.add_admin("Armageddon", "12345")
+
+    @_admin.on_prvmsg(".", "join|part|quit")
+    async def login(command, **kwargs):
+        logger.debug("command={} kwargs={}".format(command, kwargs))
+        if command:
+            bot.send(command, **kwargs)
+
+    logger.info("Connecting to {}:{} as {}".format(host, port, nick))
+    bot.loop.create_task(bot.connect())
+
+    bot.loop.run_forever()
+    bot.loop.close()
+
+
+if __name__ == '__main__':
+    main()

boots/eightball.py

@@ -0,0 +1,127 @@
+import types
+import logging
+import random
+import re
+import functools
+import asyncio
+import robot
+
+class EightBall:
+    """
+    EightBall Game for the IRC bot
+    """
+    ANSWERS = [
+        "It is certain",
+        "It is decidedly so",
+        "Without a doubt",
+        "Yes definitely",
+        "You may rely on it",
+        "As I see it, yes",
+        "Most likely",
+        "Outlook good",
+        "Yes",
+        "Signs point to yes",
+        "Reply hazy try again",
+        "Ask again later",
+        "Better not tell you now",
+        "Cannot predict now",
+        "Concentrate and ask again",
+        "Don't count on it",
+        "My reply is no",
+        "My sources say no",
+        "Outlook not so good",
+        "Very doubtful"
+    ]
+
+    def __init__(self, client: robot.Bot, keyword: str = ".8ball"):
+        self.client = client
+        self.questions = {}
+        self.keyword_pattern = r'^{}\s+(.+)$'.format(keyword)
+        self.logger = logging.getLogger()
+        self.logger.debug("Initializing...")
+        client.on("privmsg")(self._handle)
+
+    def _handle(self, nick, target, message, **kwargs) -> None:
+        """
+        client callback on event trigger
+
+        :param nick str: nickname of the caller
+        :param target str: nickname to reply to
+        :param message str: the question asked
+        :param kwargs dict:
+        :return: None
+        """
+
+        nick, target = self._prefix(nick, target)
+
+        # Check that it did not call itself
+        if not nick and not target:
+            return
+
+        for regex, (func, pattern) in self.questions.items():
+            match = regex.match(message)
+            if match:
+                if is_question(match.group(1)):
+                    message = "{}, {}".format(nick, self._get_answer())
+                else:
+                    message = \
+                        "{}, I did not detect a question for me to answer".format(nick)
+                self.logger.debug("PRIVMSG {}".format(message))
+                self.client.loop.create_task(
+                    func(target, message, **kwargs))
+
+    def _prefix(self, nick: str, target: str) -> tuple:
+        """
+        Private method to check the prefix
+
+        This will make sure that it did not call itself
+        and that the nick is set to prefix the 8ball
+        message properly addressing the user calling it
+
+        :param nick str: the nickname calling the game
+        :param target str: the target we aim to reply to
+        :return tuple: the nick and target in a tuple
+        """
+        if nick == self.client.nick:
+            return None, None
+        if target == self.client.nick:
+            target = nick
+        return nick, target
+
+    def _get_answer(self):
+        return random.choice(self.ANSWERS)
+
+    def on_keyword(self,
+                   func: types.FunctionType = None,
+                   **kwargs: dict):
+        """
+        Decorator function for the 8ball game
+
+        :param keyword str: the keyword to activate the 8ball game
+        :param func types.FunctionType: the function being decorated
+        :param kwargs dict: for API compatibility
+        :return: function the function that called it
+        """
+        if func is None:
+            return functools.partial(self.on_keyword, self.keyword_pattern)
+
+        wrapped = func
+        if not asyncio.iscoroutinefunction(wrapped):
+            wrapped = asyncio.coroutine(wrapped)
+
+        compiled = re.compile(self.keyword_pattern)
+        self.questions[compiled] = (wrapped, self.keyword_pattern)
+        return func
+
+
+def is_question(phrase: str) -> bool:
+    """
+    Method to check if a string is a question
+
+    :param phrase str: the phrase to check
+    :return: bool if the phrase is a question or not
+    """
+    compiled = re.compile(r'^[^\?]+\?\s*$')
+    if compiled.match(phrase):
+        return True
+    return False

boots/robot.py

@@ -0,0 +1,63 @@
+import logging
+import asyncio
+import bottom
+
+
+class Bot(bottom.Client):
+    """
+    The Bot class adds features to the bottom.Client class
+    """
+    def __init__(self,
+                 host: str,
+                 port: int,
+                 ssl: bool,
+                 nick: str,
+                 user: str = None,
+                 realname: str = None,
+                 channels: list = None) -> None:
+        super().__init__(host=host, port=port, ssl=ssl)
+        self.logger = logging.getLogger()
+        self.nick = nick
+        self.user = user or self.nick
+        self.realname = realname or self.nick
+        self.channels = channels or []
+        self.on("ping", self.keepalive)
+        self.on("CLIENT_CONNECT", self.on_connect)
+
+    async def keepalive(self, message: str = None, **kwargs: dict) -> None:
+        """
+        Essential keepalive method to pong the server back
+        automagically everytime it pings
+
+        :param message str: the ping parameters
+        :param kwargs dict: for API compatibility
+        :return: None
+        """
+        message = message or ""
+        self.logger.debug("PONG {}".format(message))
+        self.send("pong", message=message)
+
+    async def on_connect(self, **kwargs: dict) -> None:
+        """
+        Essential user information to send at the
+        beginning of the connection with the server
+
+        :param kwargs dict: for API compatibility
+        :return: None
+        """
+        self.send('NICK', nick=self.nick)
+        self.send('USER', user=self.user,
+                         realname=self.realname)
+
+        done, pending = await asyncio.wait(
+            [self.wait("RPL_ENDOFMOTD"),
+             self.wait("ERR_NOMOTD")],
+            loop=self.loop,
+            return_when=asyncio.FIRST_COMPLETED
+        )
+
+        # Cancel whichever waiter's event didn't come in
+        for future in pending:
+            future.cancel()
+        for channel in self.channels:
+            self.send('JOIN', channel=channel)

requirements.txt

@@ -0,0 +1 @@
+bottom