1"""Helpers for building robust Slack messages"""
  2
  3import json
  4import sys
  5import typing as t
  6from enum import Enum
  7from textwrap import dedent
  8
  9if sys.version_info >= (3, 8):
 10    from typing import TypedDict
 11else:
 12    from typing_extensions import TypedDict
 13
 14
 15SLACK_MAX_TEXT_LENGTH = 3000
 16SLACK_MAX_ALERT_PREVIEW_BLOCKS = 5
 17SLACK_MAX_ATTACHMENTS_BLOCKS = 50
 18CONTINUATION_SYMBOL = "..."
 19
 20
 21TSlackBlock = t.Dict[str, t.Any]
 22
 23
 24class TSlackBlocks(TypedDict):
 25    blocks: t.List[TSlackBlock]
 26
 27
 28class TSlackMessage(TSlackBlocks):
 29    attachments: t.List[TSlackBlocks]
 30
 31
 32class SlackMessageComposer:
 33    """Builds Slack message with primary and secondary blocks"""
 34
 35    def __init__(self, initial_message: t.Optional[TSlackMessage] = None) -> None:
 36        """Initialize the Slack message builder"""
 37        self.slack_message = initial_message or {"blocks": [], "attachments": [{"blocks": []}]}
 38
 39    def add_primary_blocks(self, *blocks: TSlackBlock) -> "SlackMessageComposer":
 40        """Add blocks to the message. Blocks are always displayed"""
 41        self.slack_message["blocks"].extend(blocks)
 42        return self
 43
 44    def add_secondary_blocks(self, *blocks: TSlackBlock) -> "SlackMessageComposer":
 45        """Add attachments to the message
 46
 47        Attachments are hidden behind "show more" button. The first 5 attachments
 48        are always displayed. NOTICE: attachments blocks are deprecated by Slack
 49        """
 50        self.slack_message["attachments"][0]["blocks"].extend(blocks)
 51        if len(self.slack_message["attachments"][0]["blocks"]) >= SLACK_MAX_ATTACHMENTS_BLOCKS:
 52            raise ValueError("Too many attachments")
 53        return self
 54
 55    def _introspect(self) -> "SlackMessageComposer":
 56        """Print the message to stdout
 57
 58        This is a debugging method. Useful during composition of the message."""
 59        print(json.dumps(self.slack_message, indent=2))
 60        return self
 61
 62
 63def normalize_message(message: t.Union[str, t.List[str], t.Tuple[str], t.Set[str]]) -> str:
 64    """Normalize message to fit Slack's max text length"""
 65    if isinstance(message, (list, tuple, set)):
 66        message = "\n".join(message)
 67    dedented_message = dedent(message)
 68    if len(dedented_message) < SLACK_MAX_TEXT_LENGTH:
 69        return dedent(dedented_message)
 70    return dedent(
 71        dedented_message[: SLACK_MAX_TEXT_LENGTH - len(CONTINUATION_SYMBOL) - 3]
 72        + CONTINUATION_SYMBOL
 73        + dedented_message[-3:]
 74    )
 75
 76
 77def divider_block() -> dict:
 78    """Create a divider block"""
 79    return {"type": "divider"}
 80
 81
 82def fields_section_block(*messages: str) -> dict:
 83    """Create a section block with multiple markdown fields"""
 84    return {
 85        "type": "section",
 86        "fields": [
 87            {
 88                "type": "mrkdwn",
 89                "text": normalize_message(message),
 90            }
 91            for message in messages
 92        ],
 93    }
 94
 95
 96def text_section_block(message: str) -> dict:
 97    """Create a section block with text"""
 98    return {
 99        "type": "section",
100        "text": {
101            "type": "mrkdwn",
102            "text": normalize_message(message),
103        },
104    }
105
106
107def preformatted_rich_text_block(message: str) -> dict:
108    """Create a "rich text" block with pre-formatted text (i.e.: a code block).
109
110    Note that this can also be acheived with text_section_block and using markdown's
111    triple backticks.
112    This function avoids issues with the text containing such backticks and also does
113    not require editing the message in order to insert such backticks.
114    """
115    return {
116        "type": "rich_text",
117        "elements": [
118            {
119                "type": "rich_text_preformatted",
120                "elements": [
121                    {
122                        "type": "text",
123                        "text": normalize_message(message),
124                    },
125                ],
126            },
127        ],
128    }
129
130
131def empty_section_block() -> dict:
132    """Create an empty section block"""
133    return {
134        "type": "section",
135        "text": {
136            "type": "mrkdwn",
137            "text": normalize_message("\t"),
138        },
139    }
140
141
142def context_block(*messages: str) -> dict:
143    """Create a context block with multiple fields"""
144    return {
145        "type": "context",
146        "elements": [
147            {
148                "type": "mrkdwn",
149                "text": normalize_message(message),
150            }
151            for message in messages
152        ],
153    }
154
155
156def header_block(message: str) -> dict:
157    """Create a header block"""
158    return {
159        "type": "header",
160        "text": {
161            "type": "plain_text",
162            "text": message,
163        },
164    }
165
166
167def button_action_block(text: str, url: str) -> dict:
168    """Create a button action block"""
169    return {
170        "type": "actions",
171        "elements": [
172            {
173                "type": "button",
174                "text": {"type": "plain_text", "text": text, "emoji": True},
175                "value": text,
176                "url": url,
177            }
178        ],
179    }
180
181
182def compacted_sections_blocks(*messages: str) -> t.List[dict]:
183    """Create a list of compacted sections blocks"""
184    return [
185        {
186            "type": "section",
187            "fields": [
188                {
189                    "type": "mrkdwn",
190                    "text": normalize_message(message),
191                }
192                for message in messages[i : i + 2]
193            ],
194        }
195        for i in range(0, len(messages), 2)
196    ]
197
198
199class SlackAlertIcon(str, Enum):
200    """Enum for status of the alert"""
201
202    # General statuses
203    X = ":x:"
204    OK = ":large_green_circle:"
205    START = ":arrow_forward:"
206    STOP = ":stop_button:"
207    WARN = ":warning:"
208    ALERT = ":rotating_light:"
209
210    # Execution statuses
211    FAILURE = ":rotating_light:"
212    SUCCESS = ":white_check_mark:"
213    WARNING = ":warning:"
214    SKIPPED = ":fast_forward:"
215    PASSED = ":white_check_mark:"
216
217    # Log levels
218    UNKNOWN = ":question:"
219    INFO = ":information_source:"
220    DEBUG = ":beetle:"
221    CRITICAL = ":fire:"
222    FATAL = ":skull_and_crossbones:"
223    EXCEPTION = ":boom:"
224
225    def __str__(self) -> str:
226        return self.value
227
228
229def stringify_list(list_variation: t.Union[t.List[str], str]) -> str:
230    """Prettify and deduplicate list of strings"""
231    if isinstance(list_variation, str):
232        return list_variation
233    if len(list_variation) == 1:
234        return list_variation[0]
235    return " ".join(list_variation)
236
237
238message = SlackMessageComposer