如何在Ubuntu20.04上用Python构建Slackbot
作为 Write for DOnations 计划的一部分,作者选择了 Tech Education Fund 来接受捐赠。
介绍
Slack 是一个专为提高工作效率而设计的交流平台。 它包括直接消息传递、公共和私人频道、语音和视频通话以及机器人集成等功能。 Slackbot 是一个自动化程序,可以在 Slack 中执行各种功能,从发送消息到触发任务再到对某些事件发出警报。
在本教程中,您将使用 Python 编程语言构建一个 Slackbot。 Python 是一种流行的语言,以简单性和可读性而自豪。 Slack 提供了丰富的 Python Slack API 用于与 Slack 集成以执行常见任务,例如发送消息、向消息添加表情符号等等。 Slack 还提供 Python Slack Events API 用于与 Slack 中的事件集成,允许您对消息和提及等事件执行操作。
作为一个有趣的概念验证,它将展示 Python 及其 Slack API 的强大功能,您将构建一个 CoinBot
- 一个监控通道并在触发时为您抛硬币的 Slackbot。 然后你可以修改你的 CoinBot
来完成任意数量的 slightly 更实际的应用。
请注意,本教程使用 Python 3,与 Python 2 不兼容。
先决条件
为了遵循本指南,您需要:
- 您可以在其中安装应用程序的 Slack 工作区。 如果您创建了工作区,您将拥有此能力。 如果您还没有,可以在 Slack 网站 上创建一个。
- (可选)用于开发的具有公共 IP 地址的服务器或计算机。 我们建议全新安装 Ubuntu 20.04,使用具有
sudo
权限的非 root 用户并启用 SSH。 您可以按照本指南来初始化您的服务器并完成这些步骤。
您可能希望在具有公共 IP 地址的服务器上测试本教程。 Slack 需要能够向您的机器人发送消息等事件。 如果您在本地计算机上进行测试,则需要将转发流量通过防火墙端口传输到本地系统。 如果您正在寻找一种在云服务器上进行开发的方法,请查看本教程 如何通过 Remote-SSH 插件使用 Visual Studio Code 进行远程开发。
第 1 步 — 在 Slack UI 中创建 Slackbot
首先在 Slack API 控制面板中创建您的 Slack 应用程序。 通过网络浏览器登录到您在 Slack 中的工作区并导航到 API 控制面板。 现在单击 创建应用程序 按钮。
接下来,系统将提示您输入应用程序的名称并选择开发 Slack 工作区。 对于本教程,将您的应用程序命名为 CoinBot
并选择您拥有管理员访问权限的工作区。 完成此操作后,单击 创建应用程序 按钮。
创建应用后,您将看到以下默认应用仪表板。 此仪表板是您通过设置权限、订阅事件、将应用程序安装到工作区等来管理应用程序的地方。
为了让您的应用能够向频道发布消息,您需要授予应用发送消息的权限。 为此,请单击控制面板中的 Permissions 按钮。
当你到达 OAuth 和权限页面,向下滚动,直到找到范围页面的部分。 然后在范围中找到 Bot Token Scopes 子部分,然后单击 Add an OAuth Scope 按钮。
单击该按钮,然后键入 chat:write
。 选择该权限以将其添加到您的机器人。 这将允许应用程序将消息发布到它可以访问的频道。 有关可用权限的更多信息,请参阅 Slack 的文档 。
现在您已经添加了适当的权限,是时候将您的应用程序安装到您的 Slack 工作区中了。 向上滚动 OAuth 和权限页面,然后单击将应用程序安装到工作区顶部的按钮。
单击此按钮并查看应用程序可以在频道中执行的操作。 满意后,单击 Allow 按钮完成安装。
安装机器人后,您将看到一个 Bot 用户 OAuth 访问令牌 ,供您的应用在尝试在工作区中执行操作时使用。 继续复制此令牌; 你稍后会需要它。
最后,将新安装的机器人添加到工作区中的通道中。 如果您尚未创建通道,则可以使用默认在 Slack 工作区中创建的 #general 通道。 在 Slack 客户端导航栏的 Apps 部分中找到该应用程序,然后单击它。 完成后,打开右上角的 Details 菜单。 如果您的 Slack 客户端没有全屏显示,它将看起来像一个圆圈中的 i
。
要完成将您的应用添加到频道,请单击详细信息页面中由三个点表示的更多按钮,然后选择将此应用添加到频道...。 在出现的模式中输入您的频道,然后单击 Add。
您现在已成功创建应用并将其添加到 Slack 工作区中的频道。 为您的应用编写代码后,它将能够在该频道中发布消息。 在下一节中,您将开始编写支持 CoinBot
的 Python 代码。
第 2 步 — 设置您的 Python 开发人员环境
首先让我们设置您的 Python 环境,以便您可以开发 Slackbot。
打开终端并将 python3
和相关工具安装到您的系统上:
sudo apt install python3 python3-venv
接下来,您将创建一个虚拟环境,以将您的 Python 包与 Python 的系统安装隔离开来。 为此,首先创建一个目录,您将在其中创建虚拟环境。 在 ~/.venvs
创建一个新目录:
mkdir ~/.venvs
现在创建您的 Python 虚拟环境:
python3 -m venv ~/.venvs/slackbot
接下来,激活您的虚拟环境,以便您可以使用其 Python 安装和安装包:
source ~/.venvs/slackbot/bin/activate
您的 shell 提示符现在将在括号中显示虚拟环境。 它看起来像这样:
现在使用 pip
将必要的 Python 包安装到您的虚拟环境中:
pip install slackclient slackeventsapi Flask
slackclient
和 slackeventsapi
促进了 Python 与 Slack 的 API 的交互。 Flask
是一个流行的微型 Web 框架,您将使用它来部署您的应用程序:
现在您已经设置了开发人员环境,您可以开始编写 Python Slackbot:
第 3 步 — 在 Python 中创建 Slackbot 消息类
Slack 中的消息通过 特定格式的 JSON 有效负载 发送。 这是您的 Slackbot 将制作并作为消息发送的 JSON 示例:
{ "channel":"channel", "blocks":[ { "type":"section", "text":{ "type":"mrkdwn", "text":"Sure! Flipping a coin....\n\n" } }, { "type":"section", "text":{ "type":"mrkdwn", "text":"*flips coin* The result is Tails." } } ] }
您可以手动制作这个 JSON 并发送它,但是让我们构建一个 Python 类,它不仅制作这个有效负载,而且还模拟硬币翻转。
首先使用 touch
命令创建一个名为 coinbot.py
的文件:
touch coinbot.py
接下来,使用 nano
或您喜欢的文本编辑器打开此文件:
nano coinbot.py
现在添加以下代码行来为您的应用导入相关库。 此类所需的唯一库是 Python 标准库中的 random
库。 这个库将允许我们模拟硬币翻转。
将以下行添加到 coinbot.py
以导入所有必需的库:
硬币机器人.py
# import the random library to help us generate the random numbers import random
接下来,创建您的 CoinBot
类和此类的一个实例来制作消息负载。 将以下行添加到 coinbot.py
以创建 CoinBot
类:
硬币机器人.py
... class CoinBot:
现在缩进一并创建类所需的常量、构造函数和方法。 首先让我们创建一个常量来保存消息负载的基础。 本节指定此常量属于节类型,并且文本通过 markdown 格式化。 它还指定您希望显示的文本。 您可以在 官方 Slack 消息负载文档 中阅读有关不同负载选项的更多信息。
将以下行附加到 coinbot.py
以创建有效负载的基本模板:
硬币机器人.py
... # Create a constant that contains the default text for the message COIN_BLOCK = { "type": "section", "text": { "type": "mrkdwn", "text": ( "Sure! Flipping a coin....\n\n" ), }, }
接下来为您的类创建一个构造函数,以便您可以为每个请求创建一个单独的机器人实例。 不用担心这里的内存开销; 一旦不再需要这些实例,Python 垃圾收集器就会清理它们。 此代码根据传递给构造函数的参数设置接收通道。
将以下行追加到 coinbot.py
以创建构造函数:
硬币机器人.py
... # The constructor for the class. It takes the channel name as the a # parameter and sets it as an instance variable. def __init__(self, channel): self.channel = channel
现在编写模拟掷硬币的代码。 我们将随机生成一个或零,分别代表正面或反面。
将以下行附加到 coinbot.py
以模拟硬币翻转并返回精心制作的有效载荷:
硬币机器人.py
... # Generate a random number to simulate flipping a coin. Then return the # crafted slack payload with the coin flip message. def _flip_coin(self): rand_int = random.randint(0,1) if rand_int == 0: results = "Heads" else: results = "Tails" text = f"The result is {results}" return {"type": "section", "text": {"type": "mrkdwn", "text": text}},
最后,通过调用您的 _flip_coin
方法,创建一个制作并返回整个消息负载的方法,包括来自您的构造函数的数据。
将以下行附加到 coinbot.py
以创建将生成完成的有效负载的方法:
硬币机器人.py
... # Craft and return the entire message payload as a dictionary. def get_message_payload(self): return { "channel": self.channel, "blocks": [ self.COIN_BLOCK, *self._flip_coin(), ], }
您现在已经完成了 CoinBot
类,可以进行测试了。 在继续之前,请确认您完成的文件 coinbot.py
包含以下内容:
硬币机器人.py
# import the random library to help us generate the random numbers import random # Create the CoinBot Class class CoinBot: # Create a constant that contains the default text for the message COIN_BLOCK = { "type": "section", "text": { "type": "mrkdwn", "text": ( "Sure! Flipping a coin....\n\n" ), }, } # The constructor for the class. It takes the channel name as the a # parameter and then sets it as an instance variable def __init__(self, channel): self.channel = channel # Generate a random number to simulate flipping a coin. Then return the # crafted slack payload with the coin flip message. def _flip_coin(self): rand_int = random.randint(0,1) if rand_int == 0: results = "Heads" else: results = "Tails" text = f"The result is {results}" return {"type": "section", "text": {"type": "mrkdwn", "text": text}}, # Craft and return the entire message payload as a dictionary. def get_message_payload(self): return { "channel": self.channel, "blocks": [ self.COIN_BLOCK, *self._flip_coin(), ], }
保存并关闭文件。
现在您已经准备好为您的 Slackbot 完成工作的 Python 类,让我们确保该类生成有用的消息有效负载并且您可以将其发送到您的工作区。
第四步——测试你的信息
现在让我们测试这个类是否产生了正确的有效载荷。 创建一个名为 coinbot_test.py
的文件:
nano coinbot_test.py
现在添加以下代码。 一定要在coinbot类的实例化中更改通道名称 coin_bot = coinbot("#YOUR_CHANNEL_HERE")。 此代码将在 Python 中创建一个 Slack 客户端,该客户端将向您指定的通道发送一条消息,表明您已将应用程序安装到:
coinbot_test.py
from slack import WebClient from coinbot import CoinBot import os # Create a slack client slack_web_client = WebClient(token=os.environ.get("SLACK_TOKEN")) # Get a new CoinBot coin_bot = CoinBot("#YOUR_CHANNEL_HERE") # Get the onboarding message payload message = coin_bot.get_message_payload() # Post the onboarding message in Slack slack_web_client.chat_postMessage(**message)
保存并关闭文件。
在运行此文件之前,您需要将在步骤 1 中保存的 Slack 令牌导出为环境变量:
export SLACK_TOKEN="your_bot_user_token"
现在测试此文件并通过在终端中运行以下脚本来验证有效负载是否已生成和发送。 确保您的虚拟环境已激活。 您可以通过查看 bash 提示符前面的 (slackbot)
文本来验证这一点。 运行此命令,您将收到来自 Slackbot 的消息,其中包含掷硬币的结果:
python coinbot_test.py
检查您安装应用程序的频道并验证您的机器人确实发送了硬币翻转消息。 你的结果将是正面或反面。
现在您已经验证了您的 Slackbot 可以掷硬币、创建消息并传递消息,让我们创建一个 Flask 来永久运行此应用程序并使其模拟硬币翻转并随时分享结果它会在频道中发送的消息中看到某些文本。
第 5 步——创建一个 Flask 应用程序来运行你的 Slackbot
既然您有一个可以向 Slack 工作区发送消息的正常运行的应用程序,您需要创建一个长时间运行的进程,以便您的机器人可以收听频道中发送的消息,并在文本符合特定条件时回复它们。 您将使用 Python Web 框架 Flask 来运行此过程并侦听频道中的事件。
在本节中,您将从具有公共 IP 地址的服务器运行 Flask 应用程序,以便 Slack API 可以向您发送事件。 如果您在个人工作站上本地运行此程序,则需要将端口从个人防火墙转发到将在您的工作站上运行的端口。 这些端口可以相同,本教程将设置使用端口3000
。
首先调整您的防火墙设置以允许通过端口 3000
的流量:
sudo ufw allow 3000
现在检查 ufw
的状态:
sudo ufw status
你会看到这样的输出:
OutputStatus: active To Action From -- ------ ---- OpenSSH ALLOW Anywhere 3000 ALLOW Anywhere OpenSSH (v6) ALLOW Anywhere (v6) 3000 (v6) ALLOW Anywhere (v6)
现在为您的 Flask 应用程序创建文件。 将此文件命名为 app.py
:
touch app.py
接下来,在您喜欢的文本编辑器中打开此文件:
nano app.py
现在添加以下导入 statements
。 由于以下原因,您将导入以下库:
import os
- 访问环境变量import logging
- 记录应用程序的事件from flask import Flask
- 创建 Flask 应用from slack import WebClient
- 通过 Slack 发送消息from slackeventsapi import SlackEventAdapter
- 从 Slack 接收事件并处理它们from coinbot import CoinBot
- 创建您的 CoinBot 实例并生成消息负载。
将以下行追加到 app.py
以导入所有必要的库:
应用程序.py
import os import logging from flask import Flask from slack import WebClient from slackeventsapi import SlackEventAdapter from coinbot import CoinBot
现在创建您的 Flask 应用程序并在 /slack/events
端点向您的 Slack 应用程序注册一个 Slack 事件适配器。 这将在您的 Slack 应用程序中创建一条路由,将发送和摄取 Slack 事件。 为此,您需要从您的 Slack 应用程序中获取另一个令牌,您将在本教程的后面部分执行此操作。 获得此变量后,您将其导出为名为 SLACK_EVENTS_TOKEN
的环境变量。 继续编写代码以在创建 SlackEventAdapter
时将其读入,即使您尚未设置令牌。
将以下行添加到 app.py
以创建 Flask 应用程序并将事件适配器注册到此应用程序中:
应用程序.py
... # Initialize a Flask app to host the events adapter app = Flask(__name__) # Create an events adapter and register it to an endpoint in the slack app for event ingestion. slack_events_adapter = SlackEventAdapter(os.environ.get("SLACK_EVENTS_TOKEN"), "/slack/events", app)
接下来创建一个 Web 客户端对象,该对象将允许您的应用程序在工作区中执行操作,特别是发送消息。 这与您之前测试 coinbot.py
文件时所做的类似。
将以下行添加到 app.py
以创建此 slack_web_client
:
应用程序.py
... # Initialize a Web API client slack_web_client = WebClient(token=os.environ.get("SLACK_TOKEN"))
现在创建一个可调用的函数,该函数将创建 CoinBot
的实例,然后使用此实例创建消息负载并将消息负载传递给 Slack Web 客户端进行传递。 该函数将采用单个参数 channel
,它将指定接收消息的通道。
将以下行添加到 app.py
以创建此函数:
应用程序.py
... def flip_coin(channel): """Craft the CoinBot, flip the coin and send the message to the channel """ # Create a new CoinBot coin_bot = CoinBot(channel) # Get the onboarding message payload message = coin_bot.get_message_payload() # Post the onboarding message in Slack slack_web_client.chat_postMessage(**message)
现在您已经创建了一个函数来处理应用程序的消息传递方面,创建一个监视 Slack 事件以执行特定操作的函数,然后执行您的机器人。 您将配置您的应用程序,使其在看到短语“Hey Sammy,Flip a coin”时响应模拟硬币翻转的结果。 您将接受此版本的任何版本 - 案例不会阻止应用程序响应。
首先用 @slack_events_adapter.on
语法装饰你的函数,允许你的函数接收事件。 指定您只需要 message
事件并让您的函数接受包含所有必要 Slack 信息的有效负载参数。 获得此有效负载后,您将解析出文本并对其进行分析。 然后,如果它收到激活短语,您的应用程序将发送模拟硬币翻转的结果。
将以下代码附加到 app.py
以接收、分析和处理传入消息:
应用程序.py
# When a 'message' event is detected by the events adapter, forward that payload # to this function. @slack_events_adapter.on("message") def message(payload): """Parse the message event, and if the activation string is in the text, simulate a coin flip and send the result. """ # Get the event data from the payload event = payload.get("event", {}) # Get the text from the event that came through text = event.get("text") # Check and see if the activation phrase was in the text of the message. # If so, execute the code to flip a coin. if "hey sammy, flip a coin" in text.lower(): # Since the activation phrase was met, get the channel ID that the event # was executed on channel_id = event.get("channel") # Execute the flip_coin function and send the results of # flipping a coin to the channel return flip_coin(channel_id)
最后,创建一个 main
部分,该部分将创建一个记录器,以便您可以查看应用程序的内部结构,并在端口 3000
上的外部 IP 地址上启动应用程序。 为了从 Slack 获取事件,例如发送新消息时,您必须在面向公众的 IP 地址上测试您的应用程序。
将以下行附加到 app.py
以设置您的主要部分:
应用程序.py
if __name__ == "__main__": # Create the logging object logger = logging.getLogger() # Set the log level to DEBUG. This will increase verbosity of logging messages logger.setLevel(logging.DEBUG) # Add the StreamHandler as a logging handler logger.addHandler(logging.StreamHandler()) # Run your app on your externally facing IP address on port 3000 instead of # running it on localhost, which is traditional for development. app.run(host='0.0.0.0', port=3000)
您现在已经完成了 Flask 应用程序,可以进行测试了。 在继续验证您完成的文件之前,app.py
包含以下内容:
应用程序.py
import os import logging from flask import Flask from slack import WebClient from slackeventsapi import SlackEventAdapter from coinbot import CoinBot # Initialize a Flask app to host the events adapter app = Flask(__name__) # Create an events adapter and register it to an endpoint in the slack app for event injestion. slack_events_adapter = SlackEventAdapter(os.environ.get("SLACK_EVENTS_TOKEN"), "/slack/events", app) # Initialize a Web API client slack_web_client = WebClient(token=os.environ.get("SLACK_TOKEN")) def flip_coin(channel): """Craft the CoinBot, flip the coin and send the message to the channel """ # Create a new CoinBot coin_bot = CoinBot(channel) # Get the onboarding message payload message = coin_bot.get_message_payload() # Post the onboarding message in Slack slack_web_client.chat_postMessage(**message) # When a 'message' event is detected by the events adapter, forward that payload # to this function. @slack_events_adapter.on("message") def message(payload): """Parse the message event, and if the activation string is in the text, simulate a coin flip and send the result. """ # Get the event data from the payload event = payload.get("event", {}) # Get the text from the event that came through text = event.get("text") # Check and see if the activation phrase was in the text of the message. # If so, execute the code to flip a coin. if "hey sammy, flip a coin" in text.lower(): # Since the activation phrase was met, get the channel ID that the event # was executed on channel_id = event.get("channel") # Execute the flip_coin function and send the results of # flipping a coin to the channel return flip_coin(channel_id) if __name__ == "__main__": # Create the logging object logger = logging.getLogger() # Set the log level to DEBUG. This will increase verbosity of logging messages logger.setLevel(logging.DEBUG) # Add the StreamHandler as a logging handler logger.addHandler(logging.StreamHandler()) # Run our app on our externally facing IP address on port 3000 instead of # running it on localhost, which is traditional for development. app.run(host='0.0.0.0', port=3000)
保存并关闭文件。
现在您的 Flask 应用程序已准备好为您的应用程序提供服务,让我们对其进行测试。
第 6 步 — 运行 Flask 应用程序
最后,将所有内容组合在一起并执行您的应用程序。
首先,将正在运行的应用程序添加为 Slackbot 的授权处理程序。
在 Slack UI 中导航到应用程序的 基本信息 部分。 向下滚动,直到找到 App Credentials 部分。
复制 Signing Secret 并将其导出为环境变量 SLACK_EVENTS_TOKEN
:
export SLACK_EVENTS_TOKEN="MY_SIGNING_SECRET_TOKEN"
有了它,您就拥有了运行应用程序所需的所有 API 令牌。 如果您需要重新了解如何导出 SLACK_TOKEN
,请参阅步骤 1。 现在您可以启动您的应用程序并验证它是否确实在运行。 确保您的虚拟环境已激活并运行以下命令来启动您的 Flask 应用程序:
python3 app.py
你会看到这样的输出:
(slackbot) [20:04:03] sammy:coinbot$ python app.py * Serving Flask app "app" (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: off * Running on http://0.0.0.0:3000/ (Press CTRL+C to quit)
要验证您的应用程序是否已启动,请打开一个新的终端窗口并 curl
使用正确端口的服务器 IP 地址 /slack/events
:
curl http://YOUR_IP_ADDRESS:3000/slack/events
curl
将返回以下内容:
OutputThese are not the slackbots you're looking for.
收到消息 These are not the slackbots you're looking for.
,表明您的应用程序已启动并正在运行。
现在,在 Slack UI 中完成应用程序的配置后,让这个 Flask 应用程序继续运行。
首先授予您的应用适当的权限,以便它可以收听消息并做出相应的响应。 单击 UI 侧栏中的 Event Subscriptions 并切换 Enable Events 单选按钮。
完成后,在 Request URL 字段中输入您的 IP 地址、端口和 /slack/events
端点。 不要忘记 HTTP
协议前缀。 Slack 将尝试连接到您的端点。 成功完成后,您将看到一个绿色复选标记,旁边带有 Verified 字样。
接下来,展开 Subscribe to bot events 并将 message.channels
权限添加到您的应用程序。 这将允许您的应用从您的频道接收消息并进行处理。
完成此操作后,您将在 Subscribe to bot events 部分中看到该事件。 接下来单击右下角的绿色 Save Changes 按钮。
执行此操作后,您会在屏幕顶部看到一个黄色横幅,通知您需要重新安装应用程序才能应用以下更改。 每次更改权限时,您都需要重新安装您的应用程序。 单击此横幅中的 重新安装您的应用程序 链接以重新安装您的应用程序。
您将看到一个确认屏幕,其中总结了您的机器人将拥有的权限,并询问您是否要允许其安装。 单击绿色的 Allow 按钮完成安装过程。
现在你已经完成了你的应用程序应该准备好了。 回到您安装 CoinBot
的频道并发送一条消息,其中包含短语 Hey Sammy,Flip a coin。 您的机器人将掷硬币并回复结果。 恭喜! 您已经创建了一个 Slackbot!
结论
完成应用程序的开发并准备将其移至生产环境后,您需要将其部署到服务器。 这是必要的,因为 Flask 开发服务器不是一个安全的生产环境。 如果您使用 WSGI 部署您的应用程序,甚至可能保护域名并为您的服务器提供 DNS 记录,您会得到更好的服务。 部署 Flask 应用程序有很多选择,下面列出了其中的一些:
- 使用 Gunicorn 和 Nginx 将 Flask 应用程序部署到 Ubuntu 20.04
- 使用 uWSGI 和 Nginx 将 Flask 应用程序部署到 Ubuntu 20.04
- 在 Ubuntu 18.04 上使用 Docker 部署 Flask 应用程序
除了这些之外,还有更多方法可以部署您的应用程序。 与往常一样,在部署和基础设施方面,做最适合 you 的事情。
无论如何,你现在有一个 Slackbot,你可以用它来抛硬币来帮助你做决定,比如午餐吃什么。
您还可以获取此基本代码并对其进行修改以满足您的需求,无论是自动化支持、资源管理、猫的图片还是您能想到的任何内容。 您可以在 此处 查看完整的 Python Slack API 文档。