a (very) simple irc client ...

irc.py

@@ -0,0 +1,56 @@
+from simplesocket import ClientSocket
+from typing import Callable
+
+
+class Client:
+    def __init__(self, server: str, port: int, nick: str, username: str, realname: str = "Python Bot"):
+        self.nickname = nick
+        self.__server = ClientSocket(server, port)
+        self.__server.sendline("USER %s ignore ignore :%s" % (username, realname))
+        self.__server.sendline("NICK %s" % self.nickname)
+        self.__function_register = {
+            'NICK': self.on_nick
+        }
+        self.receive()
+
+    def receive(self):
+        while line := self.__server.recvline():
+            print(line)
+            if line.startswith("PING"):
+                self.__server.sendline("PONG " + line.split()[1])
+                continue
+
+            (msg_from, msg_type, msg_to, *msg) = line[1:].split()
+            if msg_type == "433":
+                self.nickname = msg[0] + "_"
+                self.__server.sendline("NICK %s" % self.nickname)
+            else:
+                if msg_type in self.__function_register:
+                    self.__function_register[msg_type](msg_from, msg_to, " ".join(msg)[1:])
+
+    def register(self, msg_type: str, func: Callable[..., None]):
+        self.__function_register[msg_type] = func
+
+    def on_nick(self, old_nick: str, new_nick: str, _):
+        old_nick = old_nick.split("!")[0]
+        if old_nick == self.nickname:
+            self.nickname = new_nick[1:]
+
+    def nick(self, new_nick: str):
+        self.__server.sendline("NICK %s" % new_nick)
+
+    def join(self, channel: str):
+        self.__server.sendline("JOIN %s" % channel)
+        self.receive()
+
+    def leave(self, channel: str):
+        self.__server.sendline("LEAVE %s" % channel)
+        self.receive()
+
+    def privmsg(self, target: str, message: str):
+        self.__server.sendline("PRIVMSG %s :%s" % (target, message))
+
+    def quit(self, message: str = "Elvis has left the building!"):
+        self.__server.sendline("QUIT :%s" % message)
+        self.receive()
+        self.__server.close()

simplesocket.py

@@ -10,7 +10,7 @@ class Socket:
         self.socket = socket.socket(family=address_family, type=socket_kind)
         self.__recv_buffer = b""
 
-    def send(self, buffer: Union[str, bytes]):
+    def send(self, buffer: Union[str, bytes]) -> int:
         if isinstance(buffer, str):
             buffer = buffer.encode("UTF-8")
 
@@ -23,11 +23,8 @@ class Socket:
     def recv(self, maxlen: int = 4096, blocking: bool = True) -> bytes:
         maxlen -= len(self.__recv_buffer)
         try:
-            if blocking:
+            self.socket.setblocking(blocking)
             ret = self.__recv_buffer + self.socket.recv(maxlen)
-            else:
-                ret = self.__recv_buffer + self.socket.recv(maxlen, socket.MSG_DONTWAIT)
-
             self.__recv_buffer = b""
             return ret
         except socket.error as e:
@@ -52,15 +49,22 @@ class Socket:
             else:
                 break
 
+        newline = b"\r\n"
+        if newline not in self.__recv_buffer:
             newline = b"\n"
             if newline not in self.__recv_buffer:
                 newline = b"\r"
                 if newline not in self.__recv_buffer:
                     ret = self.__recv_buffer.decode("UTF-8")
                     self.__recv_buffer = b""
+                    return ret
 
         ret = self.__recv_buffer[:self.__recv_buffer.index(newline)].decode("UTF-8")
-        self.__recv_buffer = self.__recv_buffer[self.__recv_buffer.index(newline) + 1:]
+        if len(self.__recv_buffer) - len(newline) > self.__recv_buffer.index(newline):
+            self.__recv_buffer = self.__recv_buffer[self.__recv_buffer.index(newline) + len(newline):]
+        else:
+            self.__recv_buffer = b""
+
         return ret
 
     def close(self):