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

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()