Compare commits

..

No commits in common. "master" and "Features" have entirely different histories.

5 changed files with 151 additions and 464 deletions

View file

@ -4,7 +4,6 @@ import logging
import pickle import pickle
import hashlib import hashlib
import aiofiles import aiofiles
import robot import robot
@ -206,8 +205,8 @@ class Admin:
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
self.logger.debug( self.logger.debug("We have received nick={} target={} "
"We have received nick={}, target={}, message={}, kwargs={}".format( "message={} kwargs={}".format(
nick, target, message, kwargs)) nick, target, message, kwargs))
# Set the admin flag # Set the admin flag
@ -215,43 +214,42 @@ class Admin:
self.logger.debug( self.logger.debug(
"We are checking to see if {} is an admin".format(nick)) "We are checking to see if {} is an admin".format(nick))
if admin: if admin:
is_admin = True kwargs['is_admin'] = True
level = self.admins[admin]['level'] kwargs['level'] = self.admins[admin]['level']
else: else:
is_admin = False kwargs['is_admin'] = False
level = 0 if kwargs.get('level', None) is not None:
del kwargs['level']
# TODO: fix
self.logger.debug("We are parsing the message sent by {}".format(nick)) self.logger.debug("We are parsing the message sent by {}".format(nick))
if self.HELP.match(message): if self.HELP.match(message):
await self.admin_help(nick, target, message, is_admin, **kwargs) self.admin_help(nick, target, message, **kwargs)
elif self.LOGIN.match(message): elif self.LOGIN.match(message):
await self.log_in(nick, target, message, is_admin, **kwargs) await self.log_in(nick, target, message, **kwargs)
elif self.LOGOUT.match(message): elif self.LOGOUT.match(message):
await self.log_out(nick, target, is_admin, **kwargs) await self.log_out(nick, target, **kwargs)
elif self.PASSWD.match(message): elif self.PASSWD.match(message):
await self.passwd(nick, target, message, is_admin, **kwargs) await self.passwd(nick, target, message, **kwargs)
elif self.ADD.match(message): elif self.ADD.match(message):
await self.admin_add(nick, target, message, is_admin, level, **kwargs) await self.admin_add(nick, target, message, **kwargs)
elif self.RM.match(message): elif self.RM.match(message):
await self.admin_rm(nick, target, message, is_admin, level, **kwargs) await self.admin_rm(nick, target, message, **kwargs)
elif self.LIST.match(message): elif self.LIST.match(message):
await self.admin_list(nick, target, is_admin, **kwargs) self.admin_list(nick, target, **kwargs)
self.logger.debug("We are modifying kwargs")
kwargs['nick'] = nick
kwargs['target'] = target
kwargs['message'] = message
self.logger.debug( self.logger.debug(
"We are triggering a new event called ADMIN with " "We are triggering a new event called ADMIN with {}".format(
"nick={}, target={}, message={}, is_admin={}, level={}, "
"**kwargs={}".format(nick, target, message, is_admin, level,
kwargs)) kwargs))
self.client.trigger('ADMIN', nick=nick, target=target, self.client.trigger('ADMIN', **kwargs)
message=message, is_admin=is_admin,
level=level, **kwargs)
async def admin_help(self, def admin_help(self,
nick: str, nick: str,
target: str, target: str,
message: str, message: str,
is_admin: bool,
**kwargs) -> None: **kwargs) -> None:
""" """
This method will reply back to the user a help manual of the This method will reply back to the user a help manual of the
@ -270,59 +268,57 @@ class Admin:
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: the target where the message was sent to :param target str: the target where the message was sent to
:param message str: the message sent to the target :param message str: the message sent to the target
:param is_admin bool: the user is an admin
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
if target == self.client.nick: if target == self.client.nick:
self.logger.debug( self.logger.debug(
"We have received a help request from {}".format(nick)) "We have received a help request from {}".format(nick))
if is_admin: if kwargs.get("is_admin", None) is True:
match_help = self.HELP.match(message) match_help = self.HELP.match(message)
match_help_cmd = self.HELP_CMD.match(message) match_help_cmd = self.HELP_CMD.match(message)
_target = nick kwargs['target'] = nick
if match_help_cmd: if match_help_cmd:
if len(match_help_cmd.groups()) == 1: if len(match_help_cmd.groups()) == 1:
if match_help_cmd.group(1) == 'login': if match_help_cmd.group(1) == 'login':
_message = \ kwargs['message'] = \
"login <user> <password> <level> - Login as " \ "login <user> <password> <level> - Login as " \
"an admin with your account" "an admin with your account"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
if match_help_cmd.group(1) == 'logout': if match_help_cmd.group(1) == 'logout':
_message = \ kwargs['message'] = \
"logout <user> - Log out from your account" "logout <user> - Log out from your account"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
if match_help_cmd.group(1) == 'passwd': if match_help_cmd.group(1) == 'passwd':
_message = \ kwargs['message'] = \
"passwd <new password> - Change your" \ "passwd <new password> - Change your" \
" account\'s password" " account\'s password"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
if match_help_cmd.group(1) == 'add': if match_help_cmd.group(1) == 'add':
_message = \ kwargs['message'] = \
"add <user> <password> <level> - adds" \ "add <user> <password> <level> - adds" \
" an admin account to the list of admins" \ " an admin account to the list of admins" \
" with provided level" " with provided level"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
if match_help_cmd.group(1) == 'rm': if match_help_cmd.group(1) == 'rm':
_message = \ kwargs['message'] = \
"rm <user> - removes an admin from the list" \ "rm <user> - removes an admin from the list" \
" of admins" " of admins"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
if match_help_cmd.group(1) == 'list': if match_help_cmd.group(1) == 'list':
_message = "list - lists all the admins" kwargs['message'] = "list - lists all the admins"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
elif match_help: elif match_help:
_message = "help [command]" kwargs['message'] = "help [command]"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
_message = \ kwargs['message'] = \
"commands: login logout passwd add rm list" "commands: login logout passwd add rm list"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
async def log_in(self, async def log_in(self,
nick: str, nick: str,
target: str, target: str,
message: str, message: str,
is_admin: bool,
**kwargs) -> None: **kwargs) -> None:
""" """
This method is called when a user attempts to login to the bot. This method is called when a user attempts to login to the bot.
@ -337,7 +333,6 @@ class Admin:
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: location the message was sent to :param target str: location the message was sent to
:param message str: message coming from `nick` :param message str: message coming from `nick`
:param is_admin bool: the user is an admin
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
@ -348,16 +343,17 @@ class Admin:
if user: if user:
login = user['LOGIN'] login = user['LOGIN']
if login.match(self.hash_pass(message)): if login.match(self.hash_pass(message)):
if is_admin and \ 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( self.logger.warn(
"We detected that {} is already logged in as" "We detected that {} is already logged in as"
" different user, logging him out".format(nick)) " different user, logging him out".format(nick))
await self.log_out(nick, target, **kwargs) await self.log_out(nick, target, **kwargs)
if match.group(1) == self.is_admin(nick, kwargs['host']): if match.group(1) == self.is_admin(nick, kwargs['host']):
_target = nick kwargs['target'] = nick
_message = "{} you are already logged in" \ kwargs['message'] = "{} you are already logged in" \
"...".format(_target) "...".format(nick)
self.logger.warn("We detected that {} is already " self.logger.warn("We detected that {} is already "
"logged in, notifying".format(nick)) "logged in, notifying".format(nick))
else: else:
@ -366,23 +362,18 @@ class Admin:
str(nick) str(nick)
self.admins[match.group(1)]['logged_in_hostname'] = \ self.admins[match.group(1)]['logged_in_hostname'] = \
str(kwargs['host']) str(kwargs['host'])
is_admin = True kwargs['is_admin'] = True
_target = nick kwargs['target'] = nick
_message = "{}, you are logged in to {}" \ kwargs['message'] = "{}, you are logged in to {}" \
" successfully".format( " successfully".format(
_target, match.group(1)) nick, match.group(1))
self.logger.debug("We have logged in {} successfully, " self.logger.debug("We have logged in {} successfully, "
"notifying".format(_target)) "notifying".format(nick))
self.client.send("PRIVMSG", **kwargs)
await self.client.msg(target=_target, message=_message)
self.logger.debug("We are calling save_config()") self.logger.debug("We are calling save_config()")
await self.save_config() await self.save_config()
async def log_out(self, async def log_out(self, nick: str, target: str, **kwargs) -> None:
nick: str,
target: str,
is_admin: bool,
**kwargs) -> None:
""" """
This method is called when a user attempts to logout to the bot. 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 It will mark the user as logged out if they are the user logged
@ -394,26 +385,24 @@ class Admin:
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: location where the message was sent to :param target str: location where the message was sent to
:param is_admin bool: the user is an admin
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
if target == self.client.nick: if target == self.client.nick:
self.logger.debug("We are being called by {}".format(nick)) self.logger.debug("We are being called by {}".format(nick))
if is_admin: if kwargs.get('is_admin', None) is True:
admin = self.is_admin(nick, kwargs['host']) admin = self.is_admin(nick, kwargs['host'])
if admin: if admin:
self.admins[admin]['logged_in'] = False self.admins[admin]['logged_in'] = False
self.admins[admin]['logged_in_nick'] = None self.admins[admin]['logged_in_nick'] = None
self.admins[admin]['logged_in_hostname'] = None self.admins[admin]['logged_in_hostname'] = None
is_admin = False kwargs['is_admin'] = False
_target = nick kwargs['target'] = nick
_message = "{}, you are logged out of " \ kwargs['message'] = "{}, you are logged out of" \
"successfully".format(_target, admin) " successfully".format(nick, admin)
self.logger.debug("We have successfully logged {}" self.logger.debug("We have successfully logged {}"
" out, notifying".format(_target)) " out, notifying".format(nick))
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
self.logger.debug("We are calling save_config") self.logger.debug("We are calling save_config")
await self.save_config() await self.save_config()
@ -421,7 +410,6 @@ class Admin:
nick: str, nick: str,
target: str, target: str,
message: str, message: str,
is_admin: bool,
**kwargs) -> None: **kwargs) -> None:
""" """
This method will change the password to the administrator currently This method will change the password to the administrator currently
@ -432,13 +420,12 @@ class Admin:
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: the target where the message was sent to :param target str: the target where the message was sent to
:param message str: the message sent to target :param message str: the message sent to target
:param is_admin bool: the user is an admin
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
if target == self.client.nick: if target == self.client.nick:
self.logger.debug("We are being called by {}".format(nick)) self.logger.debug("We are being called by {}".format(nick))
if is_admin: if kwargs.get('is_admin', None) is True:
match = self.PASSWD.match(message) match = self.PASSWD.match(message)
if len(match.groups()) == 1: if len(match.groups()) == 1:
admin = self.is_admin(nick, kwargs['host']) admin = self.is_admin(nick, kwargs['host'])
@ -446,26 +433,24 @@ class Admin:
self.admins[admin]['LOGIN'] = re.compile( self.admins[admin]['LOGIN'] = re.compile(
r'^login\s+({})\s+({})\s*$'.format( r'^login\s+({})\s+({})\s*$'.format(
admin, re.escape(self.hash(match.group(1))))) admin, re.escape(self.hash(match.group(1)))))
_target = nick kwargs['target'] = nick
_message = \ kwargs['message'] = \
'{}, password for {} has been successfully' \ '{}, password for {} has been successfully' \
' changed...'.format(_target, admin) ' changed...'.format(nick, admin)
self.logger.debug( self.logger.debug(
"We have successfully changed {}'s password," "We have successfully changed {}'s password,"
" notifying".format(_target)) " notifying".format(nick))
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
self.logger.debug("We are calling save_config()") self.logger.debug("We are calling save_config()")
await self.save_config() await self.save_config()
_target = self.client.nick kwargs['target'] = self.client.nick
self.logger.debug("We are logging {} out".format(_target)) self.logger.debug("We are logging {} out".format(nick))
await self.log_out(nick=nick, target=_target, is_admin=is_admin, **kwargs) await self.log_out(nick, **kwargs)
async def admin_add(self, async def admin_add(self,
nick: str, nick: str,
target: str, target: str,
message: str, message: str,
is_admin: bool,
level: int,
**kwargs) -> None: **kwargs) -> None:
""" """
This method will add an administrator to the list of administrators This method will add an administrator to the list of administrators
@ -476,42 +461,39 @@ class Admin:
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: location where the message was sent to :param target str: location where the message was sent to
:param message str: the message being sent to the target :param message str: the message being sent to the target
:param is_admin bool: the user is an admin
:param level int: the access level of the user being added
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
if target == self.client.nick: if target == self.client.nick:
self.logger.debug("We are being called by {}".format(nick)) self.logger.debug("We are being called by {}".format(nick))
if is_admin: if kwargs.get('is_admin', None) is True:
match = self.ADD.match(message) match = self.ADD.match(message)
if len(match.groups()) == 3: if len(match.groups()) == 3:
if self.admins.get(match.group(1), None) is None: if self.admins.get(match.group(1), None) is None:
_target = nick kwargs['target'] = nick
level = self.level_up(level, level = self.level_up(kwargs['level'],
int(match.group(3))) int(match.group(3)))
_message = "{} has been added with " \ kwargs['message'] = "{} has been added with " \
"level {}...".format(match.group(1), level) "level {}...".format(
match.group(1), level)
self.logger.debug( self.logger.debug(
"We have added {} with level {}, notifying" "We have added {} with level {}, notifying"
" {}".format(match.group(1), level, _target)) " {}".format(match.group(1), level, nick))
await self.add_admin( await self.add_admin(
match.group(1), match.group(2), level) match.group(1), match.group(2), level)
else: else:
_target = nick kwargs['target'] = nick
_message = "{} has already been added" \ kwargs['message'] = "{} has already been added" \
"...".format(match.group(1)) "...".format(match.group(1))
self.logger.warn( self.logger.warn(
"We detected that {} has already been added," "We detected that {} has already been added,"
" notifying {}".format(match.group(1), _target)) " notifying {}".format(match.group(1), nick))
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
async def admin_rm(self, async def admin_rm(self,
nick: str, nick: str,
target: str, target: str,
message: str, message: str,
is_admin: bool,
level: int,
**kwags) -> None: **kwags) -> None:
""" """
This method will remove an administrator from the list of This method will remove an administrator from the list of
@ -524,59 +506,56 @@ class Admin:
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: location where the message was sent to :param target str: location where the message was sent to
:param message str: the message being sent to the target :param message str: the message being sent to the target
:param is_admin bool: the user is an admin
:param level int: the access level of the user being removed
:param kwags: for API compatibility :param kwags: for API compatibility
:return: None :return: None
""" """
if target == self.client.nick: if target == self.client.nick:
self.logger.debug("We are being called by {}".format(nick)) self.logger.debug("We are being called by {}".format(nick))
if is_admin: if kwags.get('is_admin', None) is True:
match = self.RM.match(message) match = self.RM.match(message)
if len(match.groups()) == 1: if len(match.groups()) == 1:
if self.admins.get(match.group(1), None) is None: if self.admins.get(match.group(1), None) is None:
_target = nick kwags['target'] = nick
_message = "{} is not on the list" \ kwags['message'] = "{} is not on the list" \
"...".format(match.group(1)) "...".format(match.group(1))
self.logger.warn("admin_rm() ") self.logger.warn("admin_rm() ")
else: else:
if level > self.admins[match.group(1)]['level'] \ if kwags['level'] > \
or (level == 1000 and self.admins[match.group(1)]['level'] self.admins[match.group(1)]['level'] \
and level == self.admins[match.group(1)]['level']): or (kwags['level'] == 1000
_target = nick and self.admins[match.group(1)]['level']
_message = "{} has been removed" \ and kwags['level'] ==
self.admins[match.group(1)]['level']):
kwags['target'] = nick
kwags['message'] = "{} has been removed" \
"...".format(match.group(1)) "...".format(match.group(1))
self.logger.debug("We removed {} successfully," self.logger.debug("We removed {} successfully,"
" notifying {}".format( " notifying {}".format(
match.group(1), _target)) match.group(1), nick))
await self.rm_admin(match.group(1)) await self.rm_admin(match.group(1))
else: else:
_target = nick kwags['target'] = nick
_message = "{}, you do not have enough" \ kwags['message'] = "{}, you do not have enough" \
" access to delete {}".format(_target, match.group(1)) " access to delete {}".format(
nick, match.group(1))
self.logger.warn( self.logger.warn(
"We detected that {0} does not have enough" "We detected that {0} does not have enough"
" access to delete {1}, notifying {0}".format( " access to delete {1}, notifying {0}".format(
_target, match.group(1))) nick, match.group(1)))
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwags)
async def admin_list(self, def admin_list(self, nick: str, target: str, **kwargs) -> None:
nick: str,
target: str,
is_admin: bool,
**kwargs) -> None:
""" """
This method returns to the admin the admin list This method returns to the admin the admin list
:param nick str: the nickname of the caller :param nick str: the nickname of the caller
:param target str: location where the message was sent to :param target str: location where the message was sent to
:param is_admin bool: the user is an admin
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
if target == self.client.nick: if target == self.client.nick:
self.logger.debug("We are being called by {}".format(nick)) self.logger.debug("We are being called by {}".format(nick))
if is_admin: if kwargs.get('is_admin', None) is True:
admins = "" admins = ""
for key, _ in self.admins.items(): for key, _ in self.admins.items():
if admins: if admins:
@ -585,13 +564,13 @@ class Admin:
else: else:
admins = "{}({})".format( admins = "{}({})".format(
key, self.admins[key]['level']) key, self.admins[key]['level'])
_target = nick kwargs['target'] = nick
_message = "List of Administrators:" kwargs['message'] = "List of Administrators:"
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
_message = admins kwargs['message'] = admins
self.logger.debug("We are returning admin list page to" self.logger.debug("We are returning admin list page to"
" {}".format(_target)) " {}".format(kwargs))
await self.client.msg(target=_target, message=_message) self.client.send("PRIVMSG", **kwargs)
def is_admin(self, user: str, host: str): def is_admin(self, user: str, host: str):
""" """

View file

@ -3,7 +3,6 @@ import re
import functools import functools
import logging import logging
import asyncio import asyncio
import admin import admin
@ -22,25 +21,18 @@ class AdminCmd:
self.services = {} self.services = {}
admin.client.on("ADMIN")(self._handle) admin.client.on("ADMIN")(self._handle)
def _handle(self, def _handle(self, target: str, message: str, **kwargs) -> None:
nick: str,
target: str,
message: str,
is_admin: bool,
**kwargs) -> None:
""" """
client callback on event trigger client callback on event trigger
:param nick str: the nickname of the user triggering the ADMIN event
:param target str: the target the message was sent to :param target str: the target the message was sent to
:param message str: the message sent to the target :param message str: the message sent to the target
:param is_admin bool: the user is an admin
:param kwargs: for API compatibility :param kwargs: for API compatibility
:return: None :return: None
""" """
if is_admin: if bool(kwargs.get('is_admin', None)):
self.logger.debug( self.logger.debug(
"We are being called by {}".format(nick)) "We are being called by {}".format(kwargs['nick']))
for regex, (func, pattern) in self.services.items(): for regex, (func, pattern) in self.services.items():
match = regex.match(message) match = regex.match(message)
if match: if match:
@ -50,7 +42,7 @@ class AdminCmd:
split_msg = message.split(" ") split_msg = message.split(" ")
message = " ".join(split_msg[1:]) message = " ".join(split_msg[1:])
self.client.loop.create_task( self.client.loop.create_task(
func(nick, target, message, **kwargs)) func(target, message, **kwargs))
def on_command(self, def on_command(self,
command: str, command: str,

View file

@ -1,4 +1,3 @@
import re
import logging import logging
import asyncio import asyncio
import robot import robot
@ -22,21 +21,21 @@ async def plugins(bot: robot.Bot):
bot.send("PRIVMSG", target=target, message=message) bot.send("PRIVMSG", target=target, message=message)
await bot.wait("NOTICE") await bot.wait("NOTICE")
# Code below is an example # Code below will not work, it is awaiting a PR upstream
@bot.on("USERMODE") # @bot.on("USERMODE")
def umode(**kwargs): # def umode(**kwargs):
logger.debug("USERMODE {}".format(kwargs)) # logger.debug("USERMODE {}".format(kwargs))
@bot.on("CHANNELMODE") # @bot.on("CHANNELMODE")
def cmode(**kwargs): # def cmode(**kwargs):
logger.debug("CHANNELMODE {}".format(kwargs)) # logger.debug("CHANNELMODE {}".format(kwargs))
# 8Ball # 8Ball
magic_ball = eightball.EightBall(bot) magic_ball = eightball.EightBall(bot)
@magic_ball.on_keyword @magic_ball.on_keyword
async def ball(target, message, **kwargs): async def ball(target, message, **kwargs):
await bot.msg(target=target, message=message) bot.send("PRIVMSG", target=target, message=message)
administrator = admin.Admin(bot) administrator = admin.Admin(bot)
await administrator.init() await administrator.init()
@ -44,102 +43,29 @@ async def plugins(bot: robot.Bot):
admin_cmd = admin_commands.AdminCmd(administrator) admin_cmd = admin_commands.AdminCmd(administrator)
@admin_cmd.on_command("join") @admin_cmd.on_command("join")
async def join(nick, target, message, **kwargs): def join(target, message, **kwargs):
await bot.join(channel=message) bot.send("JOIN", channel=message)
@admin_cmd.on_command("part") @admin_cmd.on_command("part")
async def part(nick, target, message, **kwargs): def part(target, message, **kwargs):
_target = message.split(' ')[0] bot.send("PART", channel=message)
_message = " ".join(message.split(' ')[1:])
await bot.part(channel=_target, message=_message)
@admin_cmd.on_command("msg") @admin_cmd.on_command("msg")
async def msg(nick, target, message, **kwargs): def msg(target, message, **kwargs):
_target = message.split(' ')[0] kwargs['target'] = message.split(' ')[0]
_message = " ".join(message.split(' ')[1:]) kwargs['message'] = " ".join(message.split(' ')[1:])
await bot.msg(target=_target, message=_message) bot.send("PRIVMSG", **kwargs)
@admin_cmd.on_command("action") @admin_cmd.on_command("action")
async def action(nick, target, message, **kwargs): def action(target, message, **kwargs):
_target = message.split(' ')[0] kwargs['target'] = message.split(' ')[0]
_message = " ".join(message.split(' ')[1:]) kwargs['message'] = \
await bot.action(target=_target, message=_message) "\x01ACTION {}\x01".format(" ".join(message.split(' ')[1:]))
bot.send("PRIVMSG", **kwargs)
@admin_cmd.on_command("notice")
async def msg(nick, arget, message, **kwargs):
_target = message.split(' ')[0]
_message = " ".join(message.split(' ')[1:])
await bot.notice(target=_target, message=_message)
@admin_cmd.on_command("mode")
async def chanmode(nick, target, message, **kwargs):
chanmode_pattern = r'^(?P<channel>(\#|\#\#|\&)[a-zA-Z0-9]+)?' \
r'((^|\s+)(?P<modes>[\+\-][a-zA-Z]+))' \
r'(\s+(?P<params>.+)?)?'
compiled = re.compile(chanmode_pattern)
m = compiled.match(message)
if m.group("channel"):
_channel = m.group("channel")
else:
_channel = target
if m.group("modes"):
_modes = m.group("modes")
else:
await bot.msg(target=target, message="{}, you did not provide a mode".format(nick))
return
if m.group("params"):
_params = m.group("params")
else:
_params = None
await bot.channelmode(channel=_channel, modes=_modes, params=_params)
@admin_cmd.on_command("op")
async def op(nick, target, message, **kwargs):
_split_msg = message.split(' ')
nick_list = None
_channel = target
if _split_msg[0]:
if _split_msg[0].startswith('#') or _split_msg[0].startswith('&'):
_channel = _split_msg[0]
if _split_msg.__len__() > 1:
nick_list = _split_msg[1:]
else:
nick_list = _split_msg
if nick_list:
_modes = "+" + "o" * nick_list.__len__()
await bot.channelmode(channel=_channel, modes=_modes, params=" ".join(nick_list))
else:
await bot.channelmode(channel=_channel, modes="+o", params=nick)
@admin_cmd.on_command("deop")
async def op(nick, target, message, **kwargs):
_split_msg = message.split(' ')
nick_list = None
_channel = target
if _split_msg[0]:
if _split_msg[0].startswith('#') or _split_msg[0].startswith('&'):
_channel = _split_msg[0]
if _split_msg.__len__() > 1:
nick_list = _split_msg[1:]
else:
nick_list = _split_msg
if nick_list:
_modes = "-" + "o" * nick_list.__len__()
await bot.channelmode(channel=_channel, modes=_modes, params=" ".join(nick_list))
else:
await bot.channelmode(channel=_channel, modes="-o", params=nick)
@admin_cmd.on_command("quit") @admin_cmd.on_command("quit")
async def quit(nick, target, message, **kwargs): async def quit(target, message, **kwargs):
await bot.quit(message=message) bot.send("QUIT", message=message)
await bot.disconnect() await bot.disconnect()
# Exit the event loop cleanly # Exit the event loop cleanly
bot.loop.stop() bot.loop.stop()
@ -152,21 +78,16 @@ def main():
ssl = False ssl = False
nick = "LuckyBoots" nick = "LuckyBoots"
channel = ["#boots"] channel = "#boots"
bot = robot.Bot(host=host, port=port, bot = robot.Bot(host=host, port=port,
ssl=ssl, nick=nick, ssl=ssl, nick=nick,
channels=channel) channels=[channel])
bot.loop.create_task(bot.connect()) bot.loop.create_task(bot.connect())
asyncio.ensure_future(plugins(bot)) asyncio.ensure_future(plugins(bot))
try:
bot.loop.run_forever() bot.loop.run_forever()
except KeyboardInterrupt:
bot.loop.stop()
bot.loop.close() bot.loop.close()

View file

@ -4,7 +4,6 @@ import random
import re import re
import functools import functools
import asyncio import asyncio
import robot import robot

View file

@ -1,9 +1,7 @@
import logging import logging
import asyncio import asyncio
import types import types
import typing
import functools import functools
import datetime
import bottom import bottom
@ -19,15 +17,8 @@ class Bot(bottom.Client):
nick: str, nick: str,
user: str = None, user: str = None,
realname: str = None, realname: str = None,
channels: list = None, channels: list = None) -> None:
msg_flood_size_threshold = 3, super().__init__(host=host, port=port, ssl=ssl)
msg_flood_time_threshold = 5,
action_flood_size_threshold=3,
action_flood_time_threshold=5,
global_flood_size_threshold=3,
global_flood_time_threshold=5,
loop: typing.Optional[asyncio.AbstractEventLoop] = None) -> None:
super().__init__(host=host, port=port, ssl=ssl, loop=loop)
self.logger = logging.getLogger(self.__class__.__name__) self.logger = logging.getLogger(self.__class__.__name__)
self.logger.debug("Initializing...") self.logger.debug("Initializing...")
self.nick = nick self.nick = nick
@ -35,19 +26,10 @@ class Bot(bottom.Client):
self.realname = realname or self.nick self.realname = realname or self.nick
self.channels = channels or [] self.channels = channels or []
self.pre_pub = [] self.pre_pub = []
self.msg_flood_size_threshold = msg_flood_size_threshold
self.msg_flood_time_threshold = msg_flood_time_threshold
self.action_flood_size_threshold = action_flood_size_threshold
self.action_flood_time_threshold = action_flood_time_threshold
self.global_flood_size_threshold = global_flood_size_threshold
self.global_flood_time_threshold = global_flood_time_threshold
self.msg_timestamp_queue = []
self.action_timestamp_queue = []
self.global_timestamp_queue = []
self.on("ping", self.keepalive) self.on("ping", self.keepalive)
self.on("CLIENT_CONNECT", self.on_connect) self.on("CLIENT_CONNECT", self.on_connect)
self.on("PRIVMSG", self.on_privmsg) self.on("PRIVMSG", self.privmsg)
self.on("NOTICE", self.on_notice) self.on("NOTICE", self.notice)
async def keepalive(self, message: str = None, **kwargs) -> None: async def keepalive(self, message: str = None, **kwargs) -> None:
""" """
@ -95,9 +77,6 @@ class Bot(bottom.Client):
self.logger.debug("Running {}".format(func)) self.logger.debug("Running {}".format(func))
await self.loop.create_task(func(**kwargs)) await self.loop.create_task(func(**kwargs))
# Wait 5 seconds to make sure everything synched up
await asyncio.sleep(5)
self.logger.info( self.logger.info(
"We are auto-joining channels {}".format(self.channels)) "We are auto-joining channels {}".format(self.channels))
for channel in self.channels: for channel in self.channels:
@ -124,193 +103,10 @@ class Bot(bottom.Client):
self.pre_pub.extend([wrapped]) self.pre_pub.extend([wrapped])
return func return func
@staticmethod def privmsg(self, **kwargs) -> None:
async def _wait(timestamp_queue: list, flood_size_threshold: int,
flood_time_threshold: int, logger: logging.Logger) -> list:
"""
This method will figure out if the calling function needs to wait.
If so, it will wait the correct amount of time before returning.
:param timestamp_queue list: list of timestamp of the size of flood_size_threshold
:param flood_size_threshold int: total size of the flood threshold
:param flood_time_threshold int: total time of the flood threshold
:param logger logging.Logger: logger to use
:return timetamp_queue list: the new timestamp_queue after modifications
"""
timestamp = datetime.datetime.now()
if timestamp_queue.__len__() < flood_size_threshold:
timestamp_queue.extend([timestamp])
else:
del timestamp_queue[0]
timestamp_queue.extend([timestamp])
time_diff = timestamp_queue[-1] - timestamp_queue[0]
if len(timestamp_queue) == flood_size_threshold and \
time_diff.total_seconds() < flood_time_threshold:
logger.debug("Waiting {}s".format(flood_time_threshold - time_diff.total_seconds()))
await asyncio.sleep(flood_time_threshold - time_diff.total_seconds())
return timestamp_queue
async def global_buffer(self, command: str, **kwargs) -> None:
"""
This method will buffer the communication sent to the server
to keep the client from getting disconnected
It uses global_* variables as configuration
NOTE: this method is to be called *only* by other buffer methods
:param command str: the command to send to the server
:param kwargs: the information required for the command
:return None
"""
self.global_timestamp_queue = await self._wait(self.global_timestamp_queue,
self.global_flood_size_threshold,
self.global_flood_time_threshold,
self.logger)
self.send(command, **kwargs)
async def msg_buffer(self, command: str, **kwargs) -> None:
"""
This method will buffer the messages sent to the server
to keep the client from getting disconnected
messages are of type: PRIVMSG which includes ACTION
It uses msg_* variables as configuration
NOTE: this method is to be called *only* by bot methods
:param command str: the message command to send to the server
:param kwargs: the information required for the message command
:return None
"""
self.msg_timestamp_queue = await self._wait(self.msg_timestamp_queue,
self.msg_flood_size_threshold,
self.msg_flood_time_threshold,
self.logger)
await self.global_buffer(command, **kwargs)
async def action_buffer(self, command: str, **kwargs):
"""
This method will buffer the actions sent to the server
to keep the client from getting disconnected
actions are of type: JOIN, PART and QUIT
It uses action_* variables as configuration
NOTE: this method is to be called *only* by bot methods
:param command str: the action command to send to the server
:param kwargs: the information required for the action command
:return None
"""
self.action_timestamp_queue = await self._wait(self.action_timestamp_queue,
self.action_flood_size_threshold,
self.action_flood_time_threshold,
self.logger)
await self.global_buffer(command, **kwargs)
async def msg(self, target: str, message: str, **kwargs) -> None:
"""
This method will send a private message to the messages buffer
:param target str: the target for the message to be sent to
:param message str: the message to send to target
:param kwargs: any other parameters that needs to be sent
:return None
"""
await self.msg_buffer("PRIVMSG", target=target, message=message, **kwargs)
async def action(self, target: str, message: str, **kwargs) -> None:
"""
This method will send an action message to the messages buffer
:param target str: the target for the message to be sent to
:param message str: the message to send to target
:param kwargs: any other parameters that needs to be sent
:return None
"""
_message = "\x01ACTION {}\x01".format(message)
await self.msg_buffer("PRIVMSG", target=target, message=_message, **kwargs)
async def notice(self, target: str, message: str, **kwargs) -> None:
"""
This method will send a notice message to the messages buffer
:param target str: the target for the notice to be sent to
:param message str: the message to send to target
:param kwargs: any other parameters that needs to be sent
:return None
"""
await self.msg_buffer("NOTICE", target=target, message=message, **kwargs)
async def join(self, channel: str, **kwargs) -> None:
"""
This method will send a join command to the action buffer
:param channel str: the channel to join
:param kwargs: any other parameters that needs to be sent
:return None
"""
await self.action_buffer("JOIN", channel=channel, **kwargs)
async def usermode(self, nick: str, modes: str, **kwargs):
"""
This method will send a usermode command to the server
:param nick str: the nick to apply usermode action to
:param modes str: the mode to apply to the nick
:param kwargs: any other parameters that needs to be sent
:return None
"""
self.send("USERMODE", nick=nick, modes=modes)
async def channelmode(self, channel: str, modes: str, params: str = None ,**kwargs):
"""
This method will send a channelmode command to the server
:param channel str: the nick to apply usermode action to
:param modes str: the mode to apply to the channel
:param params str: the parameters to apply to the modes
:param kwargs: any other parameters that needs to be sent
:return None
"""
if params:
self.send("CHANNELMODE", channel=channel, modes=modes, params=params)
else:
self.send("CHANNELMODE", channel=channel, modes=modes)
async def part(self, channel: str, message: str = None, **kwargs) -> None:
"""
This method will send a part command to the action buffer
:param channel str: the channel to part
:param message str: the message to send on part
:param kwargs: any other parameters that needs to be sent
:return None
"""
await self.action_buffer("PART", channel=channel, message=message, **kwargs)
async def quit(self, message: str, **kwargs) -> None:
"""
This method will send a quit command to the action buffer
:param message str: the message to send on quit
:param kwargs: any other parameters that needs to be sent
:return None
"""
await self.action_buffer("QUIT", message=message, **kwargs)
def on_privmsg(self, **kwargs) -> None:
self.logger.debug("PRIVMSG {}".format(kwargs)) self.logger.debug("PRIVMSG {}".format(kwargs))
def on_notice(self, **kwargs) -> None: def notice(self, **kwargs) -> None:
self.logger.debug("NOTICE {}".format(kwargs)) self.logger.debug("NOTICE {}".format(kwargs))
async def on_disconnect(self, **kwargs) -> None: async def on_disconnect(self, **kwargs) -> None: