269 lines
7.4 KiB
Python
269 lines
7.4 KiB
Python
"""
|
|
irc (GLaDOS)
|
|
============
|
|
|
|
Description
|
|
-----------
|
|
Manage the IRC layer of GLaDOS.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
import socket
|
|
import ssl
|
|
|
|
from functools import wraps
|
|
from queue import Queue
|
|
from threading import Thread
|
|
|
|
|
|
class IRC:
|
|
"""Manage connexion to an IRC server, authentication and callbacks.
|
|
|
|
Attributes
|
|
----------
|
|
connected : bool, public
|
|
If the bot is connected to an IRC server or not.
|
|
|
|
socket : ssl.SSLSocket, private
|
|
The IRC's socket.
|
|
inbox : Queue, private
|
|
Queue of the incomming messages.
|
|
handler : Thread, private
|
|
callbacks : list, private
|
|
List of the registred callbacks.
|
|
|
|
Methods
|
|
-------
|
|
start : NoneType, public
|
|
Starts the IRC layer and manage authentication.
|
|
run : NoneType, public
|
|
Mainloop, allows to handle public messages.
|
|
send : NoneType, public
|
|
Sends a message to a given channel.
|
|
receive : Message, public
|
|
Same as ``run`` for private messages.
|
|
join : NoneType, public
|
|
Allows to join a given channel.
|
|
on : function, public
|
|
Add a callback on a given message.
|
|
|
|
handle : NoneType, private
|
|
Handles the ping and store incoming messages into the inbox attribute.
|
|
send : NoneType, private
|
|
Send message to a target.
|
|
recv : str, private
|
|
Get the oldest incoming message and returns it.
|
|
waitfor : str, private
|
|
Wait for a raw message that matches the given condition.
|
|
"""
|
|
def __init__(self, host: str, port: int):
|
|
"""Initialize an IRC wrapper.
|
|
|
|
Parameters
|
|
----------
|
|
host : str
|
|
The adress of the IRC server.
|
|
port : int
|
|
The port of the IRC server.
|
|
"""
|
|
|
|
# Public attributes
|
|
self.connected = False # Simple lock
|
|
|
|
# Private attributes
|
|
self.__socket = ssl.create_default_context().wrap_socket(
|
|
socket.create_connection((host, port)),
|
|
server_hostname=host
|
|
)
|
|
self.__inbox = Queue()
|
|
self.__handler = Thread(target=self.__handle)
|
|
self.__callbacks = []
|
|
|
|
# Public methods
|
|
def start(self, nick: str, password: str):
|
|
"""Start the IRC layer. Manage authentication as well.
|
|
|
|
Parameters
|
|
----------
|
|
nick : str
|
|
The username for login and nickname once connected.
|
|
password : str
|
|
The password for authentification.
|
|
"""
|
|
self.__handler.start()
|
|
|
|
self.__send(f"USER {nick} * * :{nick}")
|
|
self.__send(f"NICK {nick}")
|
|
self.__waitfor(lambda m: "NOTICE" in m and "/AUTH" in m)
|
|
self.__send(f"AUTH {nick}:{password}")
|
|
self.__waitfor(lambda m: "You are now logged in" in m)
|
|
|
|
self.connected = True
|
|
|
|
def run(self):
|
|
"""Handle new messages."""
|
|
while True:
|
|
message = self.receive()
|
|
logging.info("received %s", message)
|
|
if message is not None:
|
|
for event, callback in self.__callbacks:
|
|
if event(message):
|
|
logging.info("matched %s", event.__name__)
|
|
callback(message)
|
|
|
|
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.__send(f"PRIVMSG {target} :{message}")
|
|
|
|
def receive(self):
|
|
"""Receive a private message.
|
|
|
|
Returns
|
|
-------
|
|
msg : Message
|
|
The incoming processed private message.
|
|
"""
|
|
while True:
|
|
message = self.__recv()
|
|
if " PRIVMSG " in message:
|
|
msg = Message(message)
|
|
if msg:
|
|
return msg
|
|
|
|
def join(self, channel: str):
|
|
"""Join a channel.
|
|
|
|
Parameters
|
|
----------
|
|
channel : str
|
|
The name of the channel to join.
|
|
"""
|
|
self.__send(f"JOIN {channel}")
|
|
logging.info("joined %s", channel)
|
|
|
|
def on(self, event):
|
|
"""Adds a callback to the IRC handler.
|
|
|
|
Parameters
|
|
----------
|
|
event : function
|
|
``event`` must taking in parameter a Message and returning ``True`` if the callback
|
|
should be executed on the message.
|
|
|
|
Returns
|
|
-------
|
|
callback : function
|
|
The callback function.
|
|
"""
|
|
def callback(func):
|
|
@wraps(func)
|
|
def wrapper(message):
|
|
func(message)
|
|
self.__callbacks.append((event, wrapper))
|
|
logging.info("added callback %s", func.__name__)
|
|
return wrapper
|
|
|
|
return callback
|
|
|
|
|
|
# Private methods
|
|
def __handle(self):
|
|
"""Handle raw messages from irc and manage ping."""
|
|
while True:
|
|
# Get incoming messages
|
|
data = self.__socket.recv(4096).decode()
|
|
|
|
# Split multiple lines
|
|
for msg in data.split('\r\n'):
|
|
# Manage ping
|
|
if msg.startswith("PING"):
|
|
self.__send(msg.replace("PING", "PONG"))
|
|
# Or add a new message to inbox
|
|
elif len(msg):
|
|
self.__inbox.put(msg)
|
|
logging.debug("received %s", msg)
|
|
|
|
def __send(self, raw: str):
|
|
"""Wrap and encode raw message to send.
|
|
|
|
Parameters
|
|
----------
|
|
raw : str
|
|
The raw message to send.
|
|
"""
|
|
self.__socket.send(f"{raw}\r\n".encode())
|
|
|
|
def __recv(self):
|
|
"""Get the older received message and returns it.
|
|
|
|
Returns
|
|
-------
|
|
m : str
|
|
The raw content of the message to return.
|
|
"""
|
|
msg = self.__inbox.get()
|
|
return msg
|
|
|
|
def __waitfor(self, condition):
|
|
"""Wait for a raw message that matches the condition.
|
|
|
|
Parameters
|
|
----------
|
|
condition : function
|
|
``condition`` is a function that must taking a raw message in parameter and returns a
|
|
boolean.
|
|
|
|
Returns
|
|
-------
|
|
msg : str
|
|
The last message received that doesn't match the condition.
|
|
"""
|
|
msg = self.__recv()
|
|
while not condition(msg):
|
|
msg = self.__recv()
|
|
return msg
|
|
|
|
|
|
class Message:
|
|
"""Parse the raw message in three fields : author, the channel, and text.
|
|
|
|
Attributes
|
|
----------
|
|
pattern : re.Pattern, public
|
|
The message parsing pattern.
|
|
author : str, public
|
|
The message's author.
|
|
to : str, public
|
|
The message's origin (channel or DM).
|
|
text : str, public
|
|
The message's content.
|
|
"""
|
|
pattern = re.compile(
|
|
r"^:(?P<author>[\w.~|\-\[\]]+)(?:!(?P<host>\S+))? PRIVMSG (?P<to>\S+) :(?P<text>.+)"
|
|
)
|
|
|
|
def __init__(self, raw: str):
|
|
match = re.search(Message.pattern, raw)
|
|
if match:
|
|
self.author = match.group("author")
|
|
self.to = match.group("to")
|
|
self.text = match.group("text")
|
|
logging.debug("sucessfully parsed %s into %s", raw, self.__str__())
|
|
else:
|
|
self.author = ""
|
|
self.to = ""
|
|
self.text = ""
|
|
logging.warning("failed to parse %s into valid message", raw)
|
|
|
|
def __str__(self):
|
|
return f"{self.author} to {self.to}: {self.text}"
|