Well, you found me. Congratulations. Was it worth it?
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

134 lines
3.8 KiB

# Manage the IRC layer of GLaDOS
import re, socket, ssl
from functools import wraps
from queue import Queue
from threading import Thread
class IRC(object):
def __init__(self, host, port):
""" Initialize an IRC wrapper """
# Public attributes
self.connected = False # Simple lock
# Private attributes
self._socket = ssl.create_default_context().wrap_socket(
socket.create_connection((host, port)),
self._inbox = Queue()
self._handler = Thread(target=self._handle)
self._callbacks = []
# Public methods
def start(self, nick, password):
""" Start the IRC layer. Manage authentication as well """
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()
for event, callback in self._callbacks:
if event(message):
def send(self, target, message):
""" Send a message to the specified target (channel or user) """
self._send(f"PRIVMSG {target} :{message}")
def receive(self):
""" Receive a private message """
while True:
message = self._recv()
print(f"receive: {message}")
if " PRIVMSG " in message:
return Message(message)
def join(self, channel):
""" Join a channel """
self._send(f"JOIN {channel}")
def on(self, event):
""" Adds a callback to the IRC handler
Event is a function taking in parameter a Message and returning
True if the callback should be executed on the message """
def callback(func):
def wrapper(message):
self._callbacks.append((event, wrapper))
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 m in data.split('\r\n'):
# Manage ping
if m.startswith("PING"):
self._send(m.replace("PING", "PONG"))
# Or add a new message to inbox
elif len(m):
print(f"_handle: <{m}>")
def _send(self, raw):
""" Wrap and encode raw message to send """
def _recv(self):
m = self._inbox.get()
return m
def _waitfor(self, condition):
""" Wait for a raw message that matches the condition """
msg = self._recv()
while not condition(msg):
msg = self._recv()
return msg
class Message(object):
r = re.compile("^:(?P<author>[\w.~]+)(?:!(?P<host>\S+))? PRIVMSG (?P<to>\S+) :(?P<text>.+)")
def __init__(self, raw):
match = re.search(Message.r, raw)
self.author = match.group("author")
self.to = match.group("to")
self.text = match.group("text")
def __str__(self):
return f"<{self.author}> à <{self.to}> : {self.text}"
if __name__ == "__main__":
bot = IRC("irc.planet-casio.com", 6697)
@bot.on(lambda m: "Hello" in m.text)
def hello(msg):
bot.send(msg.to, f"Nice to see you {msg.author}!")
bot.start("glados", "abcdef123456")