# Copyright (C) 2012-2020 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), 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.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see
|
${'This is a notification from RhodeCode. %(instance_url)s' % {'instance_url': instance_url}}
''') class EmailSettingsSchema(colander.Schema): @colander.instantiate(validator=colander.Length(min=1)) class recipients(colander.SequenceSchema): title = _('Recipients') description = _('Email addresses to send push events to') widget = deform.widget.SequenceWidget(min_len=1) recipient = colander.SchemaNode( colander.String(), title=_('Email address'), description=_('Email address'), default='', validator=colander.Email(), widget=deform.widget.TextInputWidget( placeholder='user@domain.com', ), ) class EmailIntegrationType(IntegrationTypeBase): key = 'email' display_name = _('Email') description = _('Send repo push summaries to a list of recipients via email') valid_events = [ events.RepoPushEvent ] @classmethod def icon(cls): return ''' ''' def settings_schema(self): schema = EmailSettingsSchema() schema.add(colander.SchemaNode( colander.Set(), widget=CheckboxChoiceWidgetDesc( values=sorted( [(e.name, e.display_name, e.description) for e in self.valid_events] ), ), description="List of events activated for this integration", name='events' )) return schema def send_event(self, event): log.debug('handling event %s with integration %s', event.name, self) if event.__class__ not in self.valid_events: log.debug('event %r not present in valid event list (%s)', event, self.valid_events) return if not self.event_enabled(event): # NOTE(marcink): for legacy reasons we're skipping this check... # since the email event haven't had any settings... pass handler = EmailEventHandler(self.settings) handler(event, event_data=event.as_dict()) class EmailEventHandler(object): def __init__(self, integration_settings): self.integration_settings = integration_settings def __call__(self, event, event_data): if isinstance(event, events.RepoPushEvent): self.repo_push_handler(event, event_data) else: log.debug('ignoring event: %r', event) def repo_push_handler(self, event, data): commit_num = len(data['push']['commits']) server_url = data['server_url'] if commit_num == 1: if data['push']['branches']: _subject = '[{repo_name}] {author} pushed {commit_num} commit on branches: {branches}' else: _subject = '[{repo_name}] {author} pushed {commit_num} commit' subject = _subject.format( author=data['actor']['username'], repo_name=data['repo']['repo_name'], commit_num=commit_num, branches=', '.join( branch['name'] for branch in data['push']['branches']) ) else: if data['push']['branches']: _subject = '[{repo_name}] {author} pushed {commit_num} commits on branches: {branches}' else: _subject = '[{repo_name}] {author} pushed {commit_num} commits' subject = _subject.format( author=data['actor']['username'], repo_name=data['repo']['repo_name'], commit_num=commit_num, branches=', '.join( branch['name'] for branch in data['push']['branches'])) email_body_plaintext = render_with_traceback( REPO_PUSH_TEMPLATE_PLAINTEXT, data=data, subject=subject, instance_url=server_url) email_body_html = render_with_traceback( REPO_PUSH_TEMPLATE_HTML, data=data, subject=subject, instance_url=server_url) recipients = self.integration_settings['recipients'] for email_address in recipients: run_task(tasks.send_email, email_address, subject, email_body_plaintext, email_body_html)