Code cleanup and refactoring
Created send_privmsg to reduce repetition and verbosity of code Refactored many recently-written command functions to further reduce repetition.
This commit is contained in:
parent
d67f7d93ec
commit
d4c88f6025
151
bot.py
151
bot.py
|
|
@ -1,4 +1,3 @@
|
|||
# TODO: Clean up message sending by consolidating all those PRIVMSG bits
|
||||
# TODO: Make separate adminhelp command for admin-only commands
|
||||
|
||||
import sys
|
||||
|
|
@ -134,43 +133,47 @@ class Bot:
|
|||
return False
|
||||
|
||||
def send_message(self, message):
|
||||
message = message + "\n"
|
||||
self.ircsock.sendall(message.encode('utf-8'))
|
||||
print("\033[31m<<\033[0m " + message, end='', flush=True)
|
||||
|
||||
def send_privmsg(self, channel, message):
|
||||
self.send_message("PRIVMSG " + channel + " :" + message)
|
||||
|
||||
def announce(self, message):
|
||||
for channel in self.channels:
|
||||
self.send_message(message)
|
||||
self.send_privmsg(channel, message)
|
||||
|
||||
def handle_exception(self, incoming, exception):
|
||||
self.exception = exception
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :An unexpected error occurred. Run +showexception to see what went wrong.\n")
|
||||
self.send_privmsg(self.get_channel(incoming), "An unexpected error occurred. Run +showexception to see what went wrong.")
|
||||
|
||||
def repeat(self, channel, message):
|
||||
self.send_message("PRIVMSG " + channel + " :" + message + "\n")
|
||||
self.send_privmsg(channel, message)
|
||||
|
||||
def connect(self):
|
||||
sys.stderr.write("Connecting to: " + self.server + ":" + str(self.port) + "\n")
|
||||
sys.stderr.write("Connecting to: " + self.server + ":" + str(self.port))
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.connect((self.server, self.port))
|
||||
self.ircsock = ssl.wrap_socket(s)
|
||||
self.send_message("USER " + self.nick + " 0 * " + self.nick + "\n")
|
||||
self.send_message("NICK "+ self.nick + "\n")
|
||||
self.send_message("USER " + self.nick + " 0 * " + self.nick)
|
||||
self.send_message("NICK "+ self.nick)
|
||||
self.sasl_auth() or self.nickserv_auth()
|
||||
for channel in self.channels:
|
||||
self.send_message("JOIN "+ channel +"\n")
|
||||
self.send_message("JOIN "+ channel)
|
||||
|
||||
def sasl_auth(self):
|
||||
self.send_message("CAP REQ :sasl\n")
|
||||
self.send_message("CAP REQ :sasl")
|
||||
incoming = self.waitfor("CAP")
|
||||
if "ACK :sasl" in incoming:
|
||||
self.send_message("AUTHENTICATE PLAIN\n")
|
||||
self.send_message("AUTHENTICATE PLAIN")
|
||||
self.waitfor("+")
|
||||
creds = self.nick + "\0" + self.nick + "\0" + self.password
|
||||
creds = base64.b64encode(creds.encode('utf8')).decode('utf8')
|
||||
self.send_message("AUTHENTICATE " + creds + "\n")
|
||||
self.send_message("AUTHENTICATE " + creds)
|
||||
incoming = self.waitfor(":SASL authentication")
|
||||
if ("903" in incoming):
|
||||
self.send_message("CAP END\n")
|
||||
self.send_message("CAP END")
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
|
@ -178,7 +181,7 @@ class Bot:
|
|||
return False
|
||||
|
||||
def nickserv_auth(self):
|
||||
self.send_message("PRIVMSG NickServ :" + self.password + "\n")
|
||||
self.send_privmsg("NickServ", "IDENTIFY " + self.password)
|
||||
incoming = self.waitfor("NickServ")
|
||||
if ("You are now identified"):
|
||||
return True
|
||||
|
|
@ -186,129 +189,135 @@ class Bot:
|
|||
return False
|
||||
|
||||
def disconnect(self):
|
||||
self.send_message("QUIT :Quitting...\n")
|
||||
self.send_message("QUIT :Quitting...")
|
||||
self.ircsock.close()
|
||||
|
||||
def quit (self, incoming):
|
||||
if self.is_admin(get_user(incoming)):
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :*sigh* alright then...\n")
|
||||
self.send_privmsg(self.get_channel(incoming), "*sigh* alright then...")
|
||||
self.disconnect()
|
||||
self.save_config()
|
||||
return True
|
||||
else:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :You don't have permission to use this command.\n")
|
||||
self.send_privmsg(self.get_channel(incoming), "You don't have permission to use this command.")
|
||||
return False
|
||||
|
||||
def command_help(self, incoming):
|
||||
try:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + self.commandList[incoming.split(' ')[4]] + "\n")
|
||||
self.send_privmsg(self.get_channel(incoming), self.commandList[incoming.split(' ')[4]])
|
||||
except:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Available commands: " + " ".join(("+" + x) for x in self.commandList.keys()) + "\n")
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :+help <command> for more information" + "\n")
|
||||
self.send_privmsg(self.get_channel(incoming), "Available commands: " + " ".join(("+" + x) for x in self.commandList.keys()))
|
||||
self.send_privmsg(self.get_channel(incoming), "+help <command> for more information")
|
||||
|
||||
def command_version(self, incoming):
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + self.nick + "version " + self.VERSION + " by " + self.AUTHOR + ".\n")
|
||||
self.send_privmsg(self.get_channel(incoming), self.nick + "version " + self.VERSION + " by " + self.AUTHOR)
|
||||
|
||||
def command_quote(self, incoming):
|
||||
with open(self.quotesfile, "r") as f:
|
||||
lines = f.read().splitlines()
|
||||
selectedLine = random.choice(lines)
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + selectedLine + "\n")
|
||||
self.send_privmsg(self.get_channel(incoming), selectedLine)
|
||||
|
||||
def command_time(self, incoming):
|
||||
try:
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :" + datetime.now(timezone(incoming.split(' ')[4])).strftime('%Y-%m-%d %H:%M:%S') + "\n"
|
||||
message = datetime.now(timezone(incoming.split(' ')[4])).strftime('%Y-%m-%d %H:%M:%S')
|
||||
except:
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :Please specify a valid timezone (e.g. +time America/Toronto).\n"
|
||||
self.send_message(message)
|
||||
message = "Please specify a valid timezone (e.g. +time America/Toronto)."
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_roll(self, incoming):
|
||||
try:
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :" + str(random.randint(1, int(incoming.split(' ')[4]))) + "\n"
|
||||
message = str(random.randint(1, int(incoming.split(' ')[4])))
|
||||
except:
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :Please specify a maximum number (e.g. '+roll 20').\n"
|
||||
self.send_message(message)
|
||||
message = "Please specify a maximum number (e.g. '+roll 20')."
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_weather(self, incoming):
|
||||
try:
|
||||
city = incoming.split(' ')[4]
|
||||
url = "https://wttr.in/" + city + "?m&format=3"
|
||||
response = requests.get(url)
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :" + response.text.strip('\n\r') + "\n"
|
||||
message = response.text.strip('\n\r')
|
||||
except:
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :Please specify a city (e.g. '+weather Toronto').\n"
|
||||
self.send_message(message)
|
||||
message = "Please specify a city (e.g. '+weather Toronto')."
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_eightball(self, incoming):
|
||||
try:
|
||||
incoming.split(' ')[4]
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :" + random.choice(self.eightBallResponses) + "\n"
|
||||
except Exception as e:
|
||||
print(e)
|
||||
message = "PRIVMSG " + self.get_channel(incoming) + " :Please ask me a question.\n"
|
||||
self.send_message(message)
|
||||
message = random.choice(self.eightBallResponses)
|
||||
except:
|
||||
message = "Please ask me a question."
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_repeatsetup(self, incoming):
|
||||
if not self.is_admin(incoming):
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :You don't have permission to use this command.\n")
|
||||
return
|
||||
try:
|
||||
channel = self.get_channel(incoming)
|
||||
interval = int(incoming.split(' ')[4])
|
||||
message = ' '.join(incoming.split(' ')[5:])
|
||||
except Exception as e:
|
||||
print(e)
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Invalid syntax. Expected +repeat <seconds> <message>\n")
|
||||
rt = RepeatedTimer(interval, self.repeat, channel, message)
|
||||
message = "You don't have permission to use this command."
|
||||
else:
|
||||
try:
|
||||
channel = self.get_channel(incoming)
|
||||
interval = int(incoming.split(' ')[4])
|
||||
message = ' '.join(incoming.split(' ')[5:])
|
||||
rt = RepeatedTimer(interval, self.repeat, channel, message)
|
||||
message = "Repeating message every " + interval + " seconds."
|
||||
except:
|
||||
message = "Invalid syntax. Expected +repeat <seconds> <message>"
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_admins(self, incoming):
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Admins: " + ', '.join(self.admins) + "\n")
|
||||
self.send_privmsg(self.get_channel(incoming), "Admins: " + ', '.join(self.admins))
|
||||
|
||||
def command_addadmin(self, incoming):
|
||||
try:
|
||||
candidate = incoming.split(' ')[4]
|
||||
if self.is_admin(self.get_user(incoming)):
|
||||
candidate = incoming.split(' ')[4]
|
||||
if self.is_admin(candidate):
|
||||
message = candidate + " is already an admin."
|
||||
else:
|
||||
self.admins.append(candidate)
|
||||
message = candidate + " granted admin privileges."
|
||||
else:
|
||||
message = "You don't have permission to use this command."
|
||||
except:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Invalid syntax. Expected +addadmin <nick>\n")
|
||||
if self.is_admin(self.get_user(incoming)):
|
||||
if self.is_admin(candidate):
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + candidate + " is already an admin.\n")
|
||||
self.admins.append(candidate)
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + candidate + " granted admin privileges.\n")
|
||||
else:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :You don't have permission to use this command.\n")
|
||||
message = "Invalid syntax. Expected +addadmin <nick>"
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_rmadmin(self, incoming):
|
||||
try:
|
||||
candidate = incoming.split(' ')[4]
|
||||
if self.is_admin(self.get_user(incoming)):
|
||||
candidate = incoming.split(' ')[4]
|
||||
if not self.is_admin(candidate):
|
||||
message = candidate + " is not an admin."
|
||||
else:
|
||||
self.admins.remove(candidate)
|
||||
message = "Revoked " + candidate + "'s admin privileges."
|
||||
else:
|
||||
message = "You don't have permission to use this command."
|
||||
except:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Invalid syntax. Expected +rmadmin <nick>\n")
|
||||
if self.is_admin(self.get_user(incoming)):
|
||||
if not self.is_admin(candidate):
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + candidate + " is not an admin.\n")
|
||||
self.admins.remove(candidate)
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " : Revoked " + candidate + "'s admin privileges.\n")
|
||||
else:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :You don't have permission to use this command.\n")
|
||||
message = "Invalid syntax. Expected +rmadmin <nick>"
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_showexception(self, incoming):
|
||||
if self.is_admin(self.get_user(incoming)):
|
||||
if not self.exception:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " : No exceptions occurred... yet.\n")
|
||||
message = "No exceptions occurred... yet."
|
||||
else:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :" + str(self.exception) + "\n")
|
||||
message = str(self.exception)
|
||||
else:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :You don't have permission to use this command.\n")
|
||||
message = "You don't have permission to use this command."
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
||||
def command_chgpass(self, incoming):
|
||||
try:
|
||||
password = incoming.split(' ')[4]
|
||||
if self.is_admin(self.get_user(incoming)):
|
||||
self.send_message("PRIVMSG NickServ :SET PASSWORD " + password + "\n")
|
||||
self.send_privmsg("NickServ", "SET PASSWORD " + password)
|
||||
response = self.waitfor("NickServ")
|
||||
if "successfully" in response:
|
||||
self.password = password
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Password changed.\n")
|
||||
message = + "Password changed."
|
||||
else:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Failed to change password.\n")
|
||||
message = + "Failed to change password."
|
||||
except:
|
||||
self.send_message("PRIVMSG " + self.get_channel(incoming) + " :Invalid syntax. Expected +chgpass <new_password>\n")
|
||||
message = "Invalid syntax. Expected +chgpass <new_password>"
|
||||
self.send_privmsg(self.get_channel(incoming), message)
|
||||
|
|
|
|||
3
run.py
3
run.py
|
|
@ -24,7 +24,7 @@ def main():
|
|||
# Respond to pings
|
||||
if "PING " in incoming:
|
||||
print("Responding to PING.")
|
||||
message = "PONG :" + bot.nick + "\n"
|
||||
message = "PONG :" + bot.nick
|
||||
bot.send_message(message)
|
||||
continue
|
||||
|
||||
|
|
@ -66,6 +66,7 @@ def main():
|
|||
except Exception as e:
|
||||
# Don't die when something goes wrong
|
||||
bot.handle_exception(incoming, e)
|
||||
print(e)
|
||||
continue
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue