# Copyright (C) 2012-2023 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 re import textwrap import dataclasses import logging import typing from mako.template import Template from rhodecode import events from rhodecode.integrations.types.base import CommitParsingDataHandler, render_with_traceback log = logging.getLogger(__name__) @dataclasses.dataclass class SlackData: title: str text: str fields: list[dict] | None = None overrides: dict | None = None def html_to_slack_links(message): return re.compile(r'(.+?)').sub(r'<\1|\2>', message) REPO_PUSH_TEMPLATE = Template(''' <% def branch_text(branch): if branch: return 'on branch: <{}|{}>'.format(branch_commits['branch']['url'], branch_commits['branch']['name']) else: ## case for SVN no branch push... return 'to trunk' %> \ % for branch, branch_commits in branches_commits.items(): ${len(branch_commits['commits'])} ${'commit' if len(branch_commits['commits']) == 1 else 'commits'} ${branch_text(branch)} % for commit in branch_commits['commits']: `<${commit['url']}|${commit['short_id']}>` - ${commit['message_html']|html_to_slack_links} % endfor % endfor ''') class SlackDataHandler(CommitParsingDataHandler): name = 'slack' def __init__(self): pass def __call__(self, event: events.RhodecodeEvent, data): if not isinstance(event, events.RhodecodeEvent): raise TypeError(f"event {event} is not subtype of events.RhodecodeEvent") actor = data["actor"]["username"] default_title = f'*{actor}* caused a *{event.name}* event' default_text = f'*{actor}* caused a *{event.name}* event' default_slack_data = SlackData(title=default_title, text=default_text) if isinstance(event, events.PullRequestCommentEvent): return self.format_pull_request_comment_event( event, data, default_slack_data ) elif isinstance(event, events.PullRequestCommentEditEvent): return self.format_pull_request_comment_event( event, data, default_slack_data ) elif isinstance(event, events.PullRequestReviewEvent): return self.format_pull_request_review_event(event, data, default_slack_data) elif isinstance(event, events.PullRequestEvent): return self.format_pull_request_event(event, data, default_slack_data) elif isinstance(event, events.RepoPushEvent): return self.format_repo_push_event(event, data, default_slack_data) elif isinstance(event, events.RepoCreateEvent): return self.format_repo_create_event(event, data, default_slack_data) else: raise ValueError( f'event type `{event.__class__}` has no handler defined') def format_pull_request_comment_event(self, event, data, slack_data): comment_text = data['comment']['text'] if len(comment_text) > 200: comment_text = '<{comment_url}|{comment_text}...>'.format( comment_text=comment_text[:200], comment_url=data['comment']['url'], ) fields = None overrides = None status_text = None if data['comment']['status']: status_color = { 'approved': '#0ac878', 'rejected': '#e85e4d'}.get(data['comment']['status']) if status_color: overrides = {"color": status_color} status_text = data['comment']['status'] if data['comment']['file']: fields = [ { "title": "file", "value": data['comment']['file'] }, { "title": "line", "value": data['comment']['line'] } ] template = Template(textwrap.dedent(r''' *${data['actor']['username']}* left ${data['comment']['type']} on pull request <${data['pullrequest']['url']}|#${data['pullrequest']['pull_request_id']}>: ''')) title = render_with_traceback( template, data=data, comment=event.comment) template = Template(textwrap.dedent(r''' *pull request title*: ${pr_title} % if status_text: *submitted status*: `${status_text}` % endif >>> ${comment_text} ''')) text = render_with_traceback( template, comment_text=comment_text, pr_title=data['pullrequest']['title'], status_text=status_text) slack_data.title = title slack_data.text = text slack_data.fields = fields slack_data.overrides = overrides return slack_data def format_pull_request_review_event(self, event, data, slack_data) -> SlackData: template = Template(textwrap.dedent(r''' *${data['actor']['username']}* changed status of pull request <${data['pullrequest']['url']}|#${data['pullrequest']['pull_request_id']} to `${data['pullrequest']['status']}`>: ''')) title = render_with_traceback(template, data=data) template = Template(textwrap.dedent(r''' *pull request title*: ${pr_title} ''')) text = render_with_traceback( template, pr_title=data['pullrequest']['title']) slack_data.title = title slack_data.text = text return slack_data def format_pull_request_event(self, event, data, slack_data) -> SlackData: action = { events.PullRequestCloseEvent: 'closed', events.PullRequestMergeEvent: 'merged', events.PullRequestUpdateEvent: 'updated', events.PullRequestCreateEvent: 'created', }.get(event.__class__, str(event.__class__)) template = Template(textwrap.dedent(r''' *${data['actor']['username']}* `${action}` pull request <${data['pullrequest']['url']}|#${data['pullrequest']['pull_request_id']}>: ''')) title = render_with_traceback(template, data=data, action=action) template = Template(textwrap.dedent(r''' *pull request title*: ${pr_title} %if data['pullrequest']['commits']: *commits*: ${len(data['pullrequest']['commits'])} %endif ''')) text = render_with_traceback( template, pr_title=data['pullrequest']['title'], data=data) slack_data.title = title slack_data.text = text return slack_data def format_repo_push_event(self, event, data, slack_data) -> SlackData: branches_commits = self.aggregate_branch_data( data['push']['branches'], data['push']['commits']) template = Template(r''' *${data['actor']['username']}* pushed to repo <${data['repo']['url']}|${data['repo']['repo_name']}>: ''') title = render_with_traceback(template, data=data) text = render_with_traceback( REPO_PUSH_TEMPLATE, data=data, branches_commits=branches_commits, html_to_slack_links=html_to_slack_links, ) slack_data.title = title slack_data.text = text return slack_data def format_repo_create_event(self, event, data, slack_data) -> SlackData: template = Template(r''' *${data['actor']['username']}* created new repository ${data['repo']['repo_name']}: ''') title = render_with_traceback(template, data=data) template = Template(textwrap.dedent(r''' repo_url: ${data['repo']['url']} repo_type: ${data['repo']['repo_type']} ''')) text = render_with_traceback(template, data=data) slack_data.title = title slack_data.text = text return slack_data