我需要在 Tornado 服務器中為 Dialogflow 創建一個 webhook。我以前用燒瓶做過這個。我是他們兩個的初學者。Flask Flask webhook代碼的代碼：from flask import render_templateimport osfrom flask import Flaskfrom flask import requestfrom flask import make_responseimport jsonimport timeapp = Flask(__name__)@app.route("/")def index(): return render_template("index.html")@app.route('/webhook', methods=['POST', 'GET'])def webhook(): req = request.get_json(silent=True, force=True) print(json.dumps(req, indent=4)) res = makeWebhookResult(req) res = json.dumps(res, indent=4) print("+++++++++++++++RESPONSE+++++++++++++++++", res) r = make_response(res) r.headers['Content-Type'] = 'application/json' return r# Right now I'm just printing a response to see if it works properlydef makeWebhookResult(req): queryResult = req.get('queryResult').get("queryText") speech = queryResult return { "fulfillmentText": 'YOLO', "source": 'App' }#./ngrok http 8090if __name__ == '__main__': port = int(os.getenv('PORT', 8090)) print("Starting app on port %d" % (port)) app.run(debug=True, port=port, host='localhost')現在我像這樣在龍卷風中試過這個：import tornado.ioloopimport tornado.web as webimport tornadoimport jsonimport osstatic_root = os.path.join(os.path.dirname(__file__), 'static')class MainHandler(tornado.web.RequestHandler): def get(self): template = "./templates/index.html" self.render(template)它適用于 Flask。當我嘗試通過龍卷風運行它（使用 ngrok 進行隧道傳輸）時，我收到警告：tornado.access:405 POST /webhook (127.0.0.1) 0.83ms我閱讀了龍卷風的文檔，但我似乎仍然無法弄清楚我該怎么做。我假設問題出在 Webhook 類中。我在這里做錯了什么？
1 回答
jeck貓
TA貢獻1909條經驗 獲得超7個贊
警告顯示POST請求有問題 - 并處理POST您需要def post()的類方法Webhook。
它應該post()代替prepare()（這是為了不同的東西）。
你可以用self.write(dictionary)它來發送'application/json'
class Webhook(tornado.web.RequestHandler):
def post(self):
if self.request.headers.get("Content-Type", "").startswith("application/json"):
data_input = json.loads(self.request.body)
print('data_input:', data_input)
print('data_input json.dumps:', json.dumps(data_input, indent=4))
data_output = self.webhook_result(data_input) # get as normal dict, not string
print('data_output:', data_output)
print('data_output json.dumps:', json.dumps(data_output, indent=4))
self.write(data_output) # it will send as JSON
else:
self.write({'error': 'Wrong Content-Type'}) # it will send as JSON
順便說一句：如果你發送值到webhook_result()那么你可以獲得這個值 - 即 as data- 并使用它而不是self.req
def webhook_result(self, data):
speech = data.get('queryResult').get("queryText")
print('speech:', speech)
return {
"fulfillmentText": 'YOLO',
"source": 'App'
}
我測試的代碼
import tornado
import tornado.web
import json
import os
static_root = os.path.join(os.path.dirname('.'), 'static')
class MainHandler(tornado.web.RequestHandler):
def get(self):
#self.render("./templates/index.html")
# to test POST request but with wrong Content-Type
self.write('''<form action="/webhook" method="POST"><button>SUBMIT</button></form>''')
class Webhook(tornado.web.RequestHandler):
def post(self):
if self.request.headers.get("Content-Type", "").startswith("application/json"):
data_input = json.loads(self.request.body)
print('data_input:', data_input)
print('data_input json.dumps:', json.dumps(data_input, indent=4))
data_output = self.webhook_result(data_input) # get as normal dict, not string
print('data_output:', data_output)
print('data_output json.dumps:', json.dumps(data_output, indent=4))
self.write(data_output) # it will send as JSON
else:
self.write({'error': 'Wrong Content-Type'}) # it will send as JSON
def webhook_result(self, data):
speech = data.get('queryResult').get("queryText")
print('speech:', speech)
return {
"fulfillmentText": 'YOLO',
"source": 'App'
}
handlers = [
(r'/', MainHandler),
(r'/webhook', Webhook),
# probably it should be as last
#(r'(.*)', web.StaticFileHandler, {'path': static_root}),
]
settings = dict(
debug=True,
static_path=static_root
)
application = tornado.web.Application(handlers, **settings)
if __name__ == "__main__":
port = 8090
application.listen(port)
print(f"Running: http://127.0.0.1:{port}")
tornado.ioloop.IOLoop.instance().start()
我用來發送POST帶有 JSON 數據的請求的代碼：
import requests
url = 'http://127.0.0.1:8090/webhook'
data = {'queryResult': {'queryText': 'Hello World'}}
r = requests.post(url, json=data)
print(r.status_code)
print(r.headers.get('Content-Type'))
print(r.json())
順便說一句：在 Flask 中你可以做
@app.route('/webhook', methods=['POST', 'GET'])
def webhook():
data_input = request.get_json(silent=True, force=True)
data_output = makeWebhookResult(data_input)
return jsonify(data_output)
添加回答
舉報
0/150
提交
取消