trac-changeset-notifier / changeset /

Joongi Kim a81a4b6 

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

from trac.core import *
from trac.config import BoolOption
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.notifications import NotifyEmail

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, self._format_subject(subject))

class ChangesetNotifier(Component):

    notify = BoolOption('changesets', 'notify_changesets', 'true',
                        """Send notifications for changeset updates.""")

    _last_cset_id = None
    # IRepositoryChangeListener methods

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

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

    # Custom methods

    def _is_duplicate(self, changeset):
        # Avoid duplicate changes with multiple scoped repositories
        cset_id = (changeset.rev, changeset.message,,
        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.
        new_path = '/'
        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': old_path,
            'old_rev': old_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
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.