如何将通过 HTTP POST 从 Python Flask 脚本接收到的数据传递到单独的 Python 脚本进行处理？
How to pass data received via HTTP POST from a Python Flask script to a separate Python script for processing?
好的伙计们，这就是我的问题。
我正在开发一个带有打包机器人的 Slack 应用程序，允许用户在 Slack 中玩游戏。我已经根据 API guidelines. Once I discovered the Interactive Messages 功能成功构建了机器人并将其与应用程序打包在一起，我决定实施该功能，以便对游戏更加人性化。
交互式消息功能允许您 post 带有按钮的消息，用户可以单击这些按钮来调用操作。我的机器人脚本，我们称它为 bot.py，提示用户（使用 Slack chat.postMessage function) with a message containing some buttons from which to choose. This script has one class (I know it should be more modular but all in good time), which opens a web-socket for communication via the Slack RTM API。因此，当脚本运行时，对于来自用户的命令总是 "listening"通道定向如下：@botname command。调用此 "always listening" 状态的脚本部分如下所示：
#bot.py
...
if slack_client.rtm_connect():
print("MYBOT v1.0 connected and running!")
while True:
command, channel, user = self.parse_slack_output(slack_client.rtm_read())
if command and channel:
if channel not in self.channel_ids_to_name.keys():
#this (most likely) means that this channel is a PM with the bot
self.handle_private_message(command, user)
else:
self.handle_command(command, channel, user)
time.sleep(READ_WEBSOCKET_DELAY)
else:
print("Connection failed. Invalid Slack token or bot ID?")
一切都很好。现在，假设用户已使用命令成功创建游戏实例并开始玩游戏。在某个时刻，系统会提示用户输入王牌花色：
#bot.py
...
attachments =[{
"title":"Please select index for trump suit:",
"fallback":"Your interface does not support interactive messages.",
"callback_id":"prompt_trump_suit",
"attachment_type":"default",
"actions":
[{"name":"diamonds","text":":diamonds:","type":"button","value":"0"},
{"name":"clubs","text":":clubs:","type":"button","value":"1"},
{"name":"hearts","text":":hearts:","type":"button","value":"2"},
{"name":"spades","text":":spades:","type":"button","value":"3"}]
}]
slack.chat.post_message(
channel=player_id,
as_user=True,
attachments=attachments
)
互动消息looks like this。单击此消息中的其中一个按钮的操作会通过 HTTP POST 将有效负载发送到 Web 服务器。我在项目中的另一个脚本，我们称之为 app.py，是一个 Flask 脚本，当用户单击其中一个按钮时，它成功地接收到这个 POST 请求。接收 POST 请求的脚本部分如下所示：
#app.py
...
# handles interactive button responses for mybot
@app.route('/actions', methods=['POST'])
def inbound():
payload = request.form.get('payload')
data = json.loads(payload)
token = data['token']
if token == SLACK_VERIFICATION_TOKEN:
print 'TOKEN is good!'
response_url = data['response_url']
channel_info = data['channel']
channel_id = channel_info['id']
user_info = data['user']
user_id = user_info['id']
user_name = user_info['name']
actions = data['actions'][0]
value = actions['value']
print 'User sending message: ',user_name
print "Value received: ",value
return Response(), 200
单击按钮时，我得到了预期的输出：
TOKEN is good!
User sending message: my_username
Value received: 3
所以到目前为止一切都成功了。现在，我想要做的是获取 POST 信息并使用它来调用我的 bot.py 脚本中的一个函数来处理王牌花色选择。问题是，如果我要调用那个函数，让我们称之为 handle_trump_suit_selection()，我首先必须在 app.py 文件中实例化一个 Bot() 对象，这当然不会像需要，因为该函数将使用新的 Bot() 实例调用，因此不会处于与当前游戏相同的状态。
那么我到底如何才能将 POST 信息返回到 bot.py 中所需的 Bot() 实例以进行进一步处理？我是 Python 的 OOP 新手，尤其是 Flask 和 Slack API 的新手，所以请放轻松 ;)。
提前致谢。
大获成功！
tl;博士:
基本上，解决方案是创建一个 Celery 任务，使用 Slack Events API 从 Flask 应用程序实例化机器人实例。您将任务设置为在输入所需输入后启动，立即 return 将所需的 Response(200) 返回给 Slack，同时机器人脚本（启动 RTM API web-套接字）并行启动。
细节：
因此，如上所述，事实证明所需的是 a queuing service of some sort. I ended up going with Celery for its relative ease at integrating with Heroku（我托管 Slack 应用程序的地方）及其易于遵循的文档。
以这种方式开发您的 Slack 应用程序需要设置并使用 Slack Events API 从发布消息的 Slack 频道接收命令（本例中为 "play my_game"）。Flask 应用程序（ app.py) 部分程序侦听此事件，当输入与您要查找的内容匹配时，它会并行启动 Celery 任务（在 tasks.py 中实例化一个 Bot()本例中 bot.py 的实例）。 :) 现在，机器人可以使用 Slack RTM API 和 Slack Events API 进行监听和响应。这允许您在 Slack 框架内构建丰富的 applications/services。
如果您想设置类似的东西，下面是我的项目布局和重要的代码详细信息。随意将它们用作模板。
项目布局：
- project_name_folder
- app_folder
- static_folder
- templates_folder
- __init__.py
- 我的_app.py
- bot.py
- tasks.py
- 过程文件
- requirements.txt
__init__.py:
from celery import Celery
app = Celery('tasks')
import os
app.conf.update(BROKER_URL=os.environ['RABBITMQ_BIGWIG_URL']) # Heroku Celery broker
我的_app.py:
from flask import Flask, request, Response, render_template
import app
from app import tasks
app = Flask(__name__)
@app.route('/events', methods=['POST'])
def events():
"""
Handles the inbound event of a post to the main Slack channel
"""
data = json.loads(request.data)
try:
for k, v in data['event'].iteritems():
ts = data['event']['ts']
channel = data['event']['channel']
user_id = data['event']['user']
team_id = data['team_id']
if 'play my_game' in str(v):
tasks.launch_bot.delay(user_id, channel, ts, team_id) # launch the bot in parallel
return Response(), 200
except Exception as e:
raise
bot.py:
from slackclient import SlackClient
class Bot():
def main():
# opening the Slack web-socket connection
READ_WEBSOCKET_DELAY = 1 # 1 second delay between reading from firehose
if self.slack_client.rtm_connect():
while True:
command, channel, user, ts = self.parse_slack_output()
if command and channel:
if channel not in self.channel_ids_to_name.keys():
# this (most likely) means that this channel is a PM with the bot
self.handle_private_message(command, user, ts)
else:
self.handle_command(command, channel, user, ts)
time.sleep(READ_WEBSOCKET_DELAY)
tasks.py:
import bot
from bot import Bot
from app import app
@app.task
def launch_bot(user_id, channel, ts, team_id):
'''
Instantiates the necessary objects to play a game
Args:
[user_id] (str) The id of the user from which the command was sent
[channel] (str) The channel the command was posted in
[ts] (str) The timestamp of the command
'''
print "launch_bot(user_id,channel)"
app.control.purge()
bot = Bot()
bot.initialize(user_id, channel)
bot.main()
Procfile（如果使用 Heroku）：
web: gunicorn --pythonpath app my_app:app
worker: celery -A app.tasks worker -B --loglevel=DEBUG
如果您有任何问题，请告诉我。这花了我一些时间才弄清楚，如果你正在努力解决这个问题，我很乐意帮助你。
好的伙计们，这就是我的问题。
我正在开发一个带有打包机器人的 Slack 应用程序，允许用户在 Slack 中玩游戏。我已经根据 API guidelines. Once I discovered the Interactive Messages 功能成功构建了机器人并将其与应用程序打包在一起，我决定实施该功能，以便对游戏更加人性化。
交互式消息功能允许您 post 带有按钮的消息，用户可以单击这些按钮来调用操作。我的机器人脚本，我们称它为 bot.py，提示用户（使用 Slack chat.postMessage function) with a message containing some buttons from which to choose. This script has one class (I know it should be more modular but all in good time), which opens a web-socket for communication via the Slack RTM API。因此，当脚本运行时，对于来自用户的命令总是 "listening"通道定向如下：@botname command。调用此 "always listening" 状态的脚本部分如下所示：
#bot.py
...
if slack_client.rtm_connect():
print("MYBOT v1.0 connected and running!")
while True:
command, channel, user = self.parse_slack_output(slack_client.rtm_read())
if command and channel:
if channel not in self.channel_ids_to_name.keys():
#this (most likely) means that this channel is a PM with the bot
self.handle_private_message(command, user)
else:
self.handle_command(command, channel, user)
time.sleep(READ_WEBSOCKET_DELAY)
else:
print("Connection failed. Invalid Slack token or bot ID?")
一切都很好。现在，假设用户已使用命令成功创建游戏实例并开始玩游戏。在某个时刻，系统会提示用户输入王牌花色：
#bot.py
...
attachments =[{
"title":"Please select index for trump suit:",
"fallback":"Your interface does not support interactive messages.",
"callback_id":"prompt_trump_suit",
"attachment_type":"default",
"actions":
[{"name":"diamonds","text":":diamonds:","type":"button","value":"0"},
{"name":"clubs","text":":clubs:","type":"button","value":"1"},
{"name":"hearts","text":":hearts:","type":"button","value":"2"},
{"name":"spades","text":":spades:","type":"button","value":"3"}]
}]
slack.chat.post_message(
channel=player_id,
as_user=True,
attachments=attachments
)
互动消息looks like this。单击此消息中的其中一个按钮的操作会通过 HTTP POST 将有效负载发送到 Web 服务器。我在项目中的另一个脚本，我们称之为 app.py，是一个 Flask 脚本，当用户单击其中一个按钮时，它成功地接收到这个 POST 请求。接收 POST 请求的脚本部分如下所示：
#app.py
...
# handles interactive button responses for mybot
@app.route('/actions', methods=['POST'])
def inbound():
payload = request.form.get('payload')
data = json.loads(payload)
token = data['token']
if token == SLACK_VERIFICATION_TOKEN:
print 'TOKEN is good!'
response_url = data['response_url']
channel_info = data['channel']
channel_id = channel_info['id']
user_info = data['user']
user_id = user_info['id']
user_name = user_info['name']
actions = data['actions'][0]
value = actions['value']
print 'User sending message: ',user_name
print "Value received: ",value
return Response(), 200
单击按钮时，我得到了预期的输出：
TOKEN is good!
User sending message: my_username
Value received: 3
所以到目前为止一切都成功了。现在，我想要做的是获取 POST 信息并使用它来调用我的 bot.py 脚本中的一个函数来处理王牌花色选择。问题是，如果我要调用那个函数，让我们称之为 handle_trump_suit_selection()，我首先必须在 app.py 文件中实例化一个 Bot() 对象，这当然不会像需要，因为该函数将使用新的 Bot() 实例调用，因此不会处于与当前游戏相同的状态。
那么我到底如何才能将 POST 信息返回到 bot.py 中所需的 Bot() 实例以进行进一步处理？我是 Python 的 OOP 新手，尤其是 Flask 和 Slack API 的新手，所以请放轻松 ;)。
提前致谢。
大获成功！
tl;博士: 基本上，解决方案是创建一个 Celery 任务，使用 Slack Events API 从 Flask 应用程序实例化机器人实例。您将任务设置为在输入所需输入后启动，立即 return 将所需的 Response(200) 返回给 Slack，同时机器人脚本（启动 RTM API web-套接字）并行启动。
细节： 因此，如上所述，事实证明所需的是 a queuing service of some sort. I ended up going with Celery for its relative ease at integrating with Heroku（我托管 Slack 应用程序的地方）及其易于遵循的文档。
以这种方式开发您的 Slack 应用程序需要设置并使用 Slack Events API 从发布消息的 Slack 频道接收命令（本例中为 "play my_game"）。Flask 应用程序（ app.py) 部分程序侦听此事件，当输入与您要查找的内容匹配时，它会并行启动 Celery 任务（在 tasks.py 中实例化一个 Bot()本例中 bot.py 的实例）。 :) 现在，机器人可以使用 Slack RTM API 和 Slack Events API 进行监听和响应。这允许您在 Slack 框架内构建丰富的 applications/services。
如果您想设置类似的东西，下面是我的项目布局和重要的代码详细信息。随意将它们用作模板。
项目布局：
- project_name_folder
- app_folder
- static_folder
- templates_folder
- __init__.py
- 我的_app.py
- bot.py
- tasks.py
- 过程文件
- requirements.txt
- app_folder
__init__.py:
from celery import Celery
app = Celery('tasks')
import os
app.conf.update(BROKER_URL=os.environ['RABBITMQ_BIGWIG_URL']) # Heroku Celery broker
我的_app.py:
from flask import Flask, request, Response, render_template
import app
from app import tasks
app = Flask(__name__)
@app.route('/events', methods=['POST'])
def events():
"""
Handles the inbound event of a post to the main Slack channel
"""
data = json.loads(request.data)
try:
for k, v in data['event'].iteritems():
ts = data['event']['ts']
channel = data['event']['channel']
user_id = data['event']['user']
team_id = data['team_id']
if 'play my_game' in str(v):
tasks.launch_bot.delay(user_id, channel, ts, team_id) # launch the bot in parallel
return Response(), 200
except Exception as e:
raise
bot.py:
from slackclient import SlackClient
class Bot():
def main():
# opening the Slack web-socket connection
READ_WEBSOCKET_DELAY = 1 # 1 second delay between reading from firehose
if self.slack_client.rtm_connect():
while True:
command, channel, user, ts = self.parse_slack_output()
if command and channel:
if channel not in self.channel_ids_to_name.keys():
# this (most likely) means that this channel is a PM with the bot
self.handle_private_message(command, user, ts)
else:
self.handle_command(command, channel, user, ts)
time.sleep(READ_WEBSOCKET_DELAY)
tasks.py:
import bot
from bot import Bot
from app import app
@app.task
def launch_bot(user_id, channel, ts, team_id):
'''
Instantiates the necessary objects to play a game
Args:
[user_id] (str) The id of the user from which the command was sent
[channel] (str) The channel the command was posted in
[ts] (str) The timestamp of the command
'''
print "launch_bot(user_id,channel)"
app.control.purge()
bot = Bot()
bot.initialize(user_id, channel)
bot.main()
Procfile（如果使用 Heroku）：
web: gunicorn --pythonpath app my_app:app
worker: celery -A app.tasks worker -B --loglevel=DEBUG
如果您有任何问题，请告诉我。这花了我一些时间才弄清楚，如果你正在努力解决这个问题，我很乐意帮助你。