rhodecode/integrations/types/handlers/slack.py

# 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 <http://www.gnu.org/licenses/>.
#
# 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

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'<a .*?href=["\'](.+?)".*?>(.+?)</a>').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) -> SlackData:
        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