# Copyright (C) 2012-2024 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 program is dual-licensed. If you wish to learn more about the # RhodeCode Enterprise Edition, including its added features, Support services, # and proprietary license terms, please see https://rhodecode.com/licenses/ import time import logging import deform # noqa import deform.widget import colander from rhodecode import events from rhodecode.model.validation_schema.widgets import CheckboxChoiceWidgetDesc from rhodecode.translation import _ from rhodecode.lib import helpers as h from rhodecode.lib.celerylib import run_task, async_task, RequestContextTask from rhodecode.lib.colander_utils import strip_whitespace from rhodecode.integrations.types.base import ( IntegrationTypeBase, requests_retry_call, ) from rhodecode.integrations.types.handlers.slack import SlackDataHandler, SlackData log = logging.getLogger(__name__) class SlackSettingsSchema(colander.Schema): service = colander.SchemaNode( colander.String(), title=_('Slack service URL'), description=h.literal(_( 'This can be setup at the ' '' 'slack app manager')), default='', preparer=strip_whitespace, validator=colander.url, widget=deform.widget.TextInputWidget( placeholder='https://hooks.slack.com/services/...', ), ) username = colander.SchemaNode( colander.String(), title=_('Username'), description=_('Username to show notifications coming from.'), missing='Rhodecode', preparer=strip_whitespace, widget=deform.widget.TextInputWidget( placeholder='Rhodecode' ), ) channel = colander.SchemaNode( colander.String(), title=_('Channel'), description=_('Channel to send notifications to.'), missing='', preparer=strip_whitespace, widget=deform.widget.TextInputWidget( placeholder='#general' ), ) icon_emoji = colander.SchemaNode( colander.String(), title=_('Emoji'), description=_('Emoji to use eg. :studio_microphone:'), missing='', preparer=strip_whitespace, widget=deform.widget.TextInputWidget( placeholder=':studio_microphone:' ), ) class SlackIntegrationType(IntegrationTypeBase): key = 'slack' display_name = _('Slack') description = _('Send events such as repo pushes and pull requests to ' 'your slack channel.') @classmethod def icon(cls): return '''''' valid_events = [ events.PullRequestCloseEvent, events.PullRequestMergeEvent, events.PullRequestUpdateEvent, events.PullRequestCommentEvent, events.PullRequestReviewEvent, events.PullRequestCreateEvent, events.RepoPushEvent, events.RepoCreateEvent, ] def settings_schema(self): schema = SlackSettingsSchema() 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): return data = event.as_dict() handler = SlackDataHandler() slack_data = handler(event, data) # title, text, fields, overrides run_task(post_text_to_slack, self.settings, slack_data) @async_task(ignore_result=True, base=RequestContextTask) def post_text_to_slack(settings, slack_data: SlackData): # because JSON serialization, if we run async with celery, deserialize to SlackData if isinstance(slack_data, dict): slack_data = SlackData(**slack_data) title = slack_data.title text = slack_data.text fields = slack_data.fields overrides = slack_data.overrides log.debug('sending %s (%s) to slack %s', title, text, settings['service']) fields = fields or [] overrides = overrides or {} message_data = { "fallback": text, "color": "#427cc9", "pretext": title, #"author_name": "Bobby Tables", #"author_link": "http://flickr.com/bobby/", #"author_icon": "http://flickr.com/icons/bobby.jpg", #"title": "Slack API Documentation", #"title_link": "https://api.slack.com/", "text": text, "fields": fields, #"image_url": "http://my-website.com/path/to/image.jpg", #"thumb_url": "http://example.com/path/to/thumb.png", "footer": "RhodeCode", #"footer_icon": "", "ts": time.time(), "mrkdwn_in": ["pretext", "text"] } message_data.update(overrides) json_message = { "icon_emoji": settings.get('icon_emoji', ':studio_microphone:'), "channel": settings.get('channel', ''), "username": settings.get('username', 'Rhodecode'), "attachments": [message_data] } req_session = requests_retry_call() resp = req_session.post(settings['service'], json=json_message, timeout=60) resp.raise_for_status() # raise exception on a failed request