Check ticket trackers for new tickets and generate notifications for them through Growl.

growl_ticket_notifier

#!/usr/bin/python

import pycurl
import re
import urlparse
from StringIO import StringIO
from Growl import GrowlNotifier

"""
Check ticket trackers for new tickets and generate notifications for them
through Growl.

Robust? Not really. Quick to script up? Yes.

Run this script under cron, with a crontab entry like:

1,30 * * * * /usr/local/bin/ticket_tracker
"""

TICKET_LOG = "/Users/jesstess/.ticket_log"

class Ticket(object):
    def __init__(self):
        """
        latest_ticket: the last ticket to generate a notification
        detail_page: the page to scrape for title and submitter details
        """
        self.latest_ticket = int(file(self.ticket_log).readline().strip())
        self.detail_page = None

    def update_ticket(self):
        """
        Update self.latest_ticket to a newer number.

        Override if a project's tickets aren't numbers incrementing by 1.
        """
        self.latest_ticket += 1

    def put_latest_ticket(self):
        """
        Write the last ticket to generate a notification to the ticket log.
        """
        f = open(self.ticket_log, 'w')
        f.write(str(self.latest_ticket))
        f.close()

    def retrieve_page(self, url):
        """
        Returns a tuple of (page source, HTTP status code). HTTP-only.
        """
        c = pycurl.Curl()
        c.setopt(pycurl.URL, url)
        page = StringIO()
        c.setopt(pycurl.WRITEFUNCTION, page.write)
        c.perform()
        code = c.getinfo(pycurl.RESPONSE_CODE)
        return (page.getvalue(), code)

    def new_ticket(self):
        """
        Return True if there is a new ticket to notify on.
        """
        raise NotImplementedError

    def title(self):
        """
        Return the ticket title for a new ticket.
        """
        raise NotImplementedError

    def submitter(self):
        """
        Return the ticket submitter for a new ticket
        """
        raise NotImplementedError

    def url(self):
        """
        Return the ticket url.
        """
        raise NotImplementedError

    def passes_sanity_check(self):
        """
        Perform any additional checks on the reasonableness of the ticket
        information returned.

        Return True if it passes, False otherwise.
        """
        return True

    def update_state(self):
        """
        Update whatever state is necesary to be ready to poll for new tickets.
        """
        self.update_ticket()
        self.put_latest_ticket()


class TwistedTicket(Ticket):
    def __init__(self):
        self.type = "twisted"
        self.name = "Twisted"
        self.base_url = "http://twistedmatrix.com/trac/ticket/"
        self.ticket_log = "/Users/jesstess/.twisted_ticket"
        self.ticket_url = None
        super(TwistedTicket, self).__init__()

    def new_ticket(self):
        """
        The existence of a new ticket's page is the easiest indicator that there
        is a new ticket.
        """
        self.ticket_url = urlparse.urljoin(self.base_url, str(self.latest_ticket + 1))
        self.detail_page, code = self.retrieve_page(self.ticket_url)
        if code == 200:
            return True
        return False

    def url(self):
        return self.ticket_url

    def title(self):
        p = re.compile("<title>(.*)</title>", re.MULTILINE|re.DOTALL)
        return p.search(self.detail_page).group(1).split("\n")[1].strip()

    def submitter(self):
        p = re.compile("<td headers=\"h_reporter\" class=\"searchable\">(.*)</td>")
        return p.search(self.detail_page).group(1)

    def passes_sanity_check(self):
        """
        The timeline page shows many recent events including ticket creation. If
        our new ticket is newer than the newest ticket on the timeline page,
        something is wrong.
        """
        page, code = self.retrieve_page("http://twistedmatrix.com/trac/timeline")
        p = re.compile("<span class=\"time\">(.*)</span> Ticket <em title=\"(.*)\">#(\d+)</em>")
        latest_timeline_ticket = int(p.search(page).group(3))
        if latest_timeline_ticket > self.latest_ticket:
            return False
        return True


class GNUBaseTicket(Ticket):
    def __init__(self):
        self.bug_page_base = "http://savannah.gnu.org/bugs/?"
        self.ticket_url = None
        self.ticket = None
        super(GNUBaseTicket, self).__init__()

    def new_ticket(self):
        self.detail_page, code = self.retrieve_page(self.base_url)
        p = re.compile("<td ><a href=\"\?(\d+)\">&nbsp;#(.*)</a></td>", re.IGNORECASE)
        ticket = int(p.search(self.detail_page).group(1))
        if ticket > self.latest_ticket:
            self.ticket = ticket
            self.ticket_url = self.bug_page_base + str(self.ticket)
            return True
        return False

    def url(self):
        return self.ticket_url

    def title(self):
        p = re.compile("<td ><a href=\"\?(\d+)\">(.*)</a></td>", re.IGNORECASE)
        return re.findall(p, self.detail_page)[1][1]

    def submitter(self):
        bug_page, code = self.retrieve_page(self.bug_page_base + str(self.ticket))
        p = re.compile("<td width=\"35%\">(.*)</td>")
        return p.search(bug_page).group(1)

    def passes_sanity_check(self):
        return True

    def update_ticket(self):
        self.latest_ticket = self.ticket


class GNUScreenTicket(GNUBaseTicket):
    def __init__(self):
        self.type = "screen"
        self.name = "GNU Screen"
        self.base_url = "http://savannah.gnu.org/bugs/?group=screen"
        self.ticket_log = "/Users/jesstess/.gnuscreen_ticket"
        super(GNUScreenTicket, self).__init__()


class GNUWgetTicket(GNUBaseTicket):
    def __init__(self):
        self.type = "wget"
        self.name = "GNU Wget"
        self.base_url = "http://savannah.gnu.org/bugs/?group=wget"
        self.ticket_log = "/Users/jesstess/.gnuwget_ticket"
        super(GNUWgetTicket, self).__init__()

class OpenHatchTicket(Ticket):
    def __init__(self):
        self.type = "openhatch"
        self.name = "OpenHatch"
        self.base_url = "https://openhatch.org/bugs/issue"
        self.ticket_log = "/Users/jesstess/.openhatch_ticket"
        super(OpenHatchTicket, self).__init__()        

    def new_ticket(self):
        """
        The existence of a new ticket's page is the easiest indicator that there
        is a new ticket.
        """
        self.ticket_url = self.base_url + str(self.latest_ticket + 1)
        self.detail_page, code = self.retrieve_page(self.ticket_url)
        if code == 200:
            return True
        return False

    def url(self):
        return self.ticket_url

    def title(self):
        p = re.compile("<title>(.*)</title>", re.MULTILINE|re.DOTALL)
        return p.search(self.detail_page).group(1).split("\n")[1].strip()

    def submitter(self):
        p = re.compile("<p>Created on <b>(.*)</b> by <b>(.*)</b>")
        return p.search(self.detail_page).group(2)


if __name__ == "__main__":
    twisted = TwistedTicket()
    screen = GNUScreenTicket()
    wget = GNUWgetTicket()
    openhatch = OpenHatchTicket()

    for ticket in [twisted, screen, wget, openhatch]:
        try:
            while ticket.new_ticket():
                g = GrowlNotifier(notifications=[ticket.type])
                g.register()
                g.notify(ticket.type, ticket.name,
                         "\n\n".join([ticket.title(), ticket.submitter(), ticket.url()]),
                         sticky=True)

                ticket.update_state()
            if not ticket.passes_sanity_check():
                g = GrowlNotifier(notifications=[ticket.type])
                g.register()
                g.notify(ticket.type, "FAILED sanity check", ticket.name, sticky=True)
        except Exception, e:
            g = GrowlNotifier(notifications=["error"])
            g.register()
            g.notify("error", "ERROR processing %s ticket" % (ticket.name,),
                     e.message, sticky=True)