diff --git a/irc.py b/irc.py index c4039d1..456d86e 100644 --- a/irc.py +++ b/irc.py @@ -19,8 +19,6 @@ from dataobjs import PesterProfile from generic import PesterList from version import _pcVersion -import scripts.irc.outgoing - PchumLog = logging.getLogger("pchumLogger") SERVICES = [ "nickserv", @@ -89,7 +87,7 @@ class PesterIRC(QtCore.QThread): self.ssl = self.config.ssl() self._end = False - self.send_irc = scripts.irc.outgoing.SendIRC() + self.send_irc = SendIRC() self.conn = None self.joined = False @@ -247,25 +245,6 @@ class PesterIRC(QtCore.QThread): self.socket.close() yield False - def parse_buffer(self, buffer): - """Parse lines from bytes buffer, returns lines and emptied buffer.""" - try: - decoded_buffer = buffer.decode(encoding="utf-8") - except UnicodeDecodeError as exception: - PchumLog.warning(f"Failed to decode with utf-8, falling back to latin-1.") - try: - decoded_buffer = buffer.decode(encoding="latin-1") - except ValueError as exception: - PchumLog.warning("latin-1 failed too xd") - return "", buffer # throw it back in the cooker - - data = decoded_buffer.split("\r\n") - if data[-1]: - # Last entry has incomplete data, add back to buffer - print(f"data[-1]: {data[-1]}") - buffer = data[-1].encode(encoding="utf-8") - return data[:-1], buffer - def parse_irc_line(self, line: str): parts = line.split(" ") tags = None @@ -478,6 +457,7 @@ class PesterIRC(QtCore.QThread): QString, QString, ) + def sendCTCP(self, handle, text): self.send_irc.ctcp(handle, text) @@ -1106,3 +1086,138 @@ class PesterIRC(QtCore.QThread): tooManyPeeps = QtCore.pyqtSignal() quirkDisable = QtCore.pyqtSignal("QString", "QString", "QString") signal_forbiddenchannel = QtCore.pyqtSignal("QString", "QString") + +class SendIRC: + """Provides functions for outgoing IRC commands.""" + + def __init__(self): + self.socket = None # INET socket connected with server. + + def send(self, *args: str, text=None): + """Send a command to the IRC server. + + Takes either a string or a list of strings. + The 'text' argument is for the final parameter, which can have spaces.""" + # Return if disconnected + if not self.socket or self.socket.fileno() == -1: + PchumLog.error( + "Send attempted while disconnected, args: %s, text: %s.", args, text + ) + return + + command = "" + # Convert command arguments to a single string if passed. + if args: + command += " ".join(args) + # If text is passed, add ':' to imply everything after it is one parameter. + if text: + command += f" :{text}" + # Add characters for end of line in IRC. + command += "\r\n" + # UTF-8 is the prefered encoding in 2023. + outgoing_bytes = command.encode(encoding="utf-8", errors="replace") + + try: + PchumLog.debug("Sending: %s", command) + self.socket.send(outgoing_bytes) + except OSError: + PchumLog.exception("Error while sending: '%s'", command.strip()) + self.socket.close() + + def ping(self, token): + """Send PING command to server to check for connectivity.""" + self.send("PING", text=token) + + def pong(self, token): + """Send PONG command to reply to server PING.""" + self.send("PONG", token) + + def nick(self, nick): + """Send USER command to communicate nick to server.""" + self.send("NICK", nick) + + def user(self, username, realname): + """Send USER command to communicate username and realname to server.""" + self.send("USER", username, "0", "*", text=realname) + + def privmsg(self, target, text): + """Send PRIVMSG command to send a message.""" + for line in text.split("\n"): + self.send("PRIVMSG", target, text=line) + + def names(self, channel): + """Send NAMES command to view channel members.""" + self.send("NAMES", channel) + + def kick(self, channel, user, reason=""): + """Send KICK command to force user from channel.""" + if reason: + self.send(f"KICK {channel} {user}", text=reason) + else: + self.send(f"KICK {channel} {user}") + + def mode(self, target, modestring="", mode_arguments=""): + """Set or remove modes from target.""" + outgoing_mode = " ".join([target, modestring, mode_arguments]).strip() + self.send("MODE", outgoing_mode) + + def ctcp(self, target, command, msg=""): + """Send Client-to-Client Protocol message.""" + outgoing_ctcp = " ".join( + [command, msg] + ).strip() # Extra spaces break protocol, so strip. + self.privmsg(target, f"\x01{outgoing_ctcp}\x01") + + def metadata(self, target, subcommand, *params): + """Send Metadata command to get or set metadata. + + See IRC metadata draft specification: + https://gist.github.com/k4bek4be/92c2937cefd49990fbebd001faf2b237 + """ + self.send("METADATA", target, subcommand, *params) + + def cap(self, subcommand, *params): + """Send IRCv3 CAP command for capability negotiation. + + See: https://ircv3.net/specs/extensions/capability-negotiation.html""" + self.send("CAP", subcommand, *params) + + def join(self, channel, key=""): + """Send JOIN command to join a channel/memo. + + Keys or joining multiple channels is possible in the specification, but unused. + """ + channel_and_key = " ".join([channel, key]).strip() + self.send("JOIN", channel_and_key) + + def part(self, channel): + """Send PART command to leave a channel/memo. + + Providing a reason or leaving multiple channels is possible in the specification. + """ + self.send("PART", channel) + + def notice(self, target, text): + """Send a NOTICE to a user or channel.""" + self.send("NOTICE", target, text=text) + + def invite(self, nick, channel): + """Send INVITE command to invite a user to a channel.""" + self.send("INVITE", nick, channel) + + def away(self, text=None): + """AWAY command to mark client as away or no longer away. + + No 'text' parameter means the client is no longer away.""" + if text: + self.send("AWAY", text=text) + else: + self.send("AWAY") + + def list(self): + """Send LIST command to get list of channels.""" + self.send("LIST") + + def quit(self, reason=""): + """Send QUIT to terminate connection.""" + self.send("QUIT", text=reason) diff --git a/scripts/irc/outgoing.py b/scripts/irc/outgoing.py deleted file mode 100644 index 93eb75f..0000000 --- a/scripts/irc/outgoing.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Class and functions for sending outgoing IRC commands.""" -import logging - -log = logging.getLogger("pchumLogger") - - -class SendIRC: - """Provides functions for outgoing IRC commands.""" - - def __init__(self): - self.socket = None # INET socket connected with server. - - def send(self, *args: str, text=None): - """Send a command to the IRC server. - - Takes either a string or a list of strings. - The 'text' argument is for the final parameter, which can have spaces.""" - # Return if disconnected - if not self.socket or self.socket.fileno() == -1: - log.error( - "Send attempted while disconnected, args: %s, text: %s.", args, text - ) - return - - command = "" - # Convert command arguments to a single string if passed. - if args: - command += " ".join(args) - # If text is passed, add ':' to imply everything after it is one parameter. - if text: - command += f" :{text}" - # Add characters for end of line in IRC. - command += "\r\n" - # UTF-8 is the prefered encoding in 2023. - outgoing_bytes = command.encode(encoding="utf-8", errors="replace") - - try: - log.debug("Sending: %s", command) - self.socket.send(outgoing_bytes) - except OSError: - log.exception("Error while sending: '%s'", command.strip()) - self.socket.close() - - def ping(self, token): - """Send PING command to server to check for connectivity.""" - self.send("PING", text=token) - - def pong(self, token): - """Send PONG command to reply to server PING.""" - self.send("PONG", token) - - def nick(self, nick): - """Send USER command to communicate nick to server.""" - self.send("NICK", nick) - - def user(self, username, realname): - """Send USER command to communicate username and realname to server.""" - self.send("USER", username, "0", "*", text=realname) - - def privmsg(self, target, text): - """Send PRIVMSG command to send a message.""" - for line in text.split("\n"): - self.send("PRIVMSG", target, text=line) - - def names(self, channel): - """Send NAMES command to view channel members.""" - self.send("NAMES", channel) - - def kick(self, channel, user, reason=""): - """Send KICK command to force user from channel.""" - if reason: - self.send(f"KICK {channel} {user}", text=reason) - else: - self.send(f"KICK {channel} {user}") - - def mode(self, target, modestring="", mode_arguments=""): - """Set or remove modes from target.""" - outgoing_mode = " ".join([target, modestring, mode_arguments]).strip() - self.send("MODE", outgoing_mode) - - def ctcp(self, target, command, msg=""): - """Send Client-to-Client Protocol message.""" - outgoing_ctcp = " ".join( - [command, msg] - ).strip() # Extra spaces break protocol, so strip. - self.privmsg(target, f"\x01{outgoing_ctcp}\x01") - - def metadata(self, target, subcommand, *params): - """Send Metadata command to get or set metadata. - - See IRC metadata draft specification: - https://gist.github.com/k4bek4be/92c2937cefd49990fbebd001faf2b237 - """ - self.send("METADATA", target, subcommand, *params) - - def cap(self, subcommand, *params): - """Send IRCv3 CAP command for capability negotiation. - - See: https://ircv3.net/specs/extensions/capability-negotiation.html""" - self.send("CAP", subcommand, *params) - - def join(self, channel, key=""): - """Send JOIN command to join a channel/memo. - - Keys or joining multiple channels is possible in the specification, but unused. - """ - channel_and_key = " ".join([channel, key]).strip() - self.send("JOIN", channel_and_key) - - def part(self, channel): - """Send PART command to leave a channel/memo. - - Providing a reason or leaving multiple channels is possible in the specification. - """ - self.send("PART", channel) - - def notice(self, target, text): - """Send a NOTICE to a user or channel.""" - self.send("NOTICE", target, text=text) - - def invite(self, nick, channel): - """Send INVITE command to invite a user to a channel.""" - self.send("INVITE", nick, channel) - - def away(self, text=None): - """AWAY command to mark client as away or no longer away. - - No 'text' parameter means the client is no longer away.""" - if text: - self.send("AWAY", text=text) - else: - self.send("AWAY") - - def list(self): - """Send LIST command to get list of channels.""" - self.send("LIST") - - def quit(self, reason=""): - """Send QUIT to terminate connection.""" - self.send("QUIT", text=reason)