Fuse into irc.py again
This commit is contained in:
Dpeta
parent
f3cacb849b
commit
228ced97ef
2 changed files with 137 additions and 162 deletions
159
irc.py
159
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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
Loading…
Reference in a new issue