~jpds/ubuntu-bots/bugs-via-launchpad-api

###

# Copyright (c) 2005-2007 Dennis Kaarsemaker

# Copyright (C) 2010 Jonathan Davies

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of version 2 of the GNU General Public License as

# published by the Free Software Foundation.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

###

from supybot.commands import *

import supybot.utils as utils

import supybot.ircmsgs as ircmsgs

import supybot.ircutils as ircutils

import supybot.callbacks as callbacks

import supybot.conf as conf

import supybot.registry as registry

import supybot.schedule as schedule

import re, os, time, imaplib

import xml.dom.minidom as minidom

from htmlentitydefs import entitydefs as entities

import email.FeedParser

import SOAPpy

# Launchpad bindings.

from launchpadlib.launchpad import Launchpad

cachedir = os.path.expanduser("~/.launchpadlib/cache/")

try:

    launchpad = Launchpad.login_anonymously('Supybot: Bugtracker plugin', 'edge',

        cachedir)

except AttributeError:

    launchpad = Launchpad.login_with('Supybot: Bugtracker plugin', 'edge',

        cachedir)

bad_words = ["fuck","fuk","fucking","fuking","fukin","fuckin","fucked","fuked","fucker","shit","cunt","bastard","nazi","nigger","nigga","cock","bitches","bitch"]

def makeClean(s):

    words = s.split()

    for word in words:

        if word.lower() in bad_words:

            words[words.index(word)] = "<censored>"

    return " ".join(words)

def registerBugtracker(name, url='', description='', trackertype=''):

    conf.supybot.plugins.Bugtracker.bugtrackers().add(name)

    group       = conf.registerGroup(conf.supybot.plugins.Bugtracker.bugtrackers, name)

    URL         = conf.registerGlobalValue(group, 'url', registry.String(url, ''))

    DESC        = conf.registerGlobalValue(group, 'description', registry.String(description, ''))

    TRACKERTYPE = conf.registerGlobalValue(group, 'trackertype', registry.String(trackertype, ''))

    if url:

        URL.setValue(url)

    if description:

        DESC.setValue(description)

    if trackertype:

        if defined_bugtrackers.has_key(trackertype.lower()):

            TRACKERTYPE.setValue(trackertype.lower())

        else:

            raise BugtrackerError("Unknown trackertype: %s" % trackertype)

entre = re.compile('&(\S*?);')

def _getnodetxt(node):

    L = []

    for childnode in node.childNodes:

        if childnode.nodeType == childnode.TEXT_NODE:

            L.append(childnode.data)

    val = ''.join(L)

    if node.hasAttribute('encoding'):

        encoding = node.getAttribute('encoding')

        if encoding == 'base64':

            try:

                val = val.decode('base64')

            except:

                val = 'Cannot convert bug data from base64.'

    while entre.search(val):

        entity = entre.search(val).group(1)

        if entity in entities:

            val = entre.sub(entities[entity], val)

        else:

            val = entre.sub('?', val)

    return val

class BugtrackerError(Exception):

    """A bugtracker error"""

    pass

class BugNotFoundError(Exception):

    """Pity, bug isn't there"""

    pass

cvere = re.compile(r'<th.*?Description.*?<td.*?>(.*?)\s*</td>', re.I | re.DOTALL)

class Bugtracker(callbacks.PluginRegexp):

    """Show a link to a bug report with a brief description"""

    threaded = True

    callBefore = ['URL']

    regexps = ['turlSnarfer', 'bugSnarfer', 'oopsSnarfer', 'cveSnarfer']

    def __init__(self, irc):

        callbacks.PluginRegexp.__init__(self, irc)

        self.db = ircutils.IrcDict()

        events = []

        for name in self.registryValue('bugtrackers'):

            registerBugtracker(name)

            group = self.registryValue('bugtrackers.%s' % name.replace('.','\\.'), value=False)

            if group.trackertype() in defined_bugtrackers.keys():

                self.db[name] = defined_bugtrackers[group.trackertype()](name, group.url(), group.description())

            else:

#                raise BugtrackerError("Unknown trackertype: %s (%s)" % (group.trackertype(), name))

                self.log.warning("Unknown trackertype: %s (%s)" % (group.trackertype(), name))

        self.shorthand = utils.abbrev(self.db.keys())

        # Schedule bug reporting

        self.shown = {}

        if self.registryValue('imap_server') and self.registryValue('reportercache'):

            try:

                schedule.removeEvent(self.name() + '.bugreporter')

            except:

                pass

            schedule.addPeriodicEvent(lambda: self.reportnewbugs(irc),  60, name=self.name() + '.bugreporter')

            self.events += [self.name() + '.bugreporter']

            self.log.info('Adding scheduled event "%s.bugreporter"' % self.name())

    def die(self):

        try:

           for event in self.events:

                self.log.info('Removing scheduled event "%s"' % event)

                schedule.removeEvent(event)

                schedule.removeEvent(self.name())

        except:

            pass

    def is_ok(self, channel, tracker, bug):

        now = time.time()

        for k in self.shown.keys():

            if self.shown[k] < now - self.registryValue('repeatdelay', channel):

                self.shown.pop(k)

        if (channel, tracker, bug) not in self.shown:

            self.shown[(channel, tracker, bug)] = now

            return True

        return False

    def is_new(self, tracker, tag, id):

        bugreporter_base = self.registryValue('reportercache')

        if not os.path.exists(os.path.join(bugreporter_base,tag,tracker.name,str(int(id/1000)),str(id))):

            try:

                os.makedirs(os.path.join(bugreporter_base,tag,tracker.name,str(int(id/1000))))

            except:

                pass

            fd = open(os.path.join(bugreporter_base,tag,tracker.name,str(int(id/1000)),str(id)),'w')

            fd.close()

            return True

        return False

    def reportnewbugs(self,irc):

        # Compile list of bugs

        self.log.info("Checking for new bugs")

        bugs = {}

        if self.registryValue('imap_ssl'):

            sc = imaplib.IMAP4_SSL(self.registryValue('imap_server'))

        else:

            sc = imaplib.IMAP4(self.registryValue('imap_server'))

        sc.login(self.registryValue('imap_user'), self.registryValue('imap_password'))

        sc.select('INBOX')

        new_mail = sc.search(None, '(UNSEEN)')[1][0].split()[:20]

        # Read all new mail

        for m in new_mail:

            msg = sc.fetch(m, 'RFC822')[1][0][1]

            fp = email.FeedParser.FeedParser()

            sc.store(m, '+FLAGS', "(\Deleted)") # Mark message deleted so we don't have to process it again

            fp.feed(msg)

            bug = fp.close()

            tag = None

            if 'X-Launchpad-Bug' not in bug.keys():

                self.log.info('Ignoring e-mail with no detectable bug (Not from Launchpad)')            

                continue

            else:

                tag = bug['X-Launchpad-Bug']

                if 'distribution=' not in tag and 'product=' not in tag:

                    self.log.info('Ignoring e-mail with no detectable bug (no distro/product)')

                    continue

                else:

                    tag = tag.split(';')[0].strip().replace("product=",'').replace("distribution=","")

            if not tag:

                self.log.info('Ignoring e-mail with no detectible bug (bad tag)')

            tag = tag[tag.find('+')+1:tag.find('@')]

            if tag not in bugs:

                bugs[tag] = {}

            # Determine bugtracker type (currently only Launchpad is supported anyway)

            if bug['X-Launchpad-Bug']:

                tracker = self.db['launchpad']

                id = int(bug['Reply-To'].split()[1])

                subj = bug['Subject'];

                if '[NEW]' not in subj: #Not a new bug

                    continue

                if self.is_new(tracker, tag, id):

                    component = bug['X-Launchpad-Bug']

                    if 'component' in component:

                        component = component[component.find('component=')+10:]

                        component = component[:component.find(';')].replace('None','')

                    else:

                        component = ''

                    try:

                        if component:

                            bugs[tag][id] = self.get_bug('',tracker, id, False)[0].replace('"','(%s) "' % component, 1)

                        else:

                            bugs[tag][id] = self.get_bug('',tracker, id, False)[0]

                        if '[apport]' in bugs[tag][id]:

                            bugs[tag].pop(id)

                    except:

                        self.log.info("Unable to get new bug %d" % id)

                        pass

            else:

                self.log.info('Ignoring e-mail with no detectable bug')

        reported_bugs = 0

        for c in irc.state.channels:

            tags = self.registryValue('bugReporter', channel=c)

            if not tags:

                continue

            for tag in tags.split(','):

                if not tag or tag not in bugs.keys():

                    continue

                for b in sorted(bugs[tag].keys()):

                    irc.queueMsg(ircmsgs.privmsg(c,'New bug: #%s' % bugs[tag][b][bugs[tag][b].find('bug ')+4:]))

                    reported_bugs = reported_bugs+1

    def add(self, irc, msg, args, name, trackertype, url, description):

        """<name> <type> <url> [<description>]

        Add a bugtracker <url> to the list of defined bugtrackers. <type> is the

        type of the tracker (currently only Launchpad, Debbugs, Bugzilla,

        Issuezilla, Mantis and Trac are known). <name> is the name that will be used to

        reference the bugzilla in all commands. Unambiguous abbreviations of

        <name> will be accepted also.  <description> is the common name for the

        bugzilla and will be listed with the bugzilla query; if not given, it

        defaults to <name>.

        """

        name = name.lower()

        if not description:

            description = name

        if url[-1] == '/':

            url = url[:-1]

        trackertype = trackertype.lower()

        if trackertype in defined_bugtrackers:

            self.db[name] = defined_bugtrackers[trackertype](name,url,description)

        else:

            irc.error("Bugtrackers of type '%s' are not understood" % trackertype)

            return

        registerBugtracker(name, url, description, trackertype)

        self.shorthand = utils.abbrev(self.db.keys())

        irc.replySuccess()

    add = wrap(add, [('checkCapability', 'admin'), 'something', 'something', 'url', additional('text')])

    def remove(self, irc, msg, args, name):

        """<abbreviation>

        Remove the bugtracker associated with <abbreviation> from the list of

        defined bugtrackers.

        """

        try:

            name = self.shorthand[name.lower()]

            del self.db[name]

            self.registryValue('bugtrackers').remove(name)

            self.shorthand = utils.abbrev(self.db.keys())

            irc.replySuccess()

        except KeyError:

            s = self.registryValue('replyNoBugtracker', msg.args[0])

            irc.error(s % name)

    remove = wrap(remove, [('checkCapability', 'admin'), 'text'])

    def rename(self, irc, msg, args, oldname, newname, newdesc):

        """<oldname> <newname>

        Rename the bugtracker associated with <oldname> to <newname>.

        """

        try:

            name = self.shorthand[oldname.lower()]

            group = self.registryValue('bugtrackers.%s' % name.replace('.','\\.'), value=False)

            d = group.description()

            if newdesc:

                d = newdesc

            self.db[newname] = defined_bugtrackers[group.trackertype()](name,group.url(),d)

            registerBugtracker(newname, group.url(), d, group.trackertype())

            del self.db[name]

            self.registryValue('bugtrackers').remove(name)

            self.shorthand = utils.abbrev(self.db.keys())

            irc.replySuccess()

        except KeyError:

            s = self.registryValue('replyNoBugtracker', msg.args[0])

            irc.error(s % name)

    rename = wrap(rename, [('checkCapability', 'admin'), 'something','something', additional('text')])

    def list(self, irc,  msg, args, name):

        """[abbreviation]

        List defined bugtrackers. If [abbreviation] is specified, list the

        information for that bugtracker.

        """

        if name:

            name = name.lower()

            try:

                name = self.shorthand[name]

                (url, description, type) = (self.db[name].url, self.db[name].description,

                                            self.db[name].__class__.__name__)

                irc.reply('%s: %s, %s [%s]' % (name, description, url, type))

            except KeyError:

                s = self.registryValue('replyNoBugtracker', msg.args[0])

                irc.error(s % name)

        else:

            if self.db:

                L = self.db.keys()

                L.sort()

                irc.reply(utils.str.commaAndify(L))

            else:

                irc.reply('I have no defined bugtrackers.')

    list = wrap(list, [additional('text')])

    def bugSnarfer(self, irc, msg, match):

        r"""\b(?P<bt>(([a-z0-9]+)?\s+bugs?|[a-z0-9]+))\s+#?(?P<bug>\d+(?!\d*[\-\.]\d+)((,|\s*(and|en|et|und|ir))\s*#?\d+(?!\d*[\-\.]\d+))*)"""

        if msg.args[0][0] == '#' and not self.registryValue('bugSnarfer', msg.args[0]):

            return

        nbugs = msg.tagged('nbugs')

        if not nbugs: nbugs = 0

        if nbugs >= 5:

            return

        # Don't double on commands

        s = str(msg).split(':')[2]

        if s and s[0] in str(conf.supybot.reply.whenAddressedBy.chars):

            return

        sure_bug = match.group('bt').endswith('bug') or match.group('bt').endswith('bug')

        # Get tracker name

        bugids = match.group('bug')

        reps = ((' ',''),('#',''),('and',','),('en',','),('et',','),('und',','),('ir',','))

        for r in reps:

            bugids = bugids.replace(r[0],r[1])

        bugids = bugids.split(',')[:5-nbugs]

        if not sure_bug:

            bugids = [x for x in bugids if int(x) > 100]

        msg.tag('nbugs', nbugs + len(bugids))

        bt = map(lambda x: x.lower(), match.group('bt').split())

        name = ''

        if len(bt) == 1 and not (bt[0] in ['bug','bugs']):

            try:

                name = bt[0].lower()

                tracker = self.db[name]

            except:

                return

        elif len(bt) == 2:

            try:

                name = bt[0].lower()

                tracker = self.db[name]

            except:

                name = ''

                pass

        if not name:

            snarfTarget = self.registryValue('snarfTarget', msg.args[0]).lower()

            if not snarfTarget:

                self.log.warning("no snarfTarget for Bugtracker")

                return

            try:

                name = self.shorthand[snarfTarget.lower()]

            except:

               s = self.registryValue('replyNoBugtracker', name)

               irc.error(s % name)

        try:

            tracker = self.db[name]

        except KeyError:

            s = self.registryValue('replyNoBugtracker', name)

            irc.error(s % name)

        else:

            for bugid in bugids:

                bugid = int(bugid)

                try:

                    report = self.get_bug(msg.args[0],tracker,bugid,self.registryValue('showassignee', msg.args[0]))

                except BugNotFoundError:

                    if self.registryValue('replyWhenNotFound'):

                        irc.error("%s bug %d could not be found" % (tracker.description, bugid))

                except BugtrackerError, e:

                    if 'private' in str(e):

                        irc.reply("Bug %d on http://launchpad.net/bugs/%d is private" % (bugid, bugid))

                        return

                    if not sure_bug and bugid < 30:

                        return

                    irc.error(str(e))

                else:

                    for r in report:

                        irc.reply(makeClean(r), prefixNick=False)

    def turlSnarfer(self, irc, msg, match):

        r"(?P<tracker>https?://\S*?)/(Bugs/0*|str.php\?L|show_bug.cgi\?id=|bugreport.cgi\?bug=|(bugs|\+bug)/|ticket/|tracker/|\S*aid=)(?P<bug>\d+)(?P<sfurl>&group_id=\d+&at_id=\d+)?"

        if msg.args[0][0] == '#' and not self.registryValue('bugSnarfer', msg.args[0]):

            return

        nbugs = msg.tagged('nbugs')

        if not nbugs: nbugs = 0

        if nbugs >= 5:

            return

        msg.tag('nbugs', nbugs+1)

        try:

            tracker = self.get_tracker(match.group(0),match.group('sfurl'))

            if not tracker:

                return

            report = self.get_bug(msg.args[0],tracker,int(match.group('bug')),self.registryValue('showassignee', msg.args[0]), do_url = False)

        except BugtrackerError, e:

            irc.error(str(e))

        except BugNotFoundError, e:

            irc.error("%s bug %s not found" % (tracker, match.group('bug')))

        else:

            for r in report:

                irc.reply(makeClean(r), prefixNick=False)

    turlSnarfer = urlSnarfer(turlSnarfer)

    # Only useful for launchpad developers

    def oopsSnarfer(self, irc, msg, match):

        r"OOPS-(?P<oopsid>\d*[\dA-Z]+)"

        if msg.args[0][0] == '#' and not self.registryValue('bugSnarfer', msg.args[0]):

            return

        oopsid = match.group(1)

        if oopsid.lower() == "tools":

            return

        irc.reply("https://lp-oops.canonical.com/oops.py/?oopsid=%s" % oopsid, prefixNick=False)

        #irc.reply("https://devpad.canonical.com/~jamesh/oops.cgi/%s" % oopsid, prefixNick=False)

    def cveSnarfer(self, irc, msg, match):

        r"(cve[- ]\d{4}[- ]\d{4})"

        if msg.args[0][0] == '#' and not self.registryValue('bugSnarfer', msg.args[0]):

            return

        cve = match.group(1).replace(' ','-').upper()

        url = 'http://cve.mitre.org/cgi-bin/cvename.cgi?name=%s' % cve

        cvedata = utils.web.getUrl(url)

        m = cvere.search(cvedata)

        if m:

            cve = m.group(1).replace('\n', ' ')

            irc.reply("%s (%s)" % (cve,url))

    def cveUrlSnarfer(self, irc, msg, match):

        pass

    def get_tracker(self,snarfurl,sfdata):

        snarfurl = snarfurl.replace('sf.net','sourceforge.net')

        snarfhost = snarfurl.replace('http://','').replace('https://','')

        if '/' in snarfurl:

            snarfhost = snarfhost[:snarfhost.index('/')]

        for t in self.db.keys():

            tracker = self.db[t]

            url = tracker.url.replace('http://','').replace('https://','')

            if 'sourceforge.net' in url:

                # Try to find the correct sf tracker

                if str(sfdata) in tracker.url:

                    return tracker

            if '/' in url:

                url = url[:url.index('/')]

            if url in snarfhost:

                return tracker

        if 'sourceforge.net' in snarfurl:

            return self.db['sourceforge']

        # No tracker found, bummer. Let's try and add one

        if 'show_bug.cgi' in snarfurl:

            tracker = Bugzilla().get_tracker(snarfurl)

            if tracker:

                self.db[tracker.name] = tracker

                self.shorthand = utils.abbrev(self.db.keys())

                return tracker

        return None

    def get_bug(self, channel, tracker, id, do_assignee, do_url = True):

        reports = []

        for r in tracker.get_bug(id):

            if not self.is_ok(channel, tracker, r[0]):

                continue

            (bid, product, title, severity, status, assignee, url) = r

            severity = severity[0].upper() + severity[1:].lower()

            status = status[0].upper() + status[1:].lower()

            if not do_url:

                url = ''

            if product:

                reports.append("%s bug %s in %s \"%s\" [%s,%s] %s" % (tracker.description, bid, product, 

                                                                      title, severity, status, url))

            else:

                reports.append("%s bug %s \"%s\" [%s,%s] %s" % (tracker.description, bid, title, severity, status, url))

            if do_assignee and assignee:

                reports[-1] = reports[-1] + (" - Assigned to %s" % assignee)

        return reports

# Define all bugtrackers

class IBugtracker:

    def __init__(self, name=None, url=None, description=None):

        self.name        = name

        self.url         = url

        self.description = description

    def get_bug(self, id):

        raise BugTrackerError("Bugtracker class does not implement get_bug")

    def get_tracker(self, url):

        raise BugTrackerError("Bugtracker class does not implement get_tracker")

class Bugzilla(IBugtracker):

    def get_tracker(self, url):

        url = url.replace('show_bug','xml')

        try:

            bugxml = utils.web.getUrl(url)

            tree = minidom.parseString(bugxml)

            url  = str(tree.getElementsByTagName('bugzilla')[0].attributes['urlbase'].childNodes[0].data)

            if url[-1] == '/':

                url = url[:-1]

            name = url[url.find('//') + 2:]

            if '/' in name:

                name = name[:name.find('/')]

            desc = name

            registerBugtracker(name, url, desc, 'bugzilla')

            tracker = Bugzilla(name, url, desc)

            return tracker

        except:

            return None

    def get_bug(self, id):

        url = "%s/xml.cgi?id=%d" % (self.url,id)

        try:

            bugxml = utils.web.getUrl(url)

            zilladom = minidom.parseString(bugxml)

        except Exception, e:

            s = 'Could not parse XML returned by %s: %s (%s)' % (self.description, e, url)

            raise BugtrackerError, s

        bug_n = zilladom.getElementsByTagName('bug')[0]

        if bug_n.hasAttribute('error'):

            errtxt = bug_n.getAttribute('error')

            if errtxt == 'NotFound':

                raise BugNotFoundError

            s = 'Error getting %s bug #%s: %s' % (self.description, id, errtxt)

            raise BugtrackerError, s

        try:

            title = _getnodetxt(bug_n.getElementsByTagName('short_desc')[0])

            status = _getnodetxt(bug_n.getElementsByTagName('bug_status')[0])

            try:

                status += ": " + _getnodetxt(bug_n.getElementsByTagName('resolution')[0])

            except:

                pass