Module bussilab.notify

Module implementing Slack notifications.

This module sends notification through an App installed in the Slack workspace. Some settings are needed first for authentication. It is recommended to add a file named .bussilabrc to your home directory with the following content:

notify:
  token: xoxb-00000
  channel: U00000

The token here should be provided by the administrator of your workspace. The channel should be the Slack ID associated to your user. It can be found looking in your Slack profile. With these settings, the tool will send notifications to you by default.

Notifications can then be sent using either the command line:

bussilab notify --message "text here"

or from python:

from bussilab.notify import notify
notify("text here")

Notice that the message is optional. Even with an empty message, the footer will allow you to reconstruct from which machine and directory the message was sent from. This might be sufficient for your goal.

Link and media previews can be disabled using the unfurl option:

bussilab notify --message "https://example.com" --no-unfurl

or from python:

notify("https://example.com", unfurl=False)

A file written in standard Markdown can be sent without adding a title or footer:

bussilab notify --markdown-file report.md

or from python:

notify(markdown_file="report.md")

You can also indicate a specific channel for the notification using the channel option:

bussilab notify --message "text here" --channel "project-myproject"

or from python:

from bussilab.notify import notify
notify("text here", channel="project-myproject")

This will only work if the App has been added to the specified channel.

The following syntax can be used to upload one or more files:

bussilab notify --message "text here" --file /path/to/file
bussilab notify --message "text here" --file first.dat second.dat

or from python:

from bussilab.notify import notify
notify("text here",file="/path/to/file")
notify("text here",file=["first.dat", "second.dat"])

The commands above will return the URL of the message. This URL can be used later to update or delete them or to post reactions:

url=$(bussilab notify --message "text here")
bussilab notify --update $url --message "revised message"
bussilab notify --react $url:heart

# this will remove only the reaction:
bussilab notify --delete $url:heart

# this will remove the entire message:
bussilab notify --delete $url

url=$(bussilab notify --message "text here")

or from python:

from bussilab.notify import notify
url=notify("text here")
notify("revised message", update=url)
notify(react=url+":heart")
notify(delete=url+":heart")
notify(delete=url)

In these cases, the channel is not needed and should not be provided. Notice that you will only be able to update or delete messages sent through the App.

Functions

def notify(message: str = '',
channel: str = None,
*,
markdown_file: str = '',
react: str = None,
update: str = None,
delete: str = None,
reply: str = None,
reply_broadcast: str = None,
title: str = '',
screenlog: str = '',
screenlog_maxlines: int = 0,
footer: bool = True,
unfurl: bool = True,
type: str = 'mrkdwn',
file: str | Sequence[str] = '',
token: str = None)

Expand source code

def notify(message: str = "",
           channel: str = None,
           *,
           markdown_file: str = "",
           react: str = None,
           update: str = None,
           delete: str = None,
           reply: str = None,
           reply_broadcast: str = None,
           title: str = "",
           screenlog: str = "",
           screenlog_maxlines: int = 0,
           footer: bool = True,
           unfurl: bool = True,
           type: str = "mrkdwn",
           file: Union[str, Sequence[str]] = "",
           token: str = None):
    """Tool to send notifications to Slack.

       Parameters
       ----------

       message: str

           A string that will form the body of the message.

       channel: None or str

           The channel. By default, taken from your `~/.bussilabrc`
           configuration file.

       markdown_file: str

           Read a standard Markdown message from this UTF-8 text file. This
           implies `type="markdown"` and cannot be combined with a non-empty
           `message` argument.

       update: None or str

           The URL of a message to be updated.

       delete: None or str

           The URL of a message to be deleted. By passing a URL
           concatenated with the string `":name_of_reaction"` you can
           delete a reaction. By passing comma-separated URLs you can delete
           a message and all the files shared with it.
       
       reply: None or str
       
           The URL of a message to be replied
       
       reply_broadcast: None or str
       
           The URL of a message to be broadcast-replied
       
       react: None or str
       
           The URL of a message to which you want to add a reaction,
           followed by the string `:name_of_the_reaction`
           
       file: str or sequence of str
       
           The path of a file to be uploaded, or paths of multiple files.

       title: str

           The title of the notification.

       screenlog: str

           The path of a GNU Screen log file. Its contents are displayed in
           a fenced native Markdown block.

       screenlog_maxlines: int

           If positive, include only this many lines from the end of the
           Screen log.

       footer: bool

           If True, a footer is added with current user, machine, and
           directory.

       unfurl: bool

           If False, link and media previews are disabled when posting a
           message or a reply. The option does not affect file-upload
           comments or message updates.

       type: str

           The type of message. Can be "mrkdwn", "plain_text", or
           "markdown". Standard Markdown uses a native Markdown block;
           `title`, `screenlog`, file uploads, and footers are not supported.

       token: None or str

           The token. By default, taken from your `~/.bussilabrc`
           configuration file.

       Returns
       -------

           str
               A string with the URL of the sent message.
               In case the `delete` keyword is used, it returns an empty
               string.
               When files are uploaded, it returns comma-separated URLs for
               the message followed by every file.


       Example
       -------

       ```python
       from bussilab.notify import notify
       notify("send this message")
       ```
       See `bussilab.notify` for more examples.
    """

    if type not in ("mrkdwn", "plain_text", "markdown"):
        raise TypeError("type should be 'mrkdwn', 'plain_text', or 'markdown'")

    if message and markdown_file:
        raise TypeError("message and markdown_file are mutually incompatible")

    if markdown_file:
        if type == "plain_text":
            raise TypeError("markdown_file is incompatible with plain_text")
        type="markdown"

    files = [file] if isinstance(file, str) else list(file)
    files = [path for path in files if path]

    if type == "markdown":
        if title:
            raise TypeError("title is not supported with standard Markdown")
        if screenlog:
            raise TypeError("screenlog is not supported with standard Markdown")
        if files:
            raise TypeError("file uploads are not supported with standard Markdown")
        footer=False

    if [bool(channel),
        bool(update),
        bool(react),
        bool(delete),
        bool(reply),
        bool(reply_broadcast)
       ].count(True)>1:
        raise TypeError("channel/update/react/delete/reply/reply_broadcast are mutually incompatible")

    if files and (update or react or delete or reply_broadcast):
        raise TypeError("files cannot be updated")

    config = None
    if token is None:
        config = coretools.config()
        token=config["notify"]["token"]

    client = WebClient(token=token)

    if delete:
        # this is to enable deletion of both a message and a file:
        delete_multi=delete.split(",")
        if len(delete_multi)>1:
            for d in delete_multi:
                notify(message,channel,delete=d,token=token)
            return ""
        delete_dict=_require_url(delete, "delete", "message", "file", "reaction")
        if delete_dict["type"]=="message":
            _try_multiple_times(client.chat_delete,
                                channel=delete_dict["channel"],
                                ts=delete_dict["ts"])
        elif delete_dict["type"]=="file":
            _try_multiple_times(client.files_delete,
                                file=delete_dict["id"])
        elif delete_dict["type"]=="reaction":
            _try_multiple_times(client.reactions_remove,
                                channel=delete_dict["channel"],
                                timestamp=delete_dict["ts"],
                                name=delete_dict["reaction"])
        else:
            raise RuntimeError("unknown type")
        # delete always returns an empty string
        return ""

    if react:
        react_dict=_require_url(react, "reaction", "reaction")
        response = _try_multiple_times(client.reactions_add,
          name=react_dict["reaction"],
          timestamp=react_dict["ts"],
          channel=react_dict["channel"])
        return react

    if markdown_file:
        with open(markdown_file, encoding="utf-8") as handler:
            message=handler.read()

    screenlog_message=""
    if len(screenlog)>0:
        # we manually removed "deleted" lines.
        # this is very useful for tdqm-like logs
        with open(screenlog,'rb') as handler:
            screenlog_message=handler.read().decode()
            screenlog_message=re.sub(r'.*\r([^\n])', r'\1', screenlog_message, flags=re.M)
        if screenlog_maxlines>0:
            screenlog_message_lines=screenlog_message.split("\n")
            if len(screenlog_message_lines) > screenlog_maxlines:
                screenlog_message_lines = screenlog_message_lines[-screenlog_maxlines:]
            screenlog_message="\n".join(screenlog_message_lines)

    screenlog_limit = (_MARKDOWN_BLOCK_LIMIT - len(_CODE_BLOCK_PREFIX)
                       - len(_CODE_BLOCK_SUFFIX))
    if len(screenlog_message)>screenlog_limit:
        screenlog_message = (screenlog_message[
            :screenlog_limit-len(_TRUNCATION_MARKER)
        ] + _TRUNCATION_MARKER)
        
    if type == "markdown":
        if len(message)>_MARKDOWN_BLOCK_LIMIT:
            message = message[
                :_MARKDOWN_BLOCK_LIMIT-len(_TRUNCATION_MARKER)
            ] + _TRUNCATION_MARKER
    elif len(message)>2900:
        message=message[:2900] + " [truncated]"

    if len(title)>2900:
        title=title[:2900] + " [truncated]"

    if update:
        update_dict=_require_url(update, "update", "message")
        organization=update_dict["organization"]
    elif reply:
        reply_dict=_require_url(reply, "reply", "message")
        organization=reply_dict["organization"]
    elif reply_broadcast:
        reply_dict=_require_url(reply_broadcast, "reply_broadcast", "message")
        organization=reply_dict["organization"]
    else:
        if channel is None:
            if config is None:
                config = coretools.config()
            channel=config["notify"]["channel"]
        if re.match(r"^https://[^/]*\.slack\.com/archives/.*", channel):
            organization = re.sub("^https://","", re.sub(r"\.slack\.com/archives/.*","",channel))
            channel=re.sub(r"^https://[^/]*\.slack\.com/archives/","",channel)
        else:
            # this is needed to set organization correctly (so as to build the
            # proper link) when passing the name of a channel
            organization = ""

    blocks=[]
    text=""

    if len(title) > 0:
        text+="*" + title+"*\n"
        blocks.append(
           {
               "type": "section",
               "text": {"type": "mrkdwn", "text": "*" + title + "*"},
           }
           )

    if len(message) > 0:
        text+=message+"\n"
        if type == "markdown":
            blocks.append({"type": "markdown", "text": message})
        else:
            blocks.append(
               {
                   "type": "section",
                   "text": {
                             "type": type,
                             "text": message
                           },
               }
               )
        
    if len(screenlog_message) > 0:
        text+=screenlog_message+"\n"
        blocks.append({
            "type": "markdown",
            "text": (_CODE_BLOCK_PREFIX + screenlog_message
                     + _CODE_BLOCK_SUFFIX)
        })

    if footer:
        footer_text = ""
        if update:
            footer_text += "Updated"
        else:
            footer_text += "Sent"
        footer_text += " by "+ os.environ['USER']
        footer_text += " at " + socket.gethostname() +'\n'
        footer_text += "pwd: " + os.getcwd() + '\n'
        footer_text += datetime.datetime.now().isoformat(sep=' ',timespec='milliseconds')
        text+=footer_text+"\n"
        blocks.append({
                          "type": "context",
                          "elements": [
                              {  # type: ignore
                                "type": "mrkdwn",
                                "text": footer_text
                              }
                          ]
                      })
    if len(blocks)==0:
        text+="(empty notification)"
        if type == "markdown":
            blocks.append({"type": "markdown", "text": "(empty notification)"})
        else:
            blocks.append({
                              "type": "section",
                              "text": {
                                         "type": type,
                                         "text": "(empty notification)"
                                      }
                          })

    unfurl_options = {}
    if not unfurl:
        unfurl_options["unfurl_links"] = False
        unfurl_options["unfurl_media"] = False

    if update:
        response = _try_multiple_times(client.chat_update,
                   channel=update_dict["channel"],
                   text=text,
                   blocks=blocks,
                   ts=update_dict["ts"])
    elif files:
        initial_comment = ""
        if len(title)>0:
            initial_comment += "*" + title + "*\n"
        if len(message)>0:
            initial_comment += message +"\n"
        if footer:
            initial_comment += footer_text

        upload_arguments = {"initial_comment": initial_comment}
        if reply:
            upload_arguments.update(channel=reply_dict["channel"],
                                    thread_ts=reply_dict["ts"])
        else:
            upload_arguments["channel"] = channel
        if len(files) == 1:
            upload_title = title or os.path.basename(files[0])
            upload_arguments.update(file=files[0], title=upload_title)
        else:
            upload_arguments["file_uploads"] = [
                {"file": path, "title": title or os.path.basename(path)}
                for path in files
            ]

        response = _try_multiple_times(client.files_upload_v2,
                                       **upload_arguments)
        uploaded_files = list(response["files"])
        uploaded_file=uploaded_files[0]
        share = _file_share(uploaded_file)
        if share is None:
            file_id=uploaded_file["id"]
            max_attempts=10
            num_attempts=0
            num_attempts_delay=3
            jittering=0.2
            time.sleep(2.0) # wait before first attempt
            while True:
                num_attempts+=1
                response = _try_multiple_times(client.files_info, file=file_id)
                share = _file_share(response["file"])
                if share is not None:
                    uploaded_files[0]=response["file"]
                    break
                if num_attempts>=max_attempts:
                    raise RuntimeError("Cannot obtain shares info for file ID "+str(file_id))
                wait=2.0
                if num_attempts>num_attempts_delay:
                    wait*=2**(num_attempts-num_attempts_delay)
                wait*=random.uniform(1,1+jittering)
                warnings.warn("Slack API, missing shares for file ID " + file_id  +", retry after "
                              +str(wait)
                              +" seconds"+
                              " ["+str(num_attempts)+"/"+str(max_attempts)+"]",
                              UserWarning)
                time.sleep(wait)
        channel, ts = share

    elif reply:
        response = _try_multiple_times(client.chat_postMessage,
                   blocks=blocks,
                   text=text,
                   channel=reply_dict["channel"],
                   thread_ts=reply_dict["ts"],
                   **unfurl_options)
    elif reply_broadcast:
        response = _try_multiple_times(client.chat_postMessage,
                   blocks=blocks,
                   text=text,
                   channel=reply_dict["channel"],
                   thread_ts=reply_dict["ts"],
                   reply_broadcast=True,
                   **unfurl_options)
    else:
        response = _try_multiple_times(client.chat_postMessage,
                   blocks=blocks,
                   text=text,
                   channel=channel,
                   **unfurl_options)

    response = cast(SlackResponse, response)

    if len(organization)==0:
        base_url=_try_multiple_times(client.auth_test)["url"]
    else:
        base_url="https://" + organization + ".slack.com/"

    if not files:
        url=base_url + "archives/" + response["channel"] + "/p" + response["ts"][:-7] + response["ts"][-6:]
    else:
        url=base_url + "archives/" + channel + "/p" + ts[:-7] + ts[-6:]
        for uploaded_file in uploaded_files:
            url+="," + base_url + "files/" + uploaded_file["user"] + "/" + uploaded_file["id"]

    return url

Tool to send notifications to Slack.

Parameters

message : str

A string that will form the body of the message.

channel : None or str

The channel. By default, taken from your ~/.bussilabrc configuration file.

markdown_file : str

Read a standard Markdown message from this UTF-8 text file. This implies type="markdown" and cannot be combined with a non-empty message argument.

update : None or str

The URL of a message to be updated.

delete : None or str

The URL of a message to be deleted. By passing a URL concatenated with the string ":name_of_reaction" you can delete a reaction. By passing comma-separated URLs you can delete a message and all the files shared with it.

reply : None or str

The URL of a message to be replied

reply_broadcast : None or str

The URL of a message to be broadcast-replied

react : None or str

The URL of a message to which you want to add a reaction, followed by the string :name_of_the_reaction

file : str or sequence of str

The path of a file to be uploaded, or paths of multiple files.

title : str

The title of the notification.

screenlog : str

The path of a GNU Screen log file. Its contents are displayed in a fenced native Markdown block.

screenlog_maxlines : int

If positive, include only this many lines from the end of the Screen log.

footer : bool

If True, a footer is added with current user, machine, and directory.

unfurl : bool

If False, link and media previews are disabled when posting a message or a reply. The option does not affect file-upload comments or message updates.

type : str

The type of message. Can be "mrkdwn", "plain_text", or "markdown". Standard Markdown uses a native Markdown block; title, screenlog, file uploads, and footers are not supported.

token : None or str

The token. By default, taken from your ~/.bussilabrc configuration file.

Returns

str
    A string with the URL of the sent message.
    In case the <code>delete</code> keyword is used, it returns an empty
    string.
    When files are uploaded, it returns comma-separated URLs for
    the message followed by every file.

Example

from bussilab.notify import notify
notify("send this message")

See bussilab.notify for more examples.