IRC logging bot

Question

I've recently been learning Python and decided to write an IRC bot as a good first project. Before now I've only really written scripts.

I am using the Python IRC library.

The IRC bot joins an IRC server and multiple channels, logs the channels and makes announcements by listening for messages on a port.

I've gotten to the point where the bot is stable and would like to get some suggestions for improvements. Please keep in mind that I am pretty new to programming, so if you can link to relevant docs I would appreciate it.

I am unsure of how I am handling the opening/closing of log files and I'm not very happy with the configuration parsing. It just seems rather ugly. I think the TCP listener could be better too, but I just don't know how to improve it.

I have a GitHub repository here.

This is the configuration file for the bot:

[irc]
network = irc.freenode.net
port = 7000
channels = ##test, ##test2
nick = autobot
nickpass = password
name = Bot
ssl = True
[tcp]
host: localhost
port: 47998
[bot]
prefix = !
log_scheme = ./logs/{channel}/%%Y-%%m-{channel}.log

This is the main bot code:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""A full featured python IRC bot"""
import configparser
import socket
import ssl
import time
import datetime
import re
import sys
import select
import irc.bot
import codecs
from threading import Thread, Timer
from plugins.passive import url_announce, LogFile
# Create our bot class
class AutoBot(irc.bot.SingleServerIRCBot):
 """Create the single server irc bot"""
 def __init__(self, nick, name, nickpass, prefix, log_scheme, channels, network, listenhost, listenport, port=6667, usessl=False):
 """Connect to the IRC server"""
 if usessl:
 factory = irc.connection.Factory(wrapper=ssl.wrap_socket)
 else:
 factory = irc.connection.Factory()
 try:
 irc.bot.SingleServerIRCBot.__init__(self, [(network, port)], nick, name, connect_factory = factory)
 except irc.client.ServerConnectionError:
 sys.stderr.write(sys.exc_info()[1])
 self.nick = nick
 self.channel_list = channels
 self.nickpass = nickpass
 self.prefix = prefix
 self.log_scheme = log_scheme
 self.logs = {}
 self.logs['autobot'] = LogFile.LogFile(datetime.datetime.utcnow().strftime(log_scheme).format(channel='autobot'))
 for ch in channels:
 log_name = datetime.datetime.utcnow().strftime(log_scheme).format(channel=ch)
 self.logs[ch] = LogFile.LogFile(log_name)
 self.periodic = Timer(960, self.refresh_logs)
 self.periodic.start()
 self.connection.add_global_handler("quit", self.alt_on_quit, -30)
 self.inputthread = TCPinput(self.connection, self, listenhost, listenport)
 self.inputthread.start()
 def start(self):
 try:
 super().start()
 except:
 self.close_logs()
 raise
 def say(self, target, text):
 """Send message to IRC and log it"""
 self.connection.privmsg(target, text)
 self.log_message(target, "<" + self.nick + ">", text)
 def do(self, target, text):
 self.connection.action(target, text)
 self.log_message(target, "*", self.connection.get_nickname() + " " + text)
 def on_nicknameinuse(self, connection, event):
 """If the nick is in use, get nick_"""
 connection.nick(connection.get_nickname() + "_")
 def on_welcome(self, connection, event):
 """Join channels and regain nick"""
 for channel in self.channel_list:
 connection.join(channel)
 self.log_message("autobot", "-->", "Joined channel %s" % (channel))
 if self.nickpass and connection.get_nickname() != self.nick:
 connection.privmsg("nickserv", "ghost %s %s" % (self.nick, self.nickpass))
 self.log_message("autobot", "-!-", "Recovered nick")
 def get_version(self):
 """CTCP version reply"""
 return "Autobot IRC bot"
 def on_privnotice(self, connection, event):
 """Identify to nickserv and log privnotices"""
 self.log_message("autobot", "<" + event.source + ">", event.arguments[0])
 if not event.source:
 return
 source = event.source.nick
 if source and source.lower() == "nickserv":
 if event.arguments[0].lower().find("identify") >= 0:
 if self.nickpass and self.nick == connection.get_nickname():
 connection.privmsg("nickserv", "identify %s %s" % (self.nick, self.nickpass))
 self.log_message("autobot", "-!-", "Identified to nickserv")
 #def on_disconnect(self, connection, event):
 def on_pubnotice(self, connection, event):
 """Log public notices"""
 self.log_message(event.target, "-!-", "(notice) " + event.source + ": " + event.arguments[0])
 def on_kick(self, connection, event):
 """Log kicked nicks and rejoin channels if bot is kicked"""
 kicked_nick = event.arguments[0]
 kicker = event.source.nick
 self.log_message(event.target, "<--", "%s was kicked from the channel by %s" % (kicked_nick, kicker))
 if kicked_nick == self.nick:
 time.sleep(10) #waits 10 seconds
 for channel in self.channel_list:
 connection.join(channel)
 def alt_on_quit(self, connection, event):
 """Log when users quit"""
 for channel in self.channels:
 if self.channels[channel].has_user(event.source.nick):
 self.log_message(channel, "<--", "%s has quit" % (event.source))
 def on_join(self, connection, event):
 """Log channel joins"""
 self.log_message(event.target, "-->", "%s joined the channel" % (event.source))
 if event.source.nick == self.nick:
 self.say(event.target, "Autobots, roll out!")
 def on_part(self, connection, event):
 """Log channel parts"""
 self.log_message(event.target, "<--", "%s left the channel" % (event.source))
 def on_nick(self, connection, event):
 """Log nick changes"""
 new_nick = event.target
 for channel in self.channels:
 if self.channels[channel].has_user(new_nick):
 self.log_message(channel, "-!-", "%s changed their nick to %s" % (event.source, new_nick))
 def on_mode(self, connection, event):
 """Log mode changes"""
 mode = " ".join(event.arguments)
 self.log_message(event.target, "-!-", "mode changed to %s by %s" % (mode, event.source.nick))
 def on_topic(self, connection, event):
 """Log topic changes"""
 self.log_message(event.target, "-!-", 'topic changed to "%s" by %s' % (event.arguments[0], event.source.nick))
 def on_action(self, connection, event):
 self.log_message(event.target, "*", event.source.nick + " " + event.arguments[0])
 def on_pubmsg(self, connection, event):
 """Log public messages and respond to command requests"""
 channel = event.target
 nick = event.source.nick
 message = event.arguments[0]
 self.log_message(channel, "<" + nick + ">", message)
 url_regex = re.compile(
 r'(?i)\b((?:https?://|[a-z0-9.\-]+[.][a-z]{2,4}/)'
 r'(?:[^\s()<>]+|\(([^\s()<>]+|(\([^\s()<>]+\)))*\))'
 r'+(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|'
 r'''[^\s`!()\[\]{};:'".,<>?«»""‘’]))''', re.IGNORECASE)
 if url_regex.search(message):
 message_list = message.split(' ')
 for element in message_list:
 if url_regex.match(element):
 title = url_announce.parse_url(element)
 if title is not None:
 self.say(channel, title)
 command_regex = re.compile(
 r'^(' + re.escape(self.nick) + '( |[:,] ?)'
 r'|' + re.escape(self.prefix) + ')'
 r'([^ ]*)( (.*))?$', re.IGNORECASE)
 if command_regex.match(message):
 command = command_regex.match(message).group(3)
 arguments = command_regex.match(message).group(5)
 if self.channels[channel].is_oper(nick):
 self.do_command(event, True, channel, command, arguments)
 else:
 self.do_command(event, False, channel, command, arguments)
 def on_privmsg(self, connection, event):
 """Log private messages and respond to command requests"""
 nick = event.source.nick
 message = event.arguments[0]
 self.log_message(nick, "<" + nick + ">", message)
 command = message.partition(' ')[0]
 arguments = message.partition(' ')[2].strip(' ')
 if arguments == '':
 self.do_command(event, False, nick, command, None)
 else:
 self.do_command(event, False, nick, command, arguments)
 def do_command(self, event, isOper, source, command, arguments):
 """Commands the bot will respond to"""
 user = event.source.nick
 connection = self.connection
 if command == "hello":
 self.say(source, "hello " + user)
 elif command == "goodbye":
 self.say(source, "goodbye " + user)
 elif command == "ugm":
 self.say(source, "good (UGT) morning to all from " + user + "!")
 elif command == "ugn":
 self.say(source, "good (UGT) night to all from " + user + "!")
 elif command == "slap":
 if arguments is None or arguments.isspace():
 self.do(source, "slaps " + user + " around a bit with a large trout")
 else:
 self.do(source, "slaps " + arguments.strip(" ") + " around a bit with a large trout")
 elif command == "rot13":
 if arguments is None:
 self.say(source, "I'm sorry, I need a message to cipher, try \"!rot13 message\"")
 else:
 self.say(source, codecs.encode(arguments, 'rot13'))
 elif command == "help":
 self.say(source, "Available commands: ![hello, goodbye, "
 "ugm, ugn, slap, rot13 <message>, "
 "disconnect, die, help]")
 elif command == "disconnect":
 if isOper:
 self.disconnect(msg="I'll be back!")
 else:
 self.say(source, "You don't have permission to do that")
 elif command == "die":
 if isOper:
 self.close_logs()
 self.periodic.cancel()
 self.die(msg="Bye, cruel world!")
 else:
 self.say(source, "You don't have permission to do that")
 else:
 connection.notice(user, "I'm sorry, " + user + ". I'm afraid I can't do that")
 def announce(self, connection, text):
 """Send notice to joined channels"""
 for channel in self.channel_list:
 connection.notice(channel, text)
 self.log_message(channel, "-!-", "(notice) " + connection.get_nickname() + ": " + text)
 def log_message(self, channel, nick, message):
 """Create IRC logs"""
 if channel not in self.logs:
 self.logs[channel] = LogFile.LogFile(datetime.datetime.utcnow().strftime(self.log_scheme).format(channel=channel))
 self.logs[channel].write("{0} {1}".format(nick, message))
 def refresh_logs(self):
 """Remove stale log files (15 min without writes)"""
 timestamp = int(time.time())
 for log in self.logs:
 if self.logs[log].is_stale(timestamp):
 self.logs[log].close()
 def close_logs(self):
 """ Close all open log files"""
 for log in self.logs:
 self.logs[log].close()
class TCPinput(Thread):
 """Listen for data on a port and send it to Autobot.announce"""
 def __init__(self, connection, AutoBot, listenhost, listenport):
 Thread.__init__(self)
 self.setDaemon(1)
 self.AutoBot = AutoBot
 self.listenport = listenport
 self.connection = connection
 self.accept_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 self.accept_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 self.accept_socket.bind((listenhost, listenport))
 self.accept_socket.listen(10)
 self.accept_socket.setblocking(False)
 self.epoll = select.epoll()
 self.epoll.register(self.accept_socket.fileno(), select.EPOLLIN)
 self.stuff = {}
 def run(self):
 while True:
 for sfd, ev in self.epoll.poll():
 if sfd == self.accept_socket.fileno():
 conn, addr = self.accept_socket.accept()
 self.epoll.register(conn.fileno(), select.EPOLLIN)
 self.stuff[conn.fileno()] = conn
 else:
 conn = self.stuff[sfd]
 buf = conn.recv(1024)
 if not buf:
 conn.close()
 continue
 self.AutoBot.announce(self.connection, buf.decode("utf-8", "replace").strip())
def main():
 config = configparser.ConfigParser()
 config.read("autobot.conf")
 network = config.get("irc", "network")
 port = int(config.get("irc", "port"))
 _ssl = config.getboolean("irc", "ssl")
 channels = [channel.strip() for channel in config.get("irc", "channels").split(",")]
 nick = config.get("irc", "nick")
 nickpass = config.get("irc", "nickpass")
 name = config.get("irc", "name")
 listenhost = config.get("tcp", "host")
 listenport = int(config.get("tcp", "port"))
 prefix = config.get("bot", "prefix")
 log_scheme = config.get("bot", "log_scheme")
 bot = AutoBot(nick, name, nickpass, prefix, log_scheme, channels, network, listenhost, listenport, port, _ssl)
 bot.start()
if __name__ == "__main__":
 main()

This is the LogFile.py that is used for making file objects:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Create log file objects"""
import sys
import os
import datetime
import time
class LogFile(object):
 """Handle open/write/close of file with error checking"""
 def __init__(self, path):
 """Create dirs if they don't exist and open file"""
 self.path = path
 self.last_write = 0
 if os.path.exists(path) is False:
 try:
 os.makedirs(os.path.dirname(path), exist_ok=True)
 except OSError as err:
 sys.stderr.write("Error when making log path for {0} - {1}\n".format(path, err))
 self.open()
 def open(self):
 try:
 self.log = open(self.path, 'a')
 sys.stderr.write("opening " + self.path + "\n")
 except PermissionError as err:
 sys.stderr.write("Permission error: " + err + "\n")
 except:
 sys.stderr.write("Error opening log " + self.path + "\n")
 def write(self, message):
 """write to file"""
 if self.log.closed:
 self.open()
 timestamp = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
 try:
 self.log.write("{0} {1}\n".format(timestamp, message))
 self.last_write = int(time.time())
 except:
 sys.stderr.write("Error writting to log " + self.path + "\n")
 def is_stale(self, timestamp):
 if timestamp - self.last_write <= 900:
 return False
 else:
 return True
 def close(self):
 """close file"""
 if not self.log.closed:
 self.log.close()
 sys.stderr.write("Log closed " + self.path + "\n")

This is the URL announcement plugin I made to announce website titles from URLs posted in a channel:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""A plugin for Autobot that announces the title for urls in IRC channels"""
import encodings
from urllib.request import urlopen, Request
from urllib.parse import quote, urlsplit
from urllib.error import URLError
from bs4 import BeautifulSoup
def parse_url(url):
 """Say Website Title information in channel"""
 #if urlopen(url).getcode() == 200:
 baseurl = '{uri.scheme}://{uri.netloc}'.format(uri=urlsplit(url))
 path = urlsplit(url).path
 query = '?{uri.query}'.format(uri=urlsplit(url))
 try:
 parsed_url = baseurl.encode("idna").decode("idna") + quote(path + query, safe='/#:=&?')
 except:
 return
 try:
 request = Request(parsed_url)
 request.add_header('Accept-Encoding', 'utf-8')
 request.add_header('User-Agent', 'Mozilla/5.0')
 response = urlopen(request)
 except:
 return
 try:
 URL = BeautifulSoup(response.read(), "html.parser")
 except URLError as e:
 sys.stderr.write("Error when fetching " + url + ": %s\n" % (e))
 return
 if not URL.title:
 return
 if URL.title.string is None:
 return
 if len(URL.title.string) > 250:
 title=URL.title.string[0:250] + '...'
 else:
 title=URL.title.string
 return title.replace('\n', ' ').strip() + " (" + urlsplit(url).netloc + ")"

Answer 1

I have some thoughts from your main bot code. There's a lot of small improvements that can help clean up or make the code more efficient. Particularly Python odditties that are useful, odd as they seem.

You nest three if statements in on_privnotice, but it'd be better to just have one group if statement where you test all the values. This would make for one very long line but you can break it up. Python will let you split lines up if the conditions are between parentheses and then you only need the one if:

 if (source and source.lower() == "nickserv"
 and event.arguments[0].lower().find("identify") >= 0
 and self.nickpass and self.nick == connection.get_nickname()):
 connection.privmsg("nickserv", "identify %s %s" % (self.nick, self.nickpass))
 self.log_message("autobot", "-!-", "Identified to nickserv")

You have overly long lines in a lot of places. Python's style guide (PEP0008) recommends a maximum of 79 characters. Using any kind of brackets (()[]{}) can split up lines and allow them to be treated as one, like I showed above. Most long lines can be split up using brackets like these.

Adding a blank line in this particular case can also make it easier to separate the if from the conditions, alternatively you could indent the second and third lines further to visually distinguish them.

Also using .find to test for the existence of a string is awkward, and Python has a much nicer way. You can just use "identify" in event.arguments[0].lower(). It's more readable and idiomatic.

Also you should use str.format instead of string concatenation to make a string. It's easier to read and get the formatting right. So instead of:

def on_pubnotice(self, connection, event):
 """Log public notices"""
 self.log_message(event.target, "-!-", "(notice) " + event.source + ": " + event.arguments[0])

do this:

 self.log_message(event.target, "-!-", "(notice) {}: {}".format(event.source, event.arguments[0]))

str.format is useful for many reasons, but primarily because it looks clean and automatically converts to string type as best it can. This is another too long line by the way, but it's easily split since log_message wraps it in parentheses:

 self.log_message(event.target, "-!-", "(notice) {}: {}"
 .format(event.source, event.arguments[0]))

In your on_pubmsg you could shorten the regex searching if you used a list comprehension. List comprehensions basically take a for loop-like expression and are built instantly from it. You can also add conditions, so in your case it will only take elements from message_list that actually match the regex. This is how it could look:

 if url_regex.search(message):
 message_list = [element for element in message.split(' ')
 if url_regex.match(element)]
 for element in message_list:
 title = url_announce.parse_url(element)
 if title is not None:
 self.say(channel, title)

Also it's redundant when you have an if test used to only differentiate a bool like this:

 if self.channels[channel].is_oper(nick):
 self.do_command(event, True, channel, command, arguments)
 else:
 self.do_command(event, False, channel, command, arguments)

You could pass self.channels[channel].is_oper(nick) directly:

 self.do_command(event, self.channels[channel].is_oper(nick),
 channel, command, arguments)

Or just set it as a variable and pass that.

 oper_test = self.channels[channel].is_oper(nick)
 self.do_command(event, oper_test, channel, command, arguments)

You have a similar one later, except you're differentiating between None and a value. You can't pass a value directly but you could use a ternary expression. Ternaries basically allow you to put a simple boolean test into an expression that will return one of two values. It looks like value1 if boolean else value2. If boolean is True then value1 is the result of the expression, otherwise value2 is returned. This could be used to replace your arguments test like so:

 self.do_command(event, False, nick, command,
 None if arguments == '' else arguments)

But that's a little backwards to how it usually works. Most people test for the existence of a value, not its absence. There's a handy way to test. Python has a concept of truthiness, where most types can be directly coerced to boolean values. This means that you can do if arguments, and when arguments is a string that means that an empty string is False while a string containing any characters will be True. This is exactly what you need:

 self.do_command(event, False, nick, command,
 arguments if arguments else None)

do_command could be shortened a lot with a dictionary. You could have a set of keys and values with formattable strings, like this:

commands = {
 "hello": "hello {}",
 "goodbye": "goodbye {}",
 "ugm": "good (UGT) morning to all from {}!",
 "ugn": "good (UGT) night to all from {}!",
 "slap": "slaps {} around a bit with a large trout",
}

And so on. Now you can call on them simply with commands[command].format(user). This will insert the user into the string, or anything else you might need to pass like arguments.strip. You still need to do tests to determine other things like whether to use do instead of say, or if other commands need to be performed, but a dictionary can remove several of your generic commands.

Also strip will remove all whitespace by default, so when you pass " " to it you're actually just preventing it from stripping newline or tab characters. That may be intentional but I doubt it. Just use strip() and there should be no whitespace.

log_message tests for the existence of channel every time it's called and it's called often. This is quite wasteful and unPythonic. Python style is to assume the key is already there and access it. This is known as EAFP, Easier to Ask Forgiveness than Permission. You can just catch the errors when they arise and handle them rather than always testing. This is faster because your test should usually be True, so it only rarely catches problems.

def log_message(self, channel, nick, message):
 """Create IRC logs"""
 try:
 log_file = self.logs[channel]
 except KeyError:
 self.logs[channel] = LogFile.LogFile(datetime.datetime.utcnow().strftime(self.log_scheme).format(channel=channel))
 log_file = self.logs[channel]
 log_file.write("{0} {1}".format(nick, message))

• meskarune: thanks, you have given me a ton of great suggestions and things to keep in mind as I continue using python. I appreciate it! :)

Answer 2

First codereview post, so bear with me. Thinks that I think could be improved.

• Use a proper logging functionality. Use the Logger class from Python, add a file handler to store it into file, and a streamhandler to redirect errors to stderr. https://docs.python.org/2/howto/logging.html
• Don't concatenate your strings yourself, use string format: https://docs.python.org/2/library/functions.html#format
• Clean Code: make your code self explanatory. That means for example that someone else could read your code and doesn't have to think about what it does.
  • Remove comments that repeat what the code does: function say and comment """Send message to IRC and log it"""
  • No clue what function do does
  • In on_nicknameinuse you set another name. Why not split this up into a seperate function (choose_new_nickname)? More clear for the reader.
  • Same goes for on_welcome. Put the regain nick functionality in a different function.
  • Same for on_kick, put the code for rejoining in seperate function
  • I would put the logic of do_command into seperate functions, create a map of the command to the function called.

Code is OK, but I always focus on readability improvements. If someone else reads this code, does he understand without thinking to much. Good book on this topic is Clean Code.

• holroy: Welcome to Code Review SE, all in all a decent review, but why the image link? It looks like spam?
• RvdK: Must read for any software developer. I've added it because I look more at style.
• holroy: There are loads of good books out there, but it is not common to link to books or include images like this advertising. It's not good style when answering (pun intended :-))
• meskarune: Thanks so much. I will add some comments and sure it is clear what is happening. on_nicknameinuse and on_welcome are events in the python irc library. Basically I am using the library modules, so that's where the names come from. For the commands I am wanting to switch entirely over to a plugin system instead of having it hardcoded into the main bot. That section of code is really meant to be temporary until I can figure out a plugin system. I will check out the book clean code, thanks for the recomendation!