If it won't be simple, it simply won't be. [source code] by Miki Tebeka, CEO, 353Solutions

Monday, May 11, 2009

Avoiding "peek" with "itertools.chain"

Sometimes you need to filter out some header information. However the only clear anchor you have is the start of real data - you need a "peek" function.

It's easy to avoid writing the "peek" logic, just use itertools.chain.

#!/usr/bin/env python
'''Avoiding the need for "peek" with itertools.chain'''

from itertools import chain

def is_first_data(line):
return line.startswith("Name:")

def skip_header(data):
data = iter(data)
for line in data:
if is_first_data(line):
return chain([line], data)

# FIXME: We might want to raise something here
return []


if __name__ == "__main__":
data = [
"this is the header",
"it might change every time",
"and you'll never find a good regexp for it",
"The first line of data is easy to know",
"Name: Duffy",
"Type: Duck",
"Anger: 10",
"",
"Name: Bugs",
"Type: Bunny",
"Anger: 0",
]

data = skip_header(data)
for line in data:
print line

Friday, May 01, 2009

Subversion IRC Bot

#!/usr/bin/env python
'''Push changes in subversion repository to IRC channel'''

from twisted.words.protocols import irc
from twisted.internet.protocol import ReconnectingClientFactory
import re
from subprocess import Popen, PIPE
from xml.etree.cElementTree import parse as xmlparse
from cStringIO import StringIO

class IRCClient(irc.IRCClient):
nickname = "svnbot"
realname = "Subversion Bot"
channel = "#dev"

instance = None # Running instance

def signedOn(self):
IRCClient.instance = self
self.join(self.channel)

def svn(self, revision, author, comment):
comment = (comment[:57] + "...") if len(comment) > 60 else comment
message = "SVN revision %s by %s: %s" % (revision, author, comment)
self.say(self.channel, message)


class SVNPoller:
def __init__(self, root, user, password):
self.pre = ["svn", "--xml", "--username", user, "--password", password]
self.root = root
self.last_revision = self.get_last_revision()

def check(self):
if not IRCClient.instance:
return

try:
last_revision = self.get_last_revision()
if (not last_revision) or (last_revision == self.last_revision):
return

for rev in range(self.last_revision + 1, last_revision + 1):
author, comment = self.revision_info(rev)
IRCClient.instance.svn(rev, author, comment)
self.last_revision = last_revision
except Exception, e:
print "ERROR: %s" % e

def svn(self, *cmd):
pipe = Popen(self.pre + list(cmd) + [self.root], stdout=PIPE)
try:
data = pipe.communicate()[0]
except IOError:
data = ""
return xmlparse(StringIO(data))

def get_last_revision(self):
tree = self.svn("info")
revision = tree.find("//commit").get("revision")
return int(revision)

def revision_info(self, revision):
tree = self.svn("log", "-r", str(revision))
author = tree.find("//author").text
comment = tree.find("//msg").text

return author, comment

if __name__ == "__main__":
from twisted.internet import reactor
from twisted.internet.task import LoopingCall

factory = ReconnectingClientFactory()
factory.protocol = IRCClient
reactor.connectTCP("irc.mycompany.com", 6667, factory)

poller = SVNPoller("http://svn.mycompany.com", "bugs", "carrot")
task = LoopingCall(poller.check)
task.start(1)

reactor.run()

Blog Archive