Created
December 12, 2018 05:35
-
-
Save torufurukawa/52c8be6503ae362984ffd06eed618022 to your computer and use it in GitHub Desktop.
Twitter Direct Message Chatbot example
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # coding: utf8 | |
| from flask import Flask | |
| app = Flask(__name__) | |
| @app.route('/') | |
| def hello(): | |
| return 'Hello World!' | |
| if __name__ == '__main__': | |
| app.run(host='127.0.0.1', port=8080, debug=True) | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # coding: utf8 | |
| import os | |
| import base64 | |
| import hashlib | |
| import hmac | |
| from flask import Flask, request, jsonify | |
| app = Flask(__name__) | |
| @app.route('/webhook', methods=['GET']) | |
| def webhook_challenge(): | |
| '''Webhook URL の正当性を確認''' | |
| key = os.getenv('CONSUMER_SECRET').encode() | |
| msg = request.args.get('crc_token').encode() | |
| hash = hmac.new(key, msg=msg, digestmod=hashlib.sha256).digest() | |
| response = {'response_token': 'sha256=' + base64.b64encode(hash).decode()} | |
| return jsonify(response) | |
| @app.route('/') | |
| def hello(): | |
| return 'Hello World!' | |
| if __name__ == '__main__': | |
| app.run(host='127.0.0.1', port=8080, debug=True) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # coding: utf8 | |
| import os | |
| import base64 | |
| import hashlib | |
| import hmac | |
| import requests | |
| from requests_oauthlib import OAuth1 | |
| from flask import Flask, request, jsonify | |
| SEND_ENDPOINT = 'https://api.twitter.com/1.1/direct_messages/events/new.json' | |
| # Twitter API への OAuth1 接続 | |
| twitter = requests.Session() | |
| twitter.auth = OAuth1(os.getenv('CONSUMER_KEY'), os.getenv('CONSUMER_SECRET'), | |
| os.getenv('ACCESS_TOKEN'), | |
| os.getenv('ACCESS_TOKEN_SECRET')) | |
| app = Flask(__name__) | |
| @app.route('/webhook', methods=['GET']) | |
| def webhook_challenge(): | |
| '''Webhook URL の正当性を確認''' | |
| key = os.getenv('CONSUMER_SECRET').encode() | |
| msg = request.args.get('crc_token').encode() | |
| hash = hmac.new(key, msg=msg, digestmod=hashlib.sha256).digest() | |
| response = {'response_token': 'sha256=' + base64.b64encode(hash).decode()} | |
| return jsonify(response) | |
| @app.route('/webhook', methods=['POST']) | |
| def webhook(): | |
| data = request.json | |
| print(data) | |
| # Direct Message イベントを順に処理する | |
| for event in data.get('direct_message_events', []): | |
| # 送信者と受信者が同じだったら、何もしない。 | |
| receiver_id = event['message_create']['target']['recipient_id'] | |
| sender_id = event['message_create']['sender_id'] | |
| if receiver_id == sender_id: | |
| print('same user ID: ', receiver_id) | |
| continue | |
| # テキストを抽出する | |
| text = event['message_create']['message_data']['text'] | |
| print('Text:', text) | |
| # テキストをそのまま返す | |
| e = { | |
| 'event': { | |
| 'type': 'message_create', | |
| 'message_create': { | |
| 'target': {'recipient_id': sender_id}, | |
| 'message_data': {'text': text} | |
| } | |
| } | |
| } | |
| resp = twitter.post(SEND_ENDPOINT, json=e) | |
| print(resp.status_code) | |
| # 200 OK 的なレスポンス | |
| return 'OK' | |
| @app.route('/') | |
| def hello(): | |
| return 'Hello World!' | |
| if __name__ == '__main__': | |
| app.run(host='127.0.0.1', port=8080, debug=True) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # coding: utf8 | |
| import os | |
| import base64 | |
| import hashlib | |
| import hmac | |
| import requests | |
| from requests_oauthlib import OAuth1 | |
| from flask import Flask, request, jsonify | |
| SEND_ENDPOINT = 'https://api.twitter.com/1.1/direct_messages/events/new.json' | |
| # Twitter API への OAuth1 接続 | |
| twitter = requests.Session() | |
| twitter.auth = OAuth1(os.getenv('CONSUMER_KEY'), os.getenv('CONSUMER_SECRET'), | |
| os.getenv('ACCESS_TOKEN'), | |
| os.getenv('ACCESS_TOKEN_SECRET')) | |
| app = Flask(__name__) | |
| @app.route('/webhook', methods=['GET']) | |
| def webhook_challenge(): | |
| '''Webhook URL の正当性を確認''' | |
| key = os.getenv('CONSUMER_SECRET').encode() | |
| msg = request.args.get('crc_token').encode() | |
| hash = hmac.new(key, msg=msg, digestmod=hashlib.sha256).digest() | |
| response = {'response_token': 'sha256=' + base64.b64encode(hash).decode()} | |
| return jsonify(response) | |
| @app.route('/webhook', methods=['POST']) | |
| def webhook(): | |
| data = request.json | |
| print(data) | |
| # Direct Message イベントを順に処理する | |
| for event in data.get('direct_message_events', []): | |
| # 送信者と受信者が同じだったら、何もしない。 | |
| receiver_id = event['message_create']['target']['recipient_id'] | |
| sender_id = event['message_create']['sender_id'] | |
| if receiver_id == sender_id: | |
| print('same user ID: ', receiver_id) | |
| continue | |
| # テキストを抽出する | |
| text = event['message_create']['message_data']['text'] | |
| print('Text:', text) | |
| # Quick Reply と Button をつけて返す | |
| e = { | |
| 'event': { | |
| 'type': 'message_create', | |
| 'message_create': { | |
| 'target': {'recipient_id': sender_id}, | |
| 'message_data': { | |
| 'text': text + 'って言った？', | |
| 'quick_reply': { | |
| 'type': 'options', | |
| 'options': [ | |
| { | |
| 'label': 'はい', | |
| 'description': '言ってたらこっちを選んでね', | |
| 'metadata': 'yes' | |
| }, | |
| { | |
| 'label': 'いいえ', | |
| 'description': '言ってなかったらこっちを', | |
| 'metadata': text | |
| } | |
| ] | |
| } | |
| } | |
| } | |
| } | |
| } | |
| resp = twitter.post(SEND_ENDPOINT, json=e) | |
| print(resp.status_code) | |
| # 200 OK 的なレスポンス | |
| return 'OK' | |
| @app.route('/') | |
| def hello(): | |
| return 'Hello World!' | |
| if __name__ == '__main__': | |
| app.run(host='127.0.0.1', port=8080, debug=True) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # coding: utf8 | |
| import os | |
| import base64 | |
| import hashlib | |
| import hmac | |
| import requests | |
| from requests_oauthlib import OAuth1 | |
| from flask import Flask, request, jsonify | |
| SEND_ENDPOINT = 'https://api.twitter.com/1.1/direct_messages/events/new.json' | |
| # Twitter API への OAuth1 接続 | |
| twitter = requests.Session() | |
| twitter.auth = OAuth1(os.getenv('CONSUMER_KEY'), os.getenv('CONSUMER_SECRET'), | |
| os.getenv('ACCESS_TOKEN'), | |
| os.getenv('ACCESS_TOKEN_SECRET')) | |
| app = Flask(__name__) | |
| @app.route('/webhook', methods=['GET']) | |
| def webhook_challenge(): | |
| '''Webhook URL の正当性を確認''' | |
| key = os.getenv('CONSUMER_SECRET').encode() | |
| msg = request.args.get('crc_token').encode() | |
| hash = hmac.new(key, msg=msg, digestmod=hashlib.sha256).digest() | |
| response = {'response_token': 'sha256=' + base64.b64encode(hash).decode()} | |
| return jsonify(response) | |
| @app.route('/webhook', methods=['POST']) | |
| def webhook(): | |
| data = request.json | |
| print(data) | |
| # Direct Message イベントを順に処理する | |
| for event in data.get('direct_message_events', []): | |
| # 送信者と受信者が同じだったら、何もしない。 | |
| receiver_id = event['message_create']['target']['recipient_id'] | |
| sender_id = event['message_create']['sender_id'] | |
| if receiver_id == sender_id: | |
| print('same user ID: ', receiver_id) | |
| continue | |
| # テキストを抽出する | |
| text = event['message_create']['message_data']['text'] | |
| print('Text:', text) | |
| # もし「いいえ」なら言い返す | |
| if text == 'いいえ': | |
| message_data = event['message_create']['message_data'] | |
| prev_text = message_data['quick_reply_response']['metadata'] | |
| message = '「' + prev_text + '」って言いましたよね。' | |
| e = { | |
| 'event': { | |
| 'type': 'message_create', | |
| 'message_create': { | |
| 'target': {'recipient_id': sender_id}, | |
| 'message_data': {'text': message}, | |
| } | |
| } | |
| } | |
| resp = twitter.post(SEND_ENDPOINT, json=e) | |
| print(resp.status_code) | |
| continue | |
| # Quick Reply と Button をつけて返す | |
| e = { | |
| 'event': { | |
| 'type': 'message_create', | |
| 'message_create': { | |
| 'target': {'recipient_id': sender_id}, | |
| 'message_data': { | |
| 'text': text + 'って言った？', | |
| 'quick_reply': { | |
| 'type': 'options', | |
| 'options': [ | |
| { | |
| 'label': 'はい', | |
| 'description': '言ってたらこっちを選んでね', | |
| 'metadata': 'yes' | |
| }, | |
| { | |
| 'label': 'いいえ', | |
| 'description': '言ってなかったらこっちを', | |
| 'metadata': text | |
| } | |
| ] | |
| } | |
| } | |
| } | |
| } | |
| } | |
| resp = twitter.post(SEND_ENDPOINT, json=e) | |
| print(resp.status_code) | |
| # 200 OK 的なレスポンス | |
| return 'OK' | |
| @app.route('/') | |
| def hello(): | |
| return 'Hello World!' | |
| if __name__ == '__main__': | |
| app.run(host='127.0.0.1', port=8080, debug=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment