bf76c3327c
* No longer using kwargs to send data, it is all parametrized now * API calls to msg, action, join, part, quit have been parametrized * New API call to notice
78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
import types
|
|
import re
|
|
import functools
|
|
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(self.__class__.__name__)
|
|
self.logger.debug("Initializing...")
|
|
self.admin = admin
|
|
self.client = admin.client
|
|
self.modifier = modifier
|
|
self.modifier_pattern = r'^{}([^\s]+).*$'.format(modifier)
|
|
self.services = {}
|
|
admin.client.on("ADMIN")(self._handle)
|
|
|
|
def _handle(self,
|
|
nick: str,
|
|
target: str,
|
|
message: str,
|
|
is_admin: bool,
|
|
**kwargs) -> None:
|
|
"""
|
|
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 message str: the message sent to the target
|
|
:param is_admin bool: the user is an admin
|
|
:param kwargs: for API compatibility
|
|
:return: None
|
|
"""
|
|
if is_admin:
|
|
self.logger.debug(
|
|
"We are being called by {}".format(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))
|
|
split_msg = message.split(" ")
|
|
message = " ".join(split_msg[1:])
|
|
self.client.loop.create_task(
|
|
func(target, message, **kwargs))
|
|
|
|
def on_command(self,
|
|
command: str,
|
|
func: types.FunctionType = None,
|
|
** 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: for API compatibility
|
|
:return: types.FunctionType the function that called it
|
|
"""
|
|
if func is None:
|
|
return functools.partial(self.on_command, command)
|
|
|
|
wrapped = func
|
|
if not asyncio.iscoroutinefunction(wrapped):
|
|
wrapped = asyncio.coroutine(wrapped)
|
|
|
|
compiled = re.compile(r'^{}{}\s*(.*)$'.format(
|
|
re.escape(self.modifier), command))
|
|
self.services[compiled] = (wrapped, command)
|
|
|
|
return func
|