diff --git a/README.md b/README.md index ee91b38..b46fbb8 100644 --- a/README.md +++ b/README.md @@ -42,4 +42,7 @@ A few default behaviours to note about the `Admin` module. # TODO -- A better way to handle help pages \ No newline at end of file +- Test suites +- Module to handle service authentication +- A better way to handle help pages +- `Sphinx` documentation generation diff --git a/boots/admin.py b/boots/admin.py index 9325a65..8dc2fb2 100644 --- a/boots/admin.py +++ b/boots/admin.py @@ -21,7 +21,7 @@ class Admin: LIST = re.compile(r'^list.*$') def __init__(self, client: robot.Bot, config: str = None): - self.logger = logging.getLogger(__class__.__name__) + self.logger = logging.getLogger(self.__class__.__name__) self.logger.debug("Initializing...") self.client = client self.config = config or "./config/admin.pkl" @@ -115,34 +115,43 @@ class Admin: self.logger.info("Loading configuration...") self.logger.debug("Attempting to load configuration {}".format(config)) try: - async with aiofiles.open(self.abspath(config or self.config), mode='rb') as f: + async with aiofiles.open( + self.abspath(config or self.config), mode='rb') as f: _file = await f.read() try: self.admins = pickle.loads(_file) except pickle.PickleError: - self.logger.warn("We failed to load configuration, creating a new one") + self.logger.warn( + "We failed to load configuration, creating a new one") self.admins = {} await self.add_admin("Admin", "Pa$$w0rd", 1000) f.close() except FileNotFoundError: - self.logger.warn("We failed to read the configuration file, creating a new configuration") + self.logger.warn( + "We failed to read the configuration file, " + "creating a new configuration") self.admins = {} await self.add_admin("Admin", "Pa$$w0rd", 1000) async def save_config(self, config: str = None) -> None: """ - This method will save the configuration of the admins into a pickled database + This method will save the configuration of the admins + into a pickled database :param config str: the path to the configuration :return: None """ self.logger.debug("We are attempting to write to drive...") - async with aiofiles.open(self.abspath(config or self.config), mode='wb+') as f: + async with aiofiles.open( + self.abspath(config or self.config), mode='wb+') as f: _file = pickle.dumps(self.admins) await f.write(_file) f.close() - async def add_admin(self, user: str, password: str, level: int = 1) -> None: + async def add_admin(self, + user: str, + password: str, + level: int = 1) -> None: """ Method to add an admin to the list of admins @@ -160,7 +169,8 @@ class Admin: 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, re.escape(self.hash(password)))) + r'^login\s+({})\s+({})\s*$'.format( + user, re.escape(self.hash(password)))) self.logger.debug("We are calling save_config()") await self.save_config() @@ -179,23 +189,30 @@ class Admin: self.logger.debug("We are calling save_config()") await self.save_config() - async def _handle(self, nick: str, target: str, message: str, **kwargs: dict) -> None: + async def _handle(self, + nick: str, + target: str, + message: str, + **kwargs) -> None: """ - Admin handler, this will check if the user is asking actions from the bot - and triggers an ADMIN event that can be consumed later + Admin handler, this will check if the user + is asking actions from the bot and triggers + an ADMIN event that can be consumed later :param nick str: nickname of the caller :param target str: location the message was sent to :param message str: messages coming from the `nick` - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ - self.logger.debug("We have received nick={} target={} message={} kwargs={}".format( - nick, target, message, kwargs)) + self.logger.debug("We have received nick={} target={} " + "message={} kwargs={}".format( + nick, target, message, kwargs)) # Set the admin flag admin = self.is_admin(nick, kwargs['host']) - self.logger.debug("We are checking to see if {} is an admin".format(nick)) + self.logger.debug( + "We are checking to see if {} is an admin".format(nick)) if admin: kwargs['is_admin'] = True kwargs['level'] = self.admins[admin]['level'] @@ -224,12 +241,19 @@ class Admin: kwargs['nick'] = nick kwargs['target'] = target kwargs['message'] = message - self.logger.debug("We are triggering a new event called ADMIN with {}".format(kwargs)) + self.logger.debug( + "We are triggering a new event called ADMIN with {}".format( + kwargs)) self.client.trigger('ADMIN', **kwargs) - def admin_help(self, nick: str, target: str, message: str, **kwargs: dict) -> None: + def admin_help(self, + nick: str, + target: str, + message: str, + **kwargs) -> None: """ - This method will reply back to the user a help manual of the available commands + This method will reply back to the user a help manual of the + available commands > help >> help [command] >> commands: login logout add rm list @@ -244,11 +268,12 @@ class Admin: :param nick str: the nickname of the caller :param target str: the target where the message was sent to :param message str: the message sent to the target - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if target == self.client.nick: - self.logger.debug("We have received a help request from {}".format(nick)) + self.logger.debug( + "We have received a help request from {}".format(nick)) if kwargs.get("is_admin", None) is True: match_help = self.HELP.match(message) match_help_cmd = self.HELP_CMD.match(message) @@ -256,21 +281,29 @@ class Admin: if match_help_cmd: if len(match_help_cmd.groups()) == 1: if match_help_cmd.group(1) == 'login': - kwargs['message'] = "login - Login as an admin with your account" + kwargs['message'] = \ + "login - Login as " \ + "an admin with your account" self.client.send("PRIVMSG", **kwargs) if match_help_cmd.group(1) == 'logout': - kwargs['message'] = "logout - Log out from your account" + kwargs['message'] = \ + "logout - Log out from your account" self.client.send("PRIVMSG", **kwargs) if match_help_cmd.group(1) == 'passwd': - kwargs['message'] = "passwd - Change your account\'s password" + kwargs['message'] = \ + "passwd - Change your" \ + " account\'s password" self.client.send("PRIVMSG", **kwargs) if match_help_cmd.group(1) == 'add': kwargs['message'] = \ - "add - adds an admin account to the list of" \ - "admins with provided level" + "add - adds" \ + " an admin account to the list of admins" \ + " with provided level" self.client.send("PRIVMSG", **kwargs) if match_help_cmd.group(1) == 'rm': - kwargs['message'] = "rm - removes an admin from the list of admins" + kwargs['message'] = \ + "rm - removes an admin from the list" \ + " of admins" self.client.send("PRIVMSG", **kwargs) if match_help_cmd.group(1) == 'list': kwargs['message'] = "list - lists all the admins" @@ -278,10 +311,15 @@ class Admin: elif match_help: kwargs['message'] = "help [command]" self.client.send("PRIVMSG", **kwargs) - kwargs['message'] = "commands: login logout passwd add rm list" + kwargs['message'] = \ + "commands: login logout passwd add rm list" self.client.send("PRIVMSG", **kwargs) - async def log_in(self, nick: str, target: str, message:str , **kwargs: dict) -> None: + async def log_in(self, + nick: str, + target: str, + message: str, + **kwargs) -> None: """ This method is called when a user attempts to login to the bot. It will mark the user as logged on on successful authentication @@ -295,7 +333,7 @@ class Admin: :param nick str: the nickname of the caller :param target str: location the message was sent to :param message str: message coming from `nick` - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if target == self.client.nick: @@ -306,28 +344,36 @@ class Admin: login = user['LOGIN'] if login.match(self.hash_pass(message)): if kwargs.get('is_admin', None) is True and \ - match.group(1) != self.is_admin(nick, kwargs['host']): + match.group(1) != self.is_admin(nick, + kwargs['host']): self.logger.warn( - "We detected that {} is already logged in as different user, logging him out".format(nick)) + "We detected that {} is already logged in as" + " different user, logging him out".format(nick)) await self.log_out(nick, target, **kwargs) if match.group(1) == self.is_admin(nick, kwargs['host']): kwargs['target'] = nick - kwargs['message'] = "{} you are already logged in...".format(nick) - self.logger.warn("We detected that {} is already logged in, notifying".format(nick)) + kwargs['message'] = "{} you are already logged in" \ + "...".format(nick) + self.logger.warn("We detected that {} is already " + "logged in, notifying".format(nick)) else: self.admins[match.group(1)]['logged_in'] = True - self.admins[match.group(1)]['logged_in_nick'] = str(nick) - self.admins[match.group(1)]['logged_in_hostname'] = str(kwargs['host']) + self.admins[match.group(1)]['logged_in_nick'] =\ + str(nick) + self.admins[match.group(1)]['logged_in_hostname'] = \ + str(kwargs['host']) kwargs['is_admin'] = True kwargs['target'] = nick - kwargs['message'] = "{}, you are logged in to {} successfully".format( + kwargs['message'] = "{}, you are logged in to {}" \ + " successfully".format( nick, match.group(1)) - self.logger.debug("We have logged in {} successfully, notifying".format(nick)) + self.logger.debug("We have logged in {} successfully, " + "notifying".format(nick)) self.client.send("PRIVMSG", **kwargs) self.logger.debug("We are calling save_config()") await self.save_config() - async def log_out(self, nick: str, target: str, **kwargs: dict) -> None: + async def log_out(self, nick: str, target: str, **kwargs) -> None: """ This method is called when a user attempts to logout to the bot. It will mark the user as logged out if they are the user logged @@ -339,7 +385,7 @@ class Admin: :param nick str: the nickname of the caller :param target str: location where the message was sent to - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if target == self.client.nick: @@ -352,24 +398,29 @@ class Admin: self.admins[admin]['logged_in_hostname'] = None kwargs['is_admin'] = False kwargs['target'] = nick - kwargs['message'] = "{}, you are logged out of {} successfully".format( - nick, admin) - self.logger.debug("We have successfully logged {} out, notifying".format(nick)) + kwargs['message'] = "{}, you are logged out of" \ + " successfully".format(nick, admin) + self.logger.debug("We have successfully logged {}" + " out, notifying".format(nick)) self.client.send("PRIVMSG", **kwargs) self.logger.debug("We are calling save_config") await self.save_config() - async def passwd(self, nick: str, target: str, message: str, **kwargs: dict) -> None: + async def passwd(self, + nick: str, + target: str, + message: str, + **kwargs) -> None: """ - This method will change the password to the administrator currently logged in - to the account. + This method will change the password to the administrator currently + logged in to the account. Saves the configuration to database :param nick str: the nickname of the caller :param target str: the target where the message was sent to :param message str: the message sent to target - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if target == self.client.nick: @@ -380,11 +431,15 @@ class Admin: admin = self.is_admin(nick, kwargs['host']) self.admins[admin]['password'] = self.hash(match.group(1)) self.admins[admin]['LOGIN'] = re.compile( - r'^login\s+({})\s+({})\s*$'.format(admin, re.escape(self.hash(match.group(1))))) + r'^login\s+({})\s+({})\s*$'.format( + admin, re.escape(self.hash(match.group(1))))) kwargs['target'] = nick - kwargs['message'] = '{}, password for {} has been successfully changed...'.format( - nick, admin) - self.logger.debug("We have successfully changed {}'s password, notifying".format(nick)) + kwargs['message'] = \ + '{}, password for {} has been successfully' \ + ' changed...'.format(nick, admin) + self.logger.debug( + "We have successfully changed {}'s password," + " notifying".format(nick)) self.client.send("PRIVMSG", **kwargs) self.logger.debug("We are calling save_config()") await self.save_config() @@ -392,17 +447,21 @@ class Admin: self.logger.debug("We are logging {} out".format(nick)) await self.log_out(nick, **kwargs) - async def admin_add(self, nick: str, target: str, message: str, **kwargs: dict) -> None: + async def admin_add(self, + nick: str, + target: str, + message: str, + **kwargs) -> None: """ - This method will add an administrator to the list of administrators only - if the administrator user exists + This method will add an administrator to the list of administrators + only if the administrator user exists Saves configuration to database :param nick str: the nickname of the caller :param target str: location where the message was sent to :param message str: the message being sent to the target - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if target == self.client.nick: @@ -412,24 +471,33 @@ class Admin: if len(match.groups()) == 3: if self.admins.get(match.group(1), None) is None: kwargs['target'] = nick - level = self.level_up(kwargs['level'], int(match.group(3))) - kwargs['message'] = "{} has been added with level {}...".format( - match.group(1), level) + level = self.level_up(kwargs['level'], + int(match.group(3))) + kwargs['message'] = "{} has been added with " \ + "level {}...".format( + match.group(1), level) self.logger.debug( - "We have added {} with level {}, notifying {}".format( - match.group(1), level, nick)) - await self.add_admin(match.group(1), match.group(2), level) + "We have added {} with level {}, notifying" + " {}".format(match.group(1), level, nick)) + await self.add_admin( + match.group(1), match.group(2), level) else: kwargs['target'] = nick - kwargs['message'] = "{} has already been added...".format(match.group(1)) + kwargs['message'] = "{} has already been added" \ + "...".format(match.group(1)) self.logger.warn( - "We detected that {} has already been added, notifying {}".format(match.group(1), nick)) + "We detected that {} has already been added," + " notifying {}".format(match.group(1), nick)) self.client.send("PRIVMSG", **kwargs) - async def admin_rm(self, nick: str, target: str, message: str, **kwags: dict) -> None: + async def admin_rm(self, + nick: str, + target: str, + message: str, + **kwags) -> None: """ - This method will remove an administrator from the list of administrators only - if the administrator user exists + This method will remove an administrator from the list of + administrators only if the administrator user exists The caller will be notified either way @@ -438,7 +506,7 @@ class Admin: :param nick str: the nickname of the caller :param target str: location where the message was sent to :param message str: the message being sent to the target - :param kwags dict: for API compatibility + :param kwags: for API compatibility :return: None """ if target == self.client.nick: @@ -448,33 +516,41 @@ class Admin: if len(match.groups()) == 1: if self.admins.get(match.group(1), None) is None: kwags['target'] = nick - kwags['message'] = "{} is not on the list...".format(match.group(1)) + kwags['message'] = "{} is not on the list" \ + "...".format(match.group(1)) self.logger.warn("admin_rm() ") else: - if kwags['level'] > self.admins[match.group(1)]['level'] or \ - (kwags['level'] == 1000 and self.admins[match.group(1)]['level'] and - kwags['level'] == self.admins[match.group(1)]['level']): + if kwags['level'] > \ + self.admins[match.group(1)]['level'] \ + or (kwags['level'] == 1000 + and self.admins[match.group(1)]['level'] + and kwags['level'] == + self.admins[match.group(1)]['level']): kwags['target'] = nick - kwags['message'] = "{} has been removed...".format(match.group(1)) - self.logger.debug("We removed {} successfully, notifying {}".format( - match.group(1), nick)) + kwags['message'] = "{} has been removed" \ + "...".format(match.group(1)) + self.logger.debug("We removed {} successfully," + " notifying {}".format( + match.group(1), nick)) await self.rm_admin(match.group(1)) else: kwags['target'] = nick - kwags['message'] = "{}, you do not have enough access to delete {}".format( + kwags['message'] = "{}, you do not have enough" \ + " access to delete {}".format( nick, match.group(1)) self.logger.warn( - "We detected that {0} does not have enough access to delete {1}, notifying {0}".format( - nick, match.group(1))) + "We detected that {0} does not have enough" + " access to delete {1}, notifying {0}".format( + nick, match.group(1))) self.client.send("PRIVMSG", **kwags) - def admin_list(self, nick: str, target: str, **kwargs: dict) -> None: + def admin_list(self, nick: str, target: str, **kwargs) -> None: """ This method returns to the admin the admin list :param nick str: the nickname of the caller :param target str: location where the message was sent to - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if target == self.client.nick: @@ -483,15 +559,17 @@ class Admin: admins = "" for key, _ in self.admins.items(): if admins: - admins = "{} {}({})".format(admins, key, self.admins[key]['level']) + admins = "{} {}({})".format( + admins, key, self.admins[key]['level']) else: - admins = "{}({})".format(key, self.admins[key]['level']) + admins = "{}({})".format( + key, self.admins[key]['level']) kwargs['target'] = nick kwargs['message'] = "List of Administrators:" self.client.send("PRIVMSG", **kwargs) kwargs['message'] = admins - self.logger.debug("We are returning admin list page to {}".format( - kwargs)) + self.logger.debug("We are returning admin list page to" + " {}".format(kwargs)) self.client.send("PRIVMSG", **kwargs) def is_admin(self, user: str, host: str): @@ -508,7 +586,9 @@ class Admin: 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("We are returning {} as an admin".format(user)) + self.logger.debug("We are returning {} as an" + " admin".format(user)) return admin - self.logger.debug("We are returning nothing, {} is not an admin".format(user)) + self.logger.debug("We are returning nothing, {} is" + " not an admin".format(user)) return None diff --git a/boots/admin_commands.py b/boots/admin_commands.py index 4a04744..8422f91 100644 --- a/boots/admin_commands.py +++ b/boots/admin_commands.py @@ -5,13 +5,14 @@ import logging import asyncio import admin + class AdminCmd: """ This is AdminCmd class that consumes ADMIN events and parses them into known actions defined by users """ def __init__(self, admin: admin.Admin, modifier: str = "!"): - self.logger = logging.getLogger(__class__.__name__) + self.logger = logging.getLogger(self.__class__.__name__) self.logger.debug("Initializing...") self.admin = admin self.client = admin.client @@ -20,21 +21,24 @@ class AdminCmd: self.services = {} admin.client.on("ADMIN")(self._handle) - def _handle(self, target: str, message: str, **kwargs: dict) -> None: + def _handle(self, target: str, message: str, **kwargs) -> None: """ client callback on event trigger :param target str: the target the message was sent to :param message str: the message sent to the target - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ if bool(kwargs.get('is_admin', None)): - self.logger.debug("We are being called by {}".format(kwargs['nick'])) + self.logger.debug( + "We are being called by {}".format(kwargs['nick'])) for regex, (func, pattern) in self.services.items(): match = regex.match(message) if match: - self.logger.debug("We are calling the function that matched the regex {}".format(regex)) + self.logger.debug( + "We are calling the function that matched the regex" + " {}".format(regex)) split_msg = message.split(" ") message = " ".join(split_msg[1:]) self.client.loop.create_task( @@ -43,13 +47,13 @@ class AdminCmd: def on_command(self, command: str, func: types.FunctionType = None, - ** kwargs: dict) -> types.FunctionType: + ** kwargs) -> types.FunctionType: """ Decorator function for the administrator commands :param commant str: the command that the admins are allowed to do :param func types.FunctionType: the function being decorated - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: types.FunctionType the function that called it """ if func is None: @@ -59,7 +63,8 @@ class AdminCmd: if not asyncio.iscoroutinefunction(wrapped): wrapped = asyncio.coroutine(wrapped) - compiled = re.compile(r'^{}{}\s*(.*)$'.format(re.escape(self.modifier), command)) + compiled = re.compile(r'^{}{}\s*(.*)$'.format( + re.escape(self.modifier), command)) self.services[compiled] = (wrapped, command) return func diff --git a/boots/boots.py b/boots/boots.py index b4f672b..3597fa8 100644 --- a/boots/boots.py +++ b/boots/boots.py @@ -11,7 +11,28 @@ logger = logging.getLogger(__name__) async def plugins(bot: robot.Bot): + + @bot.pre_public + async def auth(**kwargs): + logger.debug("Asking Help") + bot.send("USERMODE", nick=bot.nick, modes="+x") + target = "E@channels.evilnet.org" + message = "login " + bot.send("PRIVMSG", target=target, message=message) + await bot.wait("NOTICE") + +# Code below will not work, it is awaiting a PR upstream +# @bot.on("USERMODE") +# def umode(**kwargs): +# logger.debug("USERMODE {}".format(kwargs)) + +# @bot.on("CHANNELMODE") +# def cmode(**kwargs): +# logger.debug("CHANNELMODE {}".format(kwargs)) + + # 8Ball magic_ball = eightball.EightBall(bot) + @magic_ball.on_keyword async def ball(target, message, **kwargs): bot.send("PRIVMSG", target=target, message=message) @@ -38,7 +59,8 @@ async def plugins(bot: robot.Bot): @admin_cmd.on_command("action") def action(target, message, **kwargs): kwargs['target'] = message.split(' ')[0] - kwargs['message'] = "\x01ACTION {}\x01".format(" ".join(message.split(' ')[1:])) + kwargs['message'] = \ + "\x01ACTION {}\x01".format(" ".join(message.split(' ')[1:])) bot.send("PRIVMSG", **kwargs) @admin_cmd.on_command("quit") @@ -48,16 +70,19 @@ async def plugins(bot: robot.Bot): # Exit the event loop cleanly bot.loop.stop() + def main(): - host = 'eu.undernet.org' + host = 'irc.evilnet.org' port = 6667 ssl = False nick = "LuckyBoots" - channel = "#msaytbeh" + channel = "#boots" - bot = robot.Bot(host=host, port=port, ssl=ssl, nick=nick, channels=[channel]) + bot = robot.Bot(host=host, port=port, + ssl=ssl, nick=nick, + channels=[channel]) bot.loop.create_task(bot.connect()) @@ -67,4 +92,4 @@ def main(): if __name__ == '__main__': - main() \ No newline at end of file + main() diff --git a/boots/eightball.py b/boots/eightball.py index 53bc4d3..c73f362 100644 --- a/boots/eightball.py +++ b/boots/eightball.py @@ -6,6 +6,7 @@ import functools import asyncio import robot + class EightBall: """ EightBall Game for the IRC bot @@ -34,7 +35,7 @@ class EightBall: ] def __init__(self, client: robot.Bot, keyword: str = ".8ball"): - self.logger = logging.getLogger(__class__.__name__) + self.logger = logging.getLogger(self.__class__.__name__) self.logger.debug("Initializing...") self.client = client self.questions = {} @@ -48,7 +49,7 @@ class EightBall: :param nick str: nickname of the caller :param target str: nickname to reply to :param message str: the question asked - :param kwargs dict: + :param kwargs: for API compatibility :return: None """ self.logger.debug("We are being called by {}".format(nick)) @@ -61,14 +62,17 @@ class EightBall: for regex, (func, pattern) in self.questions.items(): match = regex.match(message) if match: - if is_question(match.group(1)): + if EightBall.is_question(match.group(1)): answer = self._get_answer() message = "{}, {}".format(nick, answer) - self.logger.debug("We found '{}', notifying {}".format(answer, nick)) + self.logger.debug("We found '{}', notifying" + " {}".format(answer, nick)) else: message = \ - "{}, I did not detect a question for me to answer".format(nick) - self.logger.debug("We did not detect a question, notifying {}".format(nick)) + "{}, I did not detect a question for me" \ + " to answer".format(nick) + self.logger.debug("We did not detect a question," + " notifying {}".format(nick)) self.client.loop.create_task( func(target, message, **kwargs)) @@ -100,13 +104,13 @@ class EightBall: def on_keyword(self, func: types.FunctionType = None, - **kwargs: dict): + **kwargs): """ 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 + :param kwargs: for API compatibility :return: function the function that called it """ if func is None: @@ -120,21 +124,15 @@ class EightBall: self.questions[compiled] = (wrapped, self.keyword_pattern) return func + @staticmethod + def is_question(phrase: str) -> bool: + """ + Method to check if a string is a question -logger = logging.getLogger(__name__) - - -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 - """ - logger.debug("We are being called for '{}'".format(phrase)) - compiled = re.compile(r'^[^\?]+\?\s*$') - if compiled.match(phrase): - logger.debug("Returning True") - return True - logger.debug("Returning False") - return False + :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/logger.py b/boots/logger.py index af2aa4d..d8aaf19 100644 --- a/boots/logger.py +++ b/boots/logger.py @@ -6,8 +6,7 @@ import yaml def setup_logging( default_path: str = None, default_level: int = logging.INFO, - env_key: str = 'LOG_CFG' -): + env_key: str = 'LOG_CFG'): if default_path: path = default_path value = os.getenv(env_key, None) @@ -18,5 +17,6 @@ def setup_logging( config = yaml.safe_load(f.read()) logging.config.dictConfig(config) else: - _format = '%(asctime)s - %(levelname)s - %(filename)s:%(name)s.%(funcName)s:%(lineno)d - %(message)s' + _format = '%(asctime)s - %(levelname)s - %(filename)s:' \ + '%(name)s.%(funcName)s:%(lineno)d - %(message)s' logging.basicConfig(level=default_level, format=_format) diff --git a/boots/robot.py b/boots/robot.py index a0bd1f1..11eaf6a 100644 --- a/boots/robot.py +++ b/boots/robot.py @@ -1,5 +1,7 @@ import logging import asyncio +import types +import functools import bottom @@ -17,41 +19,46 @@ class Bot(bottom.Client): realname: str = None, channels: list = None) -> None: super().__init__(host=host, port=port, ssl=ssl) - self.logger = logging.getLogger(__class__.__name__) + self.logger = logging.getLogger(self.__class__.__name__) self.logger.debug("Initializing...") self.nick = nick self.user = user or self.nick self.realname = realname or self.nick self.channels = channels or [] + self.pre_pub = [] self.on("ping", self.keepalive) self.on("CLIENT_CONNECT", self.on_connect) + self.on("PRIVMSG", self.privmsg) + self.on("NOTICE", self.notice) - async def keepalive(self, message: str = None, **kwargs: dict) -> None: + async def keepalive(self, message: str = None, **kwargs) -> 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 + :param kwargs: 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: + async def on_connect(self, **kwargs) -> None: """ Essential user information to send at the beginning of the connection with the server - :param kwargs dict: for API compatibility + :param kwargs: for API compatibility :return: None """ self.logger.debug("We are being called") self.logger.info("Connecting...") - self.logger.debug("We are sending NICK as {} to server ".format(self.nick)) + self.logger.debug( + "We are sending NICK as {} to server ".format(self.nick)) self.send('NICK', nick=self.nick) - self.logger.debug("We are sending USER {} {}".format(self.user, self.realname)) + self.logger.debug( + "We are sending USER {} {}".format(self.user, self.realname)) self.send('USER', user=self.user, realname=self.realname) @@ -65,11 +72,44 @@ class Bot(bottom.Client): # Cancel whichever waiter's event didn't come in for future in pending: future.cancel() - self.logger.info("We are auto-joining channels {}".format(self.channels)) + + for func in self.pre_pub: + self.logger.debug("Running {}".format(func)) + await self.loop.create_task(func(**kwargs)) + + self.logger.info( + "We are auto-joining channels {}".format(self.channels)) for channel in self.channels: self.send('JOIN', channel=channel) - async def on_disconnect(self, **kwargs: dict) -> None: + def pre_public(self, + func: types.FunctionType = None, + **kwargs) -> types.FunctionType: + """ + This method will inject a function to be trigger before + the Bot module joins channels + + :param func types.FunctionType: the method being decorated + :param kwargs: for API compatibility + :return func types.FunctionType: always return the function itself + """ + if func is None: + return functools.partial(self.pre_public) + + wrapped = func + if not asyncio.iscoroutinefunction(wrapped): + wrapped = asyncio.coroutine(wrapped) + + self.pre_pub.extend([wrapped]) + return func + + def privmsg(self, **kwargs) -> None: + self.logger.debug("PRIVMSG {}".format(kwargs)) + + def notice(self, **kwargs) -> None: + self.logger.debug("NOTICE {}".format(kwargs)) + + async def on_disconnect(self, **kwargs) -> None: self.logger.debug("We are being called") await self.disconnect() self.logger.debug("We are calling loop.stop()")