hgext/bugzilla.py

add email notification hook. hook written in python.
email headers and body can be customized using template code.

# bugzilla.py - bugzilla integration for mercurial
#
# Copyright 2006 Vadim Gelfer <[e-mail]>
#
# This software may be used and distributed according to the terms
# of the GNU General Public License, incorporated herein by reference.
#
# hook extension to update comments of bugzilla bugs when changesets
# that refer to bugs by id are seen.  this hook does not change bug
# status, only comments.
#
# to configure, add items to '[bugzilla]' section of hgrc.
#
# to use, configure bugzilla extension and enable like this:
#
#   [extensions]
#   hgext.bugzilla =
#
#   [hooks]
#   # run bugzilla hook on every change pulled or pushed in here
#   incoming.bugzilla = python:hgext.bugzilla.hook
#
# config items:
#
# REQUIRED:
#   host = bugzilla # mysql server where bugzilla database lives
#   password = **   # user's password
#   version = 2.16  # version of bugzilla installed
#
# OPTIONAL:
#   bzuser = ...    # bugzilla user id to record comments with
#   db = bugs       # database to connect to
#   notify = ...    # command to run to get bugzilla to send mail
#   regexp = ...    # regexp to match bug ids (must contain one "()" group)
#   strip = 0       # number of slashes to strip for url paths
#   style = ...     # style file to use when formatting comments
#   template = ...  # template to use when formatting comments
#   timeout = 5     # database connection timeout (seconds)
#   user = bugs     # user to connect to database as
#   [web]
#   baseurl = http://hgserver/... # root of hg web site for browsing commits

from mercurial.demandload import *
from mercurial.i18n import gettext as _
from mercurial.node import *
demandload(globals(), 'mercurial:templater,util os re time')

try:
    import MySQLdb
except ImportError:
    raise util.Abort(_('python mysql support not available'))

def buglist(ids):
    return '(' + ','.join(map(str, ids)) + ')'

class bugzilla_2_16(object):
    '''support for bugzilla version 2.16.'''

    def __init__(self, ui):
        self.ui = ui
        host = self.ui.config('bugzilla', 'host', 'localhost')
        user = self.ui.config('bugzilla', 'user', 'bugs')
        passwd = self.ui.config('bugzilla', 'password')
        db = self.ui.config('bugzilla', 'db', 'bugs')
        timeout = int(self.ui.config('bugzilla', 'timeout', 5))
        self.ui.note(_('connecting to %s:%s as %s, password %s\n') %
                     (host, db, user, '*' * len(passwd)))
        self.conn = MySQLdb.connect(host=host, user=user, passwd=passwd,
                                    db=db, connect_timeout=timeout)
        self.cursor = self.conn.cursor()
        self.run('select fieldid from fielddefs where name = "longdesc"')
        ids = self.cursor.fetchall()
        if len(ids) != 1:
            raise util.Abort(_('unknown database schema'))
        self.longdesc_id = ids[0][0]
        self.user_ids = {}

    def run(self, *args, **kwargs):
        '''run a query.'''
        self.ui.note(_('query: %s %s\n') % (args, kwargs))
        try:
            self.cursor.execute(*args, **kwargs)
        except MySQLdb.MySQLError, err:
            self.ui.note(_('failed query: %s %s\n') % (args, kwargs))
            raise

    def filter_real_bug_ids(self, ids):
        '''filter not-existing bug ids from list.'''
        self.run('select bug_id from bugs where bug_id in %s' % buglist(ids))
        ids = [c[0] for c in self.cursor.fetchall()]
        ids.sort()
        return ids

    def filter_unknown_bug_ids(self, node, ids):
        '''filter bug ids from list that already refer to this changeset.'''

        self.run('''select bug_id from longdescs where
                    bug_id in %s and thetext like "%%%s%%"''' %
                 (buglist(ids), short(node)))
        unknown = dict.fromkeys(ids)
        for (id,) in self.cursor.fetchall():
            self.ui.status(_('bug %d already knows about changeset %s\n') %
                           (id, short(node)))
            unknown.pop(id, None)
        ids = unknown.keys()
        ids.sort()
        return ids

    def notify(self, ids):
        '''tell bugzilla to send mail.'''

        self.ui.status(_('telling bugzilla to send mail:\n'))
        for id in ids:
            self.ui.status(_('  bug %s\n') % id)
            cmd = self.ui.config('bugzilla', 'notify',
                               'cd /var/www/html/bugzilla && '
                               './processmail %s [e-mail]') % id
            fp = os.popen('(%s) 2>&1' % cmd)
            out = fp.read()
            ret = fp.close()
            if ret:
                self.ui.warn(out)
                raise util.Abort(_('bugzilla notify command %s') %
                                 util.explain_exit(ret)[0])
        self.ui.status(_('done\n'))

    def get_user_id(self, user):
        '''look up numeric bugzilla user id.'''
        try:
            return self.user_ids[user]
        except KeyError:
            try:
                userid = int(user)
            except ValueError:
                self.ui.note(_('looking up user %s\n') % user)
                self.run('''select userid from profiles
                            where login_name like %s''', user)
                all = self.cursor.fetchall()
                if len(all) != 1:
                    raise KeyError(user)
                userid = int(all[0][0])
            self.user_ids[user] = userid
            return userid

    def add_comment(self, bugid, text, prefuser):
        '''add comment to bug. try adding comment as committer of
        changeset, otherwise as default bugzilla user.'''
        try:
            userid = self.get_user_id(prefuser)
        except KeyError:
            try:
                defaultuser = self.ui.config('bugzilla', 'bzuser')
                userid = self.get_user_id(defaultuser)
            except KeyError:
                raise util.Abort(_('cannot find user id for %s or %s') %
                                 (prefuser, defaultuser))
        now = time.strftime('%Y-%m-%d %H:%M:%S')
        self.run('''insert into longdescs
                    (bug_id, who, bug_when, thetext)
                    values (%s, %s, %s, %s)''',
                 (bugid, userid, now, text))
        self.run('''insert into bugs_activity (bug_id, who, bug_when, fieldid)
                    values (%s, %s, %s, %s)''',
                 (bugid, userid, now, self.longdesc_id))

class bugzilla(object):
    # supported versions of bugzilla. different versions have
    # different schemas.
    _versions = {
        '2.16': bugzilla_2_16,
        }

    _default_bug_re = (r'bugs?\s*,?\s*(?:#|nos?\.?|num(?:ber)?s?)?\s*'
                       r'((?:\d+\s*(?:,?\s*(?:and)?)?\s*)+)')

    _bz = None

    def __init__(self, ui, repo):
        self.ui = ui
        self.repo = repo

    def bz(self):
        '''return object that knows how to talk to bugzilla version in
        use.'''

        if bugzilla._bz is None:
            bzversion = self.ui.config('bugzilla', 'version')
            try:
                bzclass = bugzilla._versions[bzversion]
            except KeyError:
                raise util.Abort(_('bugzilla version %s not supported') %
                                 bzversion)
            bugzilla._bz = bzclass(self.ui)
        return bugzilla._bz

    def __getattr__(self, key):
        return getattr(self.bz(), key)

    _bug_re = None
    _split_re = None

    def find_bug_ids(self, node, desc):
        '''find valid bug ids that are referred to in changeset
        comments and that do not already have references to this
        changeset.'''

        if bugzilla._bug_re is None:
            bugzilla._bug_re = re.compile(
                self.ui.config('bugzilla', 'regexp', bugzilla._default_bug_re),
                re.IGNORECASE)
            bugzilla._split_re = re.compile(r'\D+')
        start = 0
        ids = {}
        while True:
            m = bugzilla._bug_re.search(desc, start)
            if not m:
                break
            start = m.end()
            for id in bugzilla._split_re.split(m.group(1)):
                ids[int(id)] = 1
        ids = ids.keys()
        if ids:
            ids = self.filter_real_bug_ids(ids)
        if ids:
            ids = self.filter_unknown_bug_ids(node, ids)
        return ids

    def update(self, bugid, node, changes):
        '''update bugzilla bug with reference to changeset.'''

        def webroot(root):
            '''strip leading prefix of repo root and turn into
            url-safe path.'''
            count = int(self.ui.config('bugzilla', 'strip', 0))
            root = util.pconvert(root)
            while count > 0:
                c = root.find('/')
                if c == -1:
                    break
                root = root[c+1:]
                count -= 1
            return root

        mapfile = self.ui.config('bugzilla', 'style')
        tmpl = self.ui.config('bugzilla', 'template')
        sio = templater.stringio()
        t = templater.changeset_templater(self.ui, self.repo, mapfile, sio)
        if not mapfile and not tmpl:
            tmpl = _('changeset {node|short} in repo {root} refers '
                     'to bug {bug}.\ndetails:\n\t{desc|tabindent}')
        if tmpl:
            tmpl = templater.parsestring(tmpl, quoted=False)
            t.use_template(tmpl)
        t.show(changenode=node, changes=changes,
               bug=str(bugid),
               hgweb=self.ui.config('web', 'baseurl'),
               root=self.repo.root,
               webroot=webroot(self.repo.root))
        self.add_comment(bugid, sio.getvalue(), templater.email(changes[1]))

def hook(ui, repo, hooktype, node=None, **kwargs):
    '''add comment to bugzilla for each changeset that refers to a
    bugzilla bug id. only add a comment once per bug, so same change
    seen multiple times does not fill bug with duplicate data.'''
    if node is None:
        raise util.Abort(_('hook type %s does not pass a changeset id') %
                         hooktype)
    try:
        bz = bugzilla(ui, repo)
        bin_node = bin(node)
        changes = repo.changelog.read(bin_node)
        ids = bz.find_bug_ids(bin_node, changes[4])
        if ids:
            for id in ids:
                bz.update(id, bin_node, changes)
            bz.notify(ids)
        return True
    except MySQLdb.MySQLError, err:
        raise util.Abort(_('database error: %s') % err[1])