1. Joongi Kim
  2. trac-changeset-notifier


trac-changeset-notifier / tracext / changesetnotifier / api.py

# -*- coding: utf-8 -*-
from StringIO import StringIO

from trac.core import *
from trac.config import BoolOption, Option
from trac.util.text import CRLF
from trac.mimeview import Mimeview
from trac.versioncontrol import IRepositoryChangeListener
from trac.versioncontrol.api import Changeset, Node
from trac.versioncontrol.diff import unified_diff
from trac.notification import NotifyEmail
from trac.web.chrome import ITemplateProvider

class ChangesetNotifyEmail(NotifyEmail):
    template_name = 'changeset_notify_email.txt'

    def __init__(self, env):
        NotifyEmail.__init__(self, env)

    def notify(self, resid, subject, diff):
            'diff_body': diff,
        NotifyEmail.notify(self, resid, subject)

    def get_recipients(self, resid):
        return (self.config.get('changeset', 'notification_recipients'), '')

class ChangesetNotifier(Component):
    implements(IRepositoryChangeListener, ITemplateProvider)

    notify = BoolOption('changeset', 'notify_changesets', 'true',
                        """Send notifications for changeset updates.""")
    recipients = Option('changeset', 'notification_recipients', ''
                        """Sets the changeset notification recipients.
                        The default notification CC option is also applied.""")

    _last_cset_id = None

    # IRepositoryChangeListener methods

    def changeset_added(self, repos, changeset):
        if not self.notify:
        if self._is_duplicate(changeset):
        diff = self._make_diff(repos, changeset)
        cn = ChangesetNotifyEmail(self.env)
        reponame = repos.reponame if repos.reponame else '(default)'
        subject = self._format_subject('Changeset %s in repository %s' %
                                       (repos.display_rev(changeset.rev), reponame))
        cn.notify(changeset.rev, subject, diff)

    def changeset_modified(self, repos, changeset, old_changeset):
        if not self.notify:
        if self._is_duplicate(changeset):
        # currently not used.
        raise NotImplementedError

    # ITemplateProvider methods

    def get_templates_dirs(self):
        from pkg_resources import resource_filename
        return [resource_filename(__name__, 'templates')]

    # Custom methods

    def _is_duplicate(self, changeset):
        # Avoid duplicate changes with multiple scoped repositories
        cset_id = (changeset.rev, changeset.message, changeset.author,
        if cset_id != self._last_cset_id:
            self._last_cset_id = cset_id
            return False
        return True

    def _format_subject(self, subject):
        prefix = self.config.get('notification', 'smtp_subject_prefix')
        if prefix == '__default__':
            prefix = '[%s]' % self.env.project_name
        return '%s %s' % (prefix, subject)

    def _make_diff(self, repos, changeset):
        """Generate a unified diff for the given changeset."""
        buf = StringIO()
        mimeview = Mimeview(self.env)

        # We always compare the whole changeset from the root.
        # The below code is borrowed from ChangesetModule._render_diff() in
        # trac.versioncontrol.web_ui.changeset.
        new_path = u''
        new_rev = changeset.rev

        prev = repos.get_node(new_path, new_rev).get_previous()
        if prev:
            prev_path, prev_rev = prev[:2]
            prev_path, prev_rev = new_path, repos.previous_rev(new_rev)
        data = {
            'new_path': new_path,
            'new_rev': new_rev,
            'old_path': prev_path,
            'old_rev': prev_rev,
            'diff': {
                'options': {}, # use default options

        for old_node, new_node, kind, change in repos.get_changes(
                new_path=data['new_path'], new_rev=data['new_rev'],
                old_path=data['old_path'], old_rev=data['old_rev']):
            # TODO: Property changes
            # Content changes
            if kind == Node.DIRECTORY:
            new_content = old_content = ''
            new_node_info = old_node_info = ('','')
            if old_node:
                #if not old_node.can_view(req.perm):
                #    continue
                if mimeview.is_binary(old_node.content_type, old_node.path):
                old_content = old_node.get_content().read()
                if mimeview.is_binary(content=old_content):
                old_node_info = (old_node.path, old_node.rev)
                old_content = mimeview.to_unicode(old_content,
            if new_node:
                #if not new_node.can_view(req.perm):
                #    continue
                if mimeview.is_binary(new_node.content_type, new_node.path):
                new_content = new_node.get_content().read()
                if mimeview.is_binary(content=new_content):
                new_node_info = (new_node.path, new_node.rev)
                new_path = new_node.path
                new_content = mimeview.to_unicode(new_content,
                old_node_path = repos.normalize_path(old_node.path)
                diff_old_path = repos.normalize_path(data['old_path'])
                new_path = pathjoin(data['new_path'],
                                    old_node_path[len(diff_old_path) + 1:])
            if old_content != new_content:
                options = data['diff']['options']
                context = options.get('contextlines', 3)
                if context < 0 or options.get('contextall'):
                    context = 3 # FIXME: unified_diff bugs with context=None
                ignore_blank_lines = options.get('ignoreblanklines')
                ignore_case = options.get('ignorecase')
                ignore_space = options.get('ignorewhitespace')
                if not old_node_info[0]:
                    old_node_info = new_node_info # support for 'A'dd changes
                buf.write('Index: ' + new_path + CRLF)
                buf.write('=' * 67 + CRLF)
                buf.write('--- %s\t(revision %s)' % old_node_info + CRLF)
                buf.write('+++ %s\t(revision %s)' % new_node_info + CRLF)
                for line in unified_diff(old_content.splitlines(),
                                         new_content.splitlines(), context,
                    buf.write(line + CRLF)

        diff_str = buf.getvalue().encode('utf-8')
        return diff_str