201 lines
5.8 KiB
Python
201 lines
5.8 KiB
Python
import logging
|
||
import re
|
||
from irc_api.irc import IRC, History
|
||
|
||
|
||
PREFIX = ""
|
||
|
||
|
||
def command(name, desc=""):
|
||
def decorator(func):
|
||
return Command(
|
||
name=name,
|
||
func=func,
|
||
events=[lambda m: m.text.startswith(PREFIX + name)],
|
||
desc=desc,
|
||
cmnd_type=1
|
||
)
|
||
return decorator
|
||
|
||
|
||
def on(event, desc=""):
|
||
def decorator(func):
|
||
return Command(
|
||
name=func.__name__,
|
||
func=func,
|
||
events=[event],
|
||
desc=desc,
|
||
cmnd_type=0
|
||
)
|
||
return decorator
|
||
|
||
|
||
def channel(channel_name, desc=""):
|
||
def decorator(func_or_cmnd):
|
||
if isinstance(func_or_cmnd, Command):
|
||
func_or_cmnd.events.append(lambda m: m.to == channel_name)
|
||
return func_or_cmnd
|
||
else:
|
||
return Command(
|
||
name=func_or_cmnd.__name__,
|
||
func=func_or_cmnd,
|
||
events=[lambda m: m.to == channel_name],
|
||
desc=desc,
|
||
cmnd_type=0
|
||
)
|
||
|
||
return decorator
|
||
|
||
|
||
class Command:
|
||
def __init__(self, name, func, events, desc, cmnd_type):
|
||
self.name = name
|
||
self.func = func
|
||
self.events = events
|
||
self.cmnd_type = cmnd_type
|
||
|
||
if desc:
|
||
self.desc = desc
|
||
else:
|
||
self.desc = "..."
|
||
if func.__doc__:
|
||
self.desc = func.__doc__
|
||
|
||
self.bot = None
|
||
|
||
def __call__(self, msg, *args):
|
||
return self.func(self.bot, msg, *args)
|
||
|
||
|
||
class Bot:
|
||
"""Run the connexion between IRC's server and V5 one.
|
||
|
||
Attributes
|
||
----------
|
||
irc : IRC, public
|
||
IRC wrapper which handle communication with IRC server.
|
||
v5 : V5, public
|
||
V5 wrapper which handle communication with V5 server.
|
||
channels : list, public
|
||
The channels the bot will listen.
|
||
|
||
Methods
|
||
-------
|
||
start : NoneType, public
|
||
Runs the bot and connects it to IRC and V5 servers.
|
||
"""
|
||
def __init__(
|
||
self,
|
||
auth: tuple,
|
||
irc_params: tuple,
|
||
channels: list=["#general"],
|
||
*commands_modules,
|
||
**kwargs
|
||
):
|
||
"""Initialize the Bot instance.
|
||
|
||
Parameters
|
||
----------
|
||
irc_params : tuple
|
||
Contains the IRC server informations (host, port)
|
||
channels : list
|
||
Contains the names of the channels on which the bot will connect.
|
||
prefix : str, optionnal
|
||
The prefix on which the bot will react.
|
||
"""
|
||
global PREFIX
|
||
if kwargs.get('prefix'):
|
||
PREFIX = kwargs.get('prefix')
|
||
|
||
self.irc = IRC(*irc_params)
|
||
self.history = History(kwargs.get('limit'))
|
||
self.channels = channels
|
||
self.auth = auth
|
||
self.callbacks = {}
|
||
|
||
if commands_modules:
|
||
self.add_commands_modules(*commands_modules)
|
||
|
||
def start(self):
|
||
"""Starts the bot and connect it to the given IRC and V5 servers."""
|
||
# Start IRC
|
||
self.irc.connexion(self.auth[0], self.auth[1])
|
||
|
||
# Join channels
|
||
for channel in self.channels:
|
||
self.irc.join(channel)
|
||
|
||
# mainloop
|
||
while True:
|
||
message = self.irc.receive()
|
||
self.history.add(message)
|
||
logging.info("received %s", message)
|
||
if message is not None:
|
||
for callback in self.callbacks.values():
|
||
if not False in [event(message) for event in callback.events]:
|
||
logging.info("matched %s", callback.name)
|
||
callback(message, *parse(message.text)[callback.cmnd_type:])
|
||
|
||
def send(self, target: str, message: str):
|
||
"""Send a message to the specified target (channel or user).
|
||
|
||
Parameters
|
||
----------
|
||
target : str
|
||
The target of the message. It can be a channel or user (private message).
|
||
message : str
|
||
The content of the message to send.
|
||
"""
|
||
self.irc.send(f"PRIVMSG {target} :{message}")
|
||
|
||
def add_command(self, command):
|
||
command.bot = self
|
||
self.callbacks[command.name] = command
|
||
|
||
def add_commands(self, *commands):
|
||
"""Add a list of commands to the bot.
|
||
|
||
Parameters
|
||
----------
|
||
commands : list
|
||
A list of command's instances.
|
||
"""
|
||
for command in commands:
|
||
self.add_command(command)
|
||
|
||
def add_commands_modules(self, *commands_modules):
|
||
for commands_module in commands_modules:
|
||
for cmnd_name in dir(commands_module):
|
||
cmnd = getattr(commands_module, cmnd_name)
|
||
if isinstance(cmnd, Command):
|
||
self.add_command(cmnd)
|
||
|
||
def remove_command(self, command_name: str):
|
||
if command_name in self.callbacks.keys():
|
||
self.callbacks.pop(command_name)
|
||
|
||
|
||
@command("aide")
|
||
def auto_help(bot, msg, *args):
|
||
"""Aide des commandes disponibles."""
|
||
if args and args[0] in bot.callbacks.keys():
|
||
bot.send(msg.to, f"Aide sur la commande : {args[0]}")
|
||
bot.send(msg.to, f" {bot.callbacks[args[0]].desc}")
|
||
else:
|
||
bot.send(msg.to, f"Liste des commandes ({PREFIX}aide <cmnd> pour plus d'info)")
|
||
for cmnd_name in bot.callbacks.keys():
|
||
bot.send(msg.to, f" – {cmnd_name}")
|
||
|
||
|
||
def parse(message):
|
||
pattern = re.compile(r"((\"[\w\ \-\']+\"\ *)|(\'[\w\ \-\"]+\'\ *)|([\w\'\-]+\ *))")
|
||
args_to_return = []
|
||
for match in re.findall(pattern, message):
|
||
match = match[0].strip().rstrip()
|
||
if (match.startswith("\"") and match.endswith("\"")) \
|
||
or (match.startswith("'") and match.endswith("'")):
|
||
args_to_return.append(match[1: -1])
|
||
else:
|
||
args_to_return.append(match)
|
||
return args_to_return
|