From 7f259bac357692e79afd8bb6467e95125ea0a3cd Mon Sep 17 00:00:00 2001 From: Elijah Lazkani Date: Wed, 6 Sep 2017 20:39:32 -0400 Subject: [PATCH] Second commit: * Adding an 8ball game * Adding an admin manager * Creating the bot shell --- LICENSE | 21 ++++++++ README.md | 18 +++++++ boots/__init__.py | 0 boots/admin.py | 122 +++++++++++++++++++++++++++++++++++++++++++ boots/boots.py | 43 +++++++++++++++ boots/eightball.py | 127 +++++++++++++++++++++++++++++++++++++++++++++ boots/robot.py | 63 ++++++++++++++++++++++ requirements.txt | 1 + 8 files changed, 395 insertions(+) create mode 100644 LICENSE create mode 100644 README.md create mode 100644 boots/__init__.py create mode 100644 boots/admin.py create mode 100644 boots/boots.py create mode 100644 boots/eightball.py create mode 100644 boots/robot.py create mode 100644 requirements.txt diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..27b6949 --- /dev/null +++ b/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. diff --git a/README.md b/README.md new file mode 100644 index 0000000..6454326 --- /dev/null +++ b/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`. \ No newline at end of file diff --git a/boots/__init__.py b/boots/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/boots/admin.py b/boots/admin.py new file mode 100644 index 0000000..888bc54 --- /dev/null +++ b/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 diff --git a/boots/boots.py b/boots/boots.py new file mode 100644 index 0000000..31c9a70 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/boots/eightball.py b/boots/eightball.py new file mode 100644 index 0000000..2713038 --- /dev/null +++ b/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 diff --git a/boots/robot.py b/boots/robot.py new file mode 100644 index 0000000..fe2e0fb --- /dev/null +++ b/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) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3becee5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +bottom \ No newline at end of file