IRC bot to keep track of descriptions of words and their karma.
import sys | |
import random | |
import re | |
import time | |
from twisted.words.protocols import irc | |
from twisted.internet import reactor, protocol, ssl | |
from twisted.python import log | |
try: | |
import json | |
except ImportError: | |
import simplejson as json | |
def created_timestamp(who, where): | |
return {"who":who, "when":time.time(), "where":where} | |
class KarmaData: | |
def __init__(self, filename): | |
self.filename = filename | |
self.data = None | |
def load(self): | |
try: | |
self.data = json.load(open(self.filename)) | |
except IOError: | |
self.data = {"things":{}} | |
def save(self): | |
json.dump(self.data, open(self.filename, "w"), sort_keys=True, indent=4) | |
@property | |
def things(self): | |
return self.data["things"] | |
def get_thing(self, name, who=None, where=None): | |
name = name.strip() | |
key = name.lower() | |
if not key in self.things: | |
self.things[key] = {"name":name, "desc":[], "karma":0, "created":created_timestamp(who, where)} | |
return self.things[key] | |
class KarmaBot(irc.IRCClient): | |
karma_re = re.compile(r"""(?: | |
(\S+) # Short form thing | |
| # --or-- | |
(?:\((.+)\)) # Long form thing (in parens) | |
) | |
(\+\+|--) # ++ or --""", re.X | re.U) | |
affirmative_prefixes = ["Affirmative", "Alright", "Done", "K", "OK", "Okay", "Sure", "Yes"] | |
huh_msgs = ["Huh?", "What?"] | |
def connectionMade(self): | |
self.nickname = self.factory.nick | |
self.password = self.factory.password | |
irc.IRCClient.connectionMade(self) | |
self.karma = KarmaData(self.factory.filename) | |
self.karma.load() | |
def connectionLost(self, reason): | |
log.msg("Disconnected") | |
irc.IRCClient.connectionLost(self, reason) | |
self.karma.save() | |
def signedOn(self): | |
log.msg("Connected") | |
for channel in self.factory.channels: | |
self.join_with_key(channel) | |
def join_with_key(self, channel): | |
if ":" in channel: | |
channel, key = channel.split(":") | |
else: | |
key = None | |
self.join(channel, key) | |
def msg(self, user, message, length = None): | |
# Force conversion from unicode to utf-8 | |
if type(message) is unicode: | |
message = message.encode("utf-8") | |
irc.IRCClient.msg(self, user, message, length) | |
def privmsg(self, user, channel, msg): | |
log.msg("[%s] %s: %s" % (channel, user, msg)) | |
msg = msg.decode("utf-8") | |
nick = user.split("!", 1)[0] | |
# Listen for karma adjustments in messages | |
for match in self.karma_re.finditer(msg): | |
short_name, long_name, op = match.groups() | |
thing_name = short_name or long_name | |
thing = self.karma.get_thing(thing_name, user, channel) | |
if op == "++": | |
thing["karma"] += 1 | |
else: | |
thing["karma"] -= 1 | |
log.msg("(%s)%s" % (repr(thing_name), op)) | |
msg = msg.replace("++","").replace("--","") | |
# Try to find a command in the message | |
command = None | |
# Addressed (either in channel or by private message) | |
if msg.startswith(self.nickname): | |
command = msg[len(self.factory.nick):].lstrip(" ,:").rstrip() | |
who = user if channel == self.nickname else channel | |
if command: | |
log.msg("Command from %s by %s: %s" % (who, user, repr(command))) | |
# Queries | |
if command.lower() in self.karma.things: | |
self.tell_about(who, command) | |
return | |
# Descriptions | |
thing_name, sep, desc = command.partition(" is ") | |
if thing_name and desc: | |
thing = self.karma.get_thing(thing_name, user, channel) | |
if desc.startswith("<reply>"): | |
thing["desc"].append({"created":created_timestamp(user, channel), "text":desc[len("<reply>"):], "reply":True}) | |
else: | |
thing["desc"].append({"created":created_timestamp(user, channel), "text":desc}) | |
self.tell_yes(who, nick) | |
return | |
# Join/leave commands | |
call, sep, args = command.partition(" ") | |
if call == "join": | |
self.join_with_key(args.encode("utf-8")) | |
self.tell_yes(who, nick) | |
return | |
elif call == "leave": | |
self.tell_yes(who, nick) | |
self.leave(args.encode("utf-8")) | |
return | |
if user in self.factory.trusted: | |
if call == "save!": | |
self.karma.save() | |
self.tell_yes(who, nick) | |
return | |
elif call == "reload!": | |
self.karma.load() | |
self.tell_yes(who, nick) | |
return | |
elif call == "quit!": | |
self.tell_yes(who, nick) | |
reactor.iterate() | |
reactor.stop() | |
return | |
# If we got to here, no suitable command was found. | |
self.msg(who, random.choice(self.huh_msgs)) | |
def tell_yes(self, who, nick): | |
self.msg(who, "%s, %s." % (random.choice(self.affirmative_prefixes), nick)) | |
def tell_about(self, who, what): | |
if what.lower() in self.karma.things: | |
log.msg("Telling %s about %s" % (who, repr(what))) | |
thing = self.karma.get_thing(what) | |
replies = (desc["text"] for desc in thing["desc"] | |
if "reply" in desc and desc["reply"]==True) | |
for reply in replies: | |
self.msg(who, reply) | |
desc = ", ".join(desc["text"] for desc in thing["desc"] | |
if "reply" not in desc or desc["reply"]==False) | |
if desc or thing["karma"] != 0: | |
self.msg(who, u"\u0002%s\u000F: %s(%s)" % (thing["name"], desc+" " if desc else "", thing["karma"])) | |
else: | |
self.msg(who, random.choice(self.huh_msgs)) | |
class KarmaBotFactory(protocol.ClientFactory): | |
protocol = KarmaBot | |
def __init__(self, filename, nick, channels, trusted, password=None): | |
self.nick = nick | |
self.channels = channels | |
self.filename = filename | |
self.trusted = trusted | |
self.password = password | |
def clientConnectionLost(self, connector, reason): | |
# FIXME: Infinite reconnects are bad | |
#connector.connect() | |
pass | |
def clientConnectionFailed(self, connector, reason): | |
reactor.stop() | |
def main(): | |
from optparse import OptionParser | |
parser = OptionParser(usage="usage: %prog [options] channels") | |
# IRC connection options | |
parser.add_option("-s", "--server", | |
action="store", dest="server", default="irc.freenode.net", | |
help="IRC server to connect to") | |
parser.add_option("-p", "--port", | |
action="store", type="int", dest="port", default=None, | |
help="IRC server to connect to") | |
parser.add_option("--ssl", | |
action="store_true", dest="ssl", default=False, | |
help="use SSL") | |
parser.add_option("--password", | |
action="store", dest="password", default=None, | |
help="server password") | |
parser.add_option("-n", "--nick", | |
action="store", dest="nick", default="karmabot", | |
help="nickname to use") | |
# Bot options | |
parser.add_option("-v", "--verbose", | |
action="store_true", dest="verbose", default=False, | |
help="enable verbose output") | |
parser.add_option("-d", "--data", | |
action="store", dest="filename", default="karma.json", | |
help="karma data file name") | |
parser.add_option("-t", "--trust", | |
action="append", dest="trusted", default=[], | |
help="trusted hostmasks") | |
(options, channels) = parser.parse_args() | |
if not channels: | |
parser.error("You must supply some channels to join.") | |
if options.verbose: | |
log.startLogging(sys.stdout) | |
if not options.port: | |
options.port = 6667 if not options.ssl else 9999 | |
factory = KarmaBotFactory(options.filename, options.nick, channels, options.trusted, options.password) | |
if not options.ssl: | |
reactor.connectTCP(options.server, options.port, factory) | |
else: | |
reactor.connectSSL(options.server, options.port, factory, ssl.ClientContextFactory()) | |
reactor.run() | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment