from simplesocket import ClientSocket from typing import Callable from enum import Enum class ServerReply(str, Enum): RPL_WELCOME = "001" RPL_YOURHOST = "002" RPL_CREATED = "003" RPL_MYINFO = "004" RPL_BOUNCE = "005" RPL_TRACELINK = "200" RPL_TRACECONNECTING = "201" RPL_TRACEHANDSHAKE = "202" RPL_TRACEUNKNOWN = "203" RPL_TRACEOPERATOR = "204" RPL_TRACEUSER = "205" RPL_TRACESERVER = "206" RPL_TRACENEWTYPE = "208" RPL_TRACECLASS = "209" RPL_TRACERECONNECT = "210" RPL_STATSLINKINFO = "211" RPL_STATSCOMMANDS = "212" RPL_STATSCLINE = "213" RPL_STATSNLINE = "214" RPL_STATSILINE = "215" RPL_STATSKLINE = "216" RPL_STATSYLINE = "218" RPL_ENDOFSTATS = "219" RPL_UMODEIS = "221" RPL_SERVLIST = "234" RPL_SERVLISTEND = "235" RPL_STATSLLINE = "241" RPL_STATSUPTIME = "242" RPL_STATSOLINE = "243" RPL_STATSHLINE = "244" RPL_LUSERCLIENT = "251" RPL_LUSEROP = "252" RPL_LUSERUNKNOWN = "253" RPL_LUSERCHANNELS = "254" RPL_LUSERME = "255" RPL_ADMINME = "256" RPL_ADMINLOC1 = "257" RPL_ADMINLOC2 = "258" RPL_ADMINEMAIL = "259" RPL_TRACELOG = "261" RPL_TRACEEND = "262" RPL_TRYAGAIN = "263" RPL_NONE = "300" RPL_AWAY = "301" RPL_USERHOST = "302" RPL_ISON = "303" RPL_UNAWAY = "305" RPL_NOWAWAY = "306" RPL_WHOISUSER = "311" RPL_WHOISSERVER = "312" RPL_WHOISOPERATOR = "313" RPL_WHOWASUSER = "314" RPL_ENDOFWHO = "315" RPL_WHOISIDLE = "317" RPL_ENDOFWHOIS = "318" RPL_WHOISCHANNELS = "319" RPL_LISTSTART = "321" RPL_LIST = "322" RPL_LISTEND = "323" RPL_CHANNELMODEIS = "324" RPL_UNIQOPIS = "325" RPL_NOTOPIC = "331" RPL_TOPIC = "332" RPL_INVITING = "341" RPL_SUMMONING = "342" RPL_INVITELIST = "346" RPL_ENDOFINVITELIST = "347" RPL_EXCEPTLIST = "348" RPL_ENDOFEXCEPTLIST = "349" RPL_VERSION = "351" RPL_WHOREPLY = "352" RPL_NAMEREPLY = "353" RPL_LINKS = "364" RPL_ENDOFLINKS = "365" RPL_ENDOFNAMES = "366" RPL_BANLIST = "367" RPL_ENDOFBANLIST = "368" RPL_ENDOFWHOWAS = "369" RPL_INFO = "371" RPL_MOTD = "372" RPL_ENDOFINFO = "374" RPL_MOTDSTART = "375" RPL_ENDOFMOTD = "376" RPL_YOUREOPER = "381" RPL_REHASHING = "382" RPL_YOURESERVICE = "383" RPL_TIME = "391" RPL_USERSTART = "392" RPL_USERS = "393" RPL_ENDOFUSERS = "394" RPL_NOUSERS = "395" ERR_NOSUCHNICK = "401" ERR_NOSUCHSERVER = "402" ERR_NOSUCHCHANNEL = "403" ERR_CANNOTSENDTOCHAN = "404" ERR_TOOMANYCHANNELS = "405" ERR_WASNOSUCHNICK = "406" ERR_TOOMANYTARGETS = "407" ERR_NOSUCHSERVICE = "408" ERR_NOORIGIN = "409" ERR_NORECIPIENT = "411" ERR_NOTEXTTOSEND = "412" ERR_NOTOPLEVEL = "413" ERR_WILDTOPLEVEL = "414" ERR_BANMASK = "415" ERR_UNKNOWNCOMMAND = "421" ERR_NOMOTD = "422" ERR_NOADMININFO = "423" ERR_FILEERROR = "424" ERR_NONICKNAMEGIVEN = "431" ERR_ERRONEUSNICKNAME = "432" ERR_NICKNAMEINUSE = "433" ERR_NICKCOLLISION = "436" ERR_UNAVAILRESOURCE = "437" ERR_USERNOTINCHANNEL = "441" ERR_NOTOONCHANNEL = "442" ERR_USERONCHANNEL = "443" ERR_NOLOGIN = "444" ERR_SUMMONDISABLED = "445" ERR_USERSDISABLED = "446" ERR_NOTREGISTERED = "451" ERR_NEEDMOREPARAMS = "461" ERR_ALREADYREGISTERED = "462" ERR_NOPERMFORHOST = "463" ERR_PASSWDMISMATH = "464" ERR_YOUREBANNEDCREEP = "465" ERR_YOUWILLBEBANNED = "466" ERR_KEYSET = "467" ERR_CHANNELISFULL = "471" ERR_UNKNOWNMODE = "472" ERR_INVITEONLYCHAN = "473" ERR_BANNEDFROMCHAN = "474" ERR_BADCHANNELKEY = "475" ERR_BADCHANMASK = "476" ERR_BASCHANMODES = "477" ERR_BANLISTFULL = "478" ERR_NOPRIVILEGES = "481" ERR_CHANOPRIVSNEEDED = "482" ERR_CANTKILLSERVER = "483" ERR_RESTRICTED = "484" ERR_UNIQOPPRIVSNEEDED = "485" ERR_NOOPERHOST = "491" ERR_UMODEUNKNOWNFLAG = "501" ERR_USERSDONTMATCH = "502" class Client: def __init__(self, server: str, port: int, nick: str, username: str, realname: str = "Python Bot"): self.nickname = None self.__server = ClientSocket(server, port) self.__server.sendline("USER %s ignore ignore :%s" % (username, realname)) self.__server.sendline("NICK %s" % nick) self.__function_register = { ServerReply.RPL_WELCOME: self.on_welcome, ServerReply.ERR_NICKNAMEINUSE: self.on_nickname_in_use, 'NICK': self.on_nick, '__default__': self.unhandled_server_message, } self.receive() def receive(self): while line := self.__server.recvline(): if line.startswith("PING"): self.__server.sendline("PONG " + line.split()[1]) continue (msg_from, msg_type, msg_to, *msg) = line[1:].split() if len(msg) > 0 and msg[0].startswith(":"): msg[0] = msg[0][1:] message = " ".join(msg) if msg_type in self.__function_register: self.__function_register[msg_type](msg_from, msg_to, message) else: self.__function_register['__default__'](msg_from, msg_type, msg_to, message) def register(self, msg_type: str, func: Callable[..., None]): self.__function_register[msg_type] = func def on_welcome(self, msg_from: str, msg_to: str, message: str): self.nickname = msg_to def on_nickname_in_use(self, msg_from: str, msg_to: str, message: str): if self.nickname is None: self.nick(message.split()[0] + "_") def on_nick(self, old_nick: str, new_nick: str, message: str): old_nick = old_nick.split("!")[0] if old_nick == self.nickname: self.nickname = new_nick[1:] def unhandled_server_message(self, msg_from: str, msg_type: str, msg_to: str, message: str): print(msg_from, msg_type, msg_to, message) 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()