如何使用 Python 构建您的第一个 Slack 机器人

机器人是与 Slack 等聊天服务进行交互的有用方式。如果您以前从未构建过机器人,这篇文章提供了一个简单的入门教程,用于将 Slack API 与 Python 结合起来创建您的第一个机器人。

我们将介绍如何设置您的开发环境、获取 Slack API 机器人令牌以及使用 Python 编写我们的简单机器人。

我们需要的工具

我们将命名为“StarterBot”的机器人需要 Python 和 Slack API。要运行我们的 Python 代码,我们需要:

  • Python 2 或 3
  • pip 和 virtualenv 来处理 Python 应用程序依赖性
  • 免费的 Slack 帐户 – 您需要登录至少一个有权构建的工作区应用程序。

在构建本教程时,手头有 Slack API 文档也很有用。

本教程的所有代码都可以在 slack-starterbot 公共存储库中根据 MIT 许可证开源获得。

建立我们的环境

我们现在知道我们的项目需要哪些工具,所以让我们来设置我们的开发环境。转到终端(或 Windows 上的命令提示符)并更改为要存储此项目的目录。在该目录中,创建一个新的 virtualenv 以将我们的应用程序依赖项与其他 Python 项目隔离开来。

virtualenv starterbot

激活虚拟环境:

source starterbot/bin/activate

您的提示现在应该类似于此屏幕截图中的提示。

已激活 starterbot 的 virtualenv 的命令提示符。

由 Slack 构建的官方 slackclient API 帮助程序库可以从 Slack 通道发送和接收消息。使用 pip 命令安装 slackclient 库:

pip install slackclient==1.3.2

pip 完成时,您应该会看到这样的输出,您将在提示符处返回。

在激活 virtualenv 的情况下使用 pip install slackclient 命令的输出。

我们还需要创建一个 Slack 应用程序来为您的机器人接收 API 令牌。使用“Starter Bot”作为您的应用程序名称。如果您登录到多个工作区,请从下拉列表中选择一个开发工作区。

创建一个填充的 Slack 应用程序表单

提交表单后,保持应用配置页面打开。

Slack API 和应用配置

我们希望我们的 Starter Bot 看起来像您团队中的任何其他用户一样 – 它将参与频道、群组和 DM 内的对话。在 SlackApp 中,这称为机器人用户,我们通过选择“功能”部分下的“机器人用户”来设置它。单击“添加机器人用户”后,您应该选择一个显示名称,选择一个默认用户名,然后单击“添加机器人用户”保存您的选择。您最终会得到一个如下所示的页面:

在 Slack 应用程序中添加了一个机器人用户

slackclient 库使得使用 Slack 的 RTM API 和 Web API 变得简单。我们将使用两者来实现 Starter Bot,它们都需要身份验证。方便的是,我们的 bot 用户之前创建的可用于对两个 API 进行身份验证。

点击“设置”部分下的“安装应用程序”。此页面上的按钮会将应用程序安装到我们的开发工作区中。安装应用程序后,它会显示一个bot 用户 oauth 访问令牌,用于作为 bot 用户进行身份验证。

开发区安装后,可以复制bot用户oauth access token

Python 开发人员的常见做法是将秘密令牌导出为环境变量。返回您的终端,导出名称为 SLACK_BOT_TOKEN 的 Slack 令牌:

export SLACK_BOT_TOKEN='your bot user access token here'

很好,现在我们被授权以机器人用户身份使用 Slack RTM 和 Web API。

编写我们的入门机器人

我们已经拥有了编写 Starter Bot 代码所需的一切。创建一个名为 starterbot.py 的新文件,并在其中包含以下代码。

import os
import time
import re
from slackclient import SlackClient

导入依赖项后,我们可以使用它们来获取环境变量值,然后实例化 Slack 客户端。

# instantiate Slack client
slack_client = SlackClient(os.environ.get('SLACK_BOT_TOKEN'))
# starterbot's user ID in Slack: value is assigned after the bot starts up
starterbot_id = None

# constants
RTM_READ_DELAY = 1 # 1 second delay between reading from RTM
EXAMPLE_COMMAND = "do"
MENTION_REGEX = "^<@(|[WU].+?)>(.*)"

代码实例化 SlackClient 客户端,我们将 SLACK_BOT_TOKEN 导出为环境变量。它还声明了一个变量,我们可以使用它来存储 Starter Bot 的 Slack 用户 ID。还声明了一些常量,每个常量都将在后面的代码中使用时进行解释。

if __name__ == "__main__":
    if slack_client.rtm_connect(with_team_state=False):
        print("Starter Bot connected and running!")
        # Read bot's user ID by calling Web API method `auth.test`
        starterbot_id = slack_client.api_call("auth.test")["user_id"]
        while True:
            command, channel = parse_bot_commands(slack_client.rtm_read())
            if command:
                handle_command(command, channel)
            time.sleep(RTM_READ_DELAY)
    else:
        print("Connection failed. Exception traceback printed above.")

Slack 客户端连接到 Slack RTM API。连接后,它会调用 Web API 方法 (auth.test) 来查找 Starter Bot 的用户 ID。

每个 bot 用户对于安装 Slack 应用程序的每个工作区都有一个用户 ID。存储此用户 ID 将有助于程序了解是否有人在消息中提到了该机器人。

接下来,程序进入无限循环,每次循环运行时,客户端都会收到来自 Slack 的 RTM API 的任何事件。请注意,在循环结束之前,程序会暂停一秒钟,以免循环太快而浪费您的 CPU 时间。

对于读取的每个事件,parse_bot_commands() 函数确定该事件是否包含 Starter Bot 的命令。如果是,则 command 将包含一个值,并且 handle_command() 函数决定如何处理该命令。

我们已经为在程序中处理 Slack 事件和调用 Slack 方法奠定了基础。接下来,在前面的代码片段上方添加三个新函数以完成处理命令:

def parse_bot_commands(slack_events):
    """
        Parses a list of events coming from the Slack RTM API to find bot commands.
        If a bot command is found, this function returns a tuple of command and channel.
        If its not found, then this function returns None, None.
    """
    for event in slack_events:
        if event["type"] == "message" and not "subtype" in event:
            user_id, message = parse_direct_mention(event["text"])
            if user_id == starterbot_id:
                return message, event["channel"]
    return None, None

def parse_direct_mention(message_text):
    """
        Finds a direct mention (a mention that is at the beginning) in message text
        and returns the user ID which was mentioned. If there is no direct mention, returns None
    """
    matches = re.search(MENTION_REGEX, message_text)
    # the first group contains the username, the second group contains the remaining message
    return (matches.group(1), matches.group(2).strip()) if matches else (None, None)

def handle_command(command, channel):
    """
        Executes bot command if the command is known
    """
    # Default response is help text for the user
    default_response = "Not sure what you mean. Try *{}*.".format(EXAMPLE_COMMAND)

    # Finds and executes the given command, filling in response
    response = None
    # This is where you start to implement more commands!
    if command.startswith(EXAMPLE_COMMAND):
        response = "Sure...write some more code then I can do that!"

    # Sends the response back to the channel
    slack_client.api_call(
        "chat.postMessage",
        channel=channel,
        text=response or default_response
    )

parse_bot_commands() 函数从 Slack 获取事件并确定它们是否是针对 Starter Bot 的命令。我们的机器人会遇到许多事件类型,但要查找命令,我们只想考虑消息事件。消息事件也有子类型,但我们要查找的命令不会定义任何子类型。该函数通过检查这些属性来过滤掉不感兴趣的事件。现在我们知道该事件表示带有一些文本的消息,但我们想知道文本中是否提到了 Starter Bot。 parse_direct_mention() 函数将计算出以提及开头的消息文本,然后我们将其与我们之前为 Starter Bot 存储的用户 ID 进行比较。如果它们相同,那么我们就知道这是一个 bot 命令,并返回带有频道 ID 的命令文本。

parse_direct_mentions() 函数使用正则表达式来确定是否在消息的开头 提到了用户。它返回用户 ID 和剩余的消息(如果未找到提及,则返回 None, None)。

最后一个函数 handle_command() 是您将来为 Starter Bot 添加所有有趣的命令、幽默和个性的地方。目前,它只有一个示例命令:do。如果命令以已知命令开头,它将有适当的响应。如果不是,则使用默认响应。通过使用通道调用chat.postMessage Web API 方法将响应发送回 Slack。

这是整个程序放在一起时的样子(您也可以在 GitHub 中查看该文件):

import os
import time
import re
from slackclient import SlackClient


# instantiate Slack client
slack_client = SlackClient(os.environ.get('SLACK_BOT_TOKEN'))
# starterbot's user ID in Slack: value is assigned after the bot starts up
starterbot_id = None

# constants
RTM_READ_DELAY = 1 # 1 second delay between reading from RTM
EXAMPLE_COMMAND = "do"
MENTION_REGEX = "^<@(|[WU].+?)>(.*)"

def parse_bot_commands(slack_events):
    """
        Parses a list of events coming from the Slack RTM API to find bot commands.
        If a bot command is found, this function returns a tuple of command and channel.
        If its not found, then this function returns None, None.
    """
    for event in slack_events:
        if event["type"] == "message" and not "subtype" in event:
            user_id, message = parse_direct_mention(event["text"])
            if user_id == starterbot_id:
                return message, event["channel"]
    return None, None

def parse_direct_mention(message_text):
    """
        Finds a direct mention (a mention that is at the beginning) in message text
        and returns the user ID which was mentioned. If there is no direct mention, returns None
    """
    matches = re.search(MENTION_REGEX, message_text)
    # the first group contains the username, the second group contains the remaining message
    return (matches.group(1), matches.group(2).strip()) if matches else (None, None)

def handle_command(command, channel):
    """
        Executes bot command if the command is known
    """
    # Default response is help text for the user
    default_response = "Not sure what you mean. Try *{}*.".format(EXAMPLE_COMMAND)

    # Finds and executes the given command, filling in response
    response = None
    # This is where you start to implement more commands!
    if command.startswith(EXAMPLE_COMMAND):
        response = "Sure...write some more code then I can do that!"

    # Sends the response back to the channel
    slack_client.api_call(
        "chat.postMessage",
        channel=channel,
        text=response or default_response
    )

if __name__ == "__main__":
    if slack_client.rtm_connect(with_team_state=False):
        print("Starter Bot connected and running!")
        # Read bot's user ID by calling Web API method `auth.test`
        starterbot_id = slack_client.api_call("auth.test")["user_id"]
        while True:
            command, channel = parse_bot_commands(slack_client.rtm_read())
            if command:
                handle_command(command, channel)
            time.sleep(RTM_READ_DELAY)
    else:
        print("Connection failed. Exception traceback printed above.")

现在我们所有的代码都已准备就绪,我们可以在命令行上使用 python starterbot.py 命令运行我们的 Starter Bot。

StarterBot 正在运行并连接到 API 时的控制台输出。

在 Slack 中,创建一个新频道并邀请 Starter Bot 或邀请它加入现有频道。

在 Slack 用户界面中创建一个新频道并邀请 StarterBot。

现在开始在您的频道中发出 Starter Bot 命令。

在您的 Slack 频道中给 StarterBot 命令。

补充说明:目前 websocket 包及其使用的 CA 证书存在问题,因此如果您遇到错误喜欢:

...
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1045)
...
slackclient.server.SlackConnectionError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1045)
Connection failed. Exception traceback printed above.

有几件事可以做:1。将 websocket-client 库降级为 0.47.02。或者,下载证书(wget https://www.tbs-certificats.com/issuerdata/DigiCertGlobalRootCA.crt),然后设置环境变量export WEBSOCKET_CLIENT_CA_BUNDLE=DigiCertGlobalRootCA.crt

总结

好了,现在您已经有了一个简单的 Starter Bot,代码中有很多地方可以添加您想要构建的任何功能。

使用 Slack RTM API 和 Python 可以完成更多的事情。查看这些帖子以了解您可以做什么:

  • 附加持久性关系数据库或 NoSQL 后端,例如 PostgreSQL、MySQL 或 SQLite,以保存和检索用户数据
  • 添加另一个通道以通过 SMS 或电话与 bot 交互
  • 集成其他 Web API,例如 GitHub 或 Twilio
  • 探索其他 Slack Platform API 以及您可能会优先使用它们的原因。
  • 使用 Slack Events API 构建入职机器人

有问题吗?通过 Twitter@fullstackpython 或@mattmakai 与我联系。我也在 GitHub 上,用户名是 mattmakai。

看到这篇文章有什么问题了吗?在 GitHub 上创建此页面的源代码并提交拉取请求。

赞(0) 打赏

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏