Chatbot with Twitter Direct Message API
pyspa Advent Calendar 2018 13日目なう。
Twitter の Directe Message API を使うと、Direct Message （以下 DM）を使った chatbot を実装できる。ときどき「どうやって作るんですか？」とか聞かれることがあって、ドキュメント読んで勝手に作れば、と思っていた。内緒だけど今でも思っている。それはそうなんだけど、この1年で DM API の beta が取れたり、機能が変わったり、手続きが増えたりとかで、分かりにくいとは思う。
というわけで、昨年からのアップデートをする。
免責事項
- 個人的な調べ物の成果である。
- この文書は概念実証を示すものであり、いかなる保証もしない。
- 公開された情報だけを使っている。
使ったもの
特に最新版である必要はないはずだけれど、使ったバージョンを記載しておく。
手続き
Account Activity API というのを使うにあたって、以下の登録作業をする。
- 開発者登録 https://developer.twitter.com/en/apply/user
- アプリ作成 https://developer.twitter.com/en/apps
Read, write, and direct messages を有効にする
consumer key と consumer secret を後で使う - Account Activity API 登録 https://developer.twitter.com/en/apply
すべての申請・登録・承認が完了すると https://developer.twitter.com/en/account/environments から dev environment label が見えるようになる。これも後でつかう。
OAuth
開発者登録したアカウントを chatbot にするなら、アプリの管理画面から手作業で Access Token と Access Token Secret を取得できるので、このセクションの作業は不要。
開発者登録アカウントとは別のアカウントを、chatbot にするのであれば以下の手順。
$ twurl authorize --consumer-key XXXXXXXX --consumer-secret XXXXXXXX
Go to https://api.twitter.com/oauth/authorize?oauth_consumer_key=XXXXXXXX&oauth_nonce=... and paste in the supplied PIN
https://api.twitter.com/oauth/authorize で始まる URL をブラウザで開く。すると Twitter OAuth の画面になるので、進めていく。
最後に番号が表示される。この番号を、ターミナルにコピペする。
1234567
Authorization successfulこれで完了。Access Token/Secret は ~/.twurl に入っている。
$ cat ~/.twurlrc
---
profiles:
torufurukawa:
XXXXXXXX:
username: torufurukawa
consumer_key: XXXXXXXX
consumer_secret: XXXXXXXX
token: XXXXXXXX
secret: XXXXXXXX
configuration:
default_profile:
- torufurukawa
- XXXXXXXXtoken と secret が、それぞれ Access Token と Access Token Secret 。後で使う。
サーバーのスケルトン
とりあえず hello を返すサーバーを作って、少しずつ chatbot サーバーにしていく。
Python のライブラリをインストール。
pip install Flask==1.0.2 requests==2.21.0 requests-oauthlib==1.0.0サーバーアプリを作ります。main1.py より抜粋。
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)
実行すると…
$ python main1.py
* Serving Flask app "main1" (lazy loading)
* Environment: production
WARNING: Do not use the development server in a production environment.
Use a production WSGI server instead.
* Debug mode: on
* Running on http://127.0.0.1:8080/ (Press CTRL+C to quit)
...http://127.0.0.1:8080 を開いて「Hello World!」が表示されたらOK。
Webhook の登録
Account Activity API は、アカウントに関連するイベント、たとえばリツートとか、DMとか、いいねとかが起こったときに、Twitter のサーバーがWebhook URL にイベントを届ける。
手順としては、(1) webhook URL を register し、(2) 特定のアカウントを subscribe する。完了すると subscribe したアカウントのイベントが webhook URL に送信される。
register するとき、Twitter が webhook URL に対して GETリクエストする。そのレスポンスを確認して、問題なければ register が成功する。
何を確認するのかと。Twitter が crc_token なる文字列を送ってくるので、consumer secret をキーに、HMAC SHA256 でハッシュ化して返す。
環境変数 CONSUMER_SECRET を定義してから、以下のコードを書く。
main2.py より抜粋。
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)...
で、サーバーを立ち上げておく。
$ python main2.pyこのサーバーが外部からアクセスできるように、ngrok でトンネルする。
$ ngrok http 8080すると下のような画面になるであろう。
ngrok by @inconshreveable (Ctrl+C to quit)Session Status online
Account Toru Furukawa (Plan: Free)
Version 2.2.8
Region United States (us)
Web Interface http://127.0.0.1:4040
Forwarding http://xxxxxxxx.ngrok.io -> localhost:8080
Forwarding https://xxxxxxxx.ngrok.io -> localhost:8080Connections ttl opn rt1 rt5 p50 p90
0 0 0.00 0.00 0.00 0.00
ありがとう、inconreveable さん。このプロセスが動いている間、https://xxxxxxxx.ngrok.io へのアクセスは、ローカルの 8080 に渡される。サブドメインは起動ごとに変わる。有料プランにすると、固定できる。
以下の手順で register する
$ twurl -X POST --consumer-key XXXXXXXX --consumer-secret XXXXXXXX --access-token XXXXXXXX --token-secret XXXXXXXX "/1.1/account_activity/all/env-beta/webhooks.json?url=https://xxxxxxxx.ngrok.io/webhook"すると、python main2.py を動かしているプロセスに、Twitter から ngrok 経由でアクセスがある。CRC トークンをつけて GETしてくるので、先程実装したチャレンジの計算をした結果を、JSONで返す。問題がなければ register が完了する。
続いて subscribe 。
$ twurl -X POST --consumer-key XXXXXXXX --consumer-secret XXXXXXXX --access-token XXXXXXXX --token-secret XXXXXXXX "/1.1/account_activity/all/env-beta/subscriptions.json"成功すると、これ以降、アカウントに関するイベントがあると、webhook に POSTリクエストがくるようになる。
試しにこのアカウントに DM を送ったりすると、POSTリクエストがくる。だが、POSTリクエスト処理を実装していないので、サーバーはエラーを
エコー
話しかけられたらオウム返しする chatbot を作る。環境変数 CONSUMER_KEY, ACCESS_TOKEN, ACCESS_TOKEN_SECRET を定義しておく。
main3.py より。
...
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=['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'
...
送信者と受信者が同じだったら、応答しないようにしている。これで独り言を延々と繰り返すことを防いでいる。
メッセージの文字列を取り出して、再送信する。それだけ。
Quick Reply
DM っぽいインタフェースでは、直接文字を入力するだけでなく、選択肢を提示することがある。Facebook Messenger しかり、LINE しかり。そして Twitter DM しかり。Twitter では quick reply とよぶ。
ユーザーに対して、選択肢を提供してタップ・クリックで入力させるためのデータ構造である。各選択肢には metadata なるパラメータがある。これについては、次のセクションで。
main4.py より
...@app.route('/webhook', methods=['POST'])
def webhook():
...
data = request.json
# Direct Message イベントを順に処理する
for event in data.get('direct_message_events', []):
... # 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) ......
Quick Reply からの返信を受け取る
選択肢をタップすると text に加えて、 metadata も一緒に webhook に届く。上述の例では、直前のユーザーの発言が metadata に格納している。（他の値を設定してもよい）
{
"direct_message_events": {
[
{
"message_create": {
"sender_id": "3945351",
"target": {"recipient_id": "919753461918416897"},
"message_data": {
"text": "いいえ",
"quick_reply_response": {
"metadata": "ちんこ"
}
}
}
}
]
}
}「いいえ」を受け取ったら、metadata を使って何か発言するように変更する。
...@app.route('/webhook', methods=['POST'])
def webhook():
...
text = event['message_create']['message_data']['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 ... ......
本番投入
実践投入するには、少なくとも以下を考慮する必要がある。
- Account Activity API の有料プランが必要かも。
- DM 送信の per app の rate limit が小さいので、別途申請が必要。
- 上のコードはエラーチェックとか、rate の制御とかなんにもしていない。そういうのが必要。
- rate limit は常に存在するので、送信ペースの制御。
まとめ
Twitter DM APIを使って、プライベートな chatbot を作る考え方を紹介した。事実上 Account Activity API は必要であろう。
明日は、初めてソフトウェア開発の仕事をし始めたときに trac を教えてくれた @feiz です。メリークリスマス、そして、ハッピーニューイヤー。
![俺たちの Trello が買収された](data:image/svg+xml,<svg xmlns="http://www.w3.org/2000/svg" width="679" height="509"><rect fill-opacity="0"/></svg>)
![Toru Furukawa](data:image/svg+xml,<svg xmlns="http://www.w3.org/2000/svg" width="20" height="20"><rect fill-opacity="0"/></svg>){kind=small}
