用Python构建Slack Bot
作者:互联网
设置环境
虽然我们现在意识到获得编码所需的工具和软件,但我们还需要通过进入命令行并更改保存项目文件夹的目录来设置我们的开发环境。然后,我们需要创建一个虚拟环境,通过将以下代码输入到我们的命令行,将我们的应用程序依赖项与其他项目分开。
virtualenv slackbot
然后,我们需要通过输入以下代码来激活虚拟环境:
source slackbot/bin/activate
In addition the slackclient
API helper library by slack can then send and receive messages from a slack channel. By using the following code below, we will then need to install the slackclient library using the pip
command below:
pip3 install slackclient
该过程的下一步是创建一个Slack应用程序来接收机器人的API令牌。然后,您需要选择一个app name
,如下所示:
下一步是选择一个工作区(如果您已经登录了一个工作区,则为一个工作区,或者创建一个新的工作区,如下所示:
API和配置
slack bot的整个想法是确保它像参与DM、渠道和群组对话的任何其他用户一样运行。这被称为abot用户,可以通过在“功能”部分下选择“机器人用户”来设置。以下是添加机器人用户所涉及的几个步骤。
- 第一步是单击已创建的工作区
- 下一步是进入导航部分的应用程序主页,然后单击
review scopes to add
按钮。 - 单击
app home
后,接下来是转到机器人页面获取令牌并开始配置API。
按照上述步骤操作将进入如下所示的页面:
如上所示,下一步是在屏幕右下角显示的按钮中生成令牌。通常的做法是将秘密令牌导出为environment variables
,如下所示,名称为SLACK_TOKEN
:
export SLACK_TOKEN='your bot user access token here'
我们现在已经授权了Slack RTM和Web API。
机器人编码
在您的新文件夹中,创建一个名为bot.py
的新文件,并输入以下代码:
import os
import time
import re
from slackclient import SlackClient
我们的依赖项现已导入,我们现在可以获取与环境变量关联的值,然后表示Slack客户端,如以下代码所示:
# represent the Slack client
slack_client = SlackClient(os.environ.get('SLACK_TOKEN'))
# bot's user ID in Slack: value is assigned after the bot starts up
bot_id = None
# constants
RTM_READ_DELAY = 1 # 1 second delay between reading from RTM
EXAMPLE_COMMAND = "do"
MENTION_REGEX = "^<@(|[WU].+?)>(.*)"
上面的代码现在用SLACK_TOKEN
表示SlackClient,然后将其导出为环境变量。这现在传达了我们可以使用变量来存储机器人的Slack用户ID。然后,我们可以解释以下代码中声明的常量:
if __name__ == "__main__":
if slack_client.rtm_connect(with_team_state=False):
print("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,该API现在调用我们的Web API方法(auth.test
)来获取机器人的ID。
机器人用户现在拥有安装Slack应用程序的每个工作区的用户ID。另一件需要注意的点是,程序现在进入一个无限循环,每次循环运行客户端时,它都会收到来自Slack的RTM API的任何事件。仔细注意,我们会看到,在循环结束之前,程序会暂停一秒钟,例如它不会快速循环,也不会浪费我们的CPU时间。
对于每个已读取的事件,parse_bot_comands()
函数现在确保该事件包含机器人的命令。这也将确保command
将包含一个值,同时确保handle command()
函数知道该如何处理该命令。
我们现在在下面的代码中解释了上述内容:
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 == bot_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_command()
函数现在接受这些事件,并确保它们指向机器人。虽然我们的机器人会遇到几种事件类型,但我们需要使用message events
找到我们想要的命令。
消息事件通常有sub-types
但在我们将找到的命令中,不会定义sub-types
。这些函数现在通过检查属性来过滤not-so-great
的事件。
正在考虑的下一步是想知道事件是否代表信息,但也想知道是否提到了机器人。
然后,我们的parse_direct_mention()
函数将自行承担任务,确定文本是否以提及开头,然后将其与我们存储在Bot中的用户ID进行比较,如果该用户ID相同,我们知道是机器人命令,然后在通道ID中返回命令文本。
parse_direct_mentions()
函数将使用一个表达式来确定用户是否在消息中是否被提及。使用它,如果没有提到,它会将用户ID带回给我们,并且none
消息。
我们最后一个函数handle_command()
是将来参考添加组成松弛机器人的地方。它带有一个已知的命令,通过使用chat.postMessage
回调我们的Web API向Slack发送响应。
下面的代码显示了解释
import os
import time
import re
from slackclient import SlackClient
# instantiate Slack client
slack_client = SlackClient(os.environ.get('SLACK_BOT_TOKEN'))
# slack user ID in Slack: value is assigned after the bot starts up
slack_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.")
我们现在可以使用以下代码运行代码:
python3 starterbot.py
下一步是创建一个频道,如下所示:
您现在可以开始向您的机器人发出命令。