I was looking for a simple IRC bot which would announce changes to the Quod Libet project, like what CIA does for version control. Finding none, I hacked one together in about an hour, using Twisted and Feed Parser. It’s trivial, but code that isn’t shared is lost, and it may save a few others some time in the future. If you maintain a GC project and want me to host an instance of the bot, send me an email.
quodlibot.py (download)1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | # Copyright (c) 2009 Steven Robertson.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or
# later, as published by the Free Software Foundation.
NAME="Google_Code_RSS_IRC_Bridge_Bot"
VERSION="0.1"
from twisted.words.protocols import irc
from twisted.internet import reactor, protocol, task
import feedparser
import re
import sys
import urllib2
class AnnounceBot(irc.IRCClient):
username = "%s-%s" % (NAME, VERSION)
sourceURL = "http://strobe.cc/"
# I am a terrible person.
instance = None
# Intentionally 'None' until we join a channel
channel = None
# Prevent flooding
lineRate = 3
def signedOn(self):
self.join(self.factory.channel)
AnnounceBot.instance = self
def joined(self, channel):
self.channel = self.factory.channel
def left(self, channel):
self.channel = None
def trysay(self, msg):
"""Attempts to send the given message to the channel."""
if self.channel:
try:
self.say(self.channel, msg)
return True
except: pass
class AnnounceBotFactory(protocol.ReconnectingClientFactory):
protocol = AnnounceBot
def __init__(self, channel):
self.channel = channel
def clientConnectionFailed(self, connector, reason):
print "connection failed:", reason
reactor.stop()
class FeedReader:
_schema = 'http://code.google.com/feeds/p/%s/updates/basic'
def __init__(self, project):
self.project = project
self.entries = {}
def update(self):
"""Returns list of new items."""
feed = feedparser.parse(self._schema % self.project)
added = []
for entry in feed['entries']:
if entry['id'] not in self.entries:
self.entries[entry['id']] = entry
added.append(entry)
return added
def strip_tags(value):
return re.sub(r'<[^>]*?>', '', value)
def announce(feed):
new = feed.update()
for entry in new:
msg = '%s: %s' % (strip_tags(entry['title']), entry['link'])
if AnnounceBot.instance:
AnnounceBot.instance.trysay(msg.replace('\n', '').encode('utf-8'))
if __name__ == '__main__':
# All per-project customizations should be done here
AnnounceBot.nickname = 'quodlibot'
fact = AnnounceBotFactory("#quodlibet")
feed = FeedReader('quodlibet')
reactor.connectTCP('irc.oftc.net', 6667, fact)
# Don't reannounce every update on startup
feed.update()
update_task = task.LoopingCall(announce, feed)
update_task.start(600, now=False)
reactor.callLater(10, announce, feed)
reactor.run()
|