はじめに
OSC広島2017のLTで発表したアプリについて少し詳しく解説しようと思います。
この記事は、SLP_KBIT Advent Calendar 2017 - Qiitaの23日目の記事です。
APIとは
APIの正式名称は"Application Programming Interface"といいます。
今回はAPIの中でもWeb APIについてです。
APIがどういうものかというと、アプリケーションとアプリケーションの間でデータをやり取りすることができるものです。
例えば、Twitter APIでは、'選挙'というwordが入っているtweetだけ抽出し、選挙に対する意識を分析したり、Hearth StoneのAPIでは、対戦相手の出したカードを保存し、相手の構築を保存することができます。(相手が対戦中に出したカードに限りますが)。
分かりづらい人はアプリケーションのデータを好きに取り出したり、送ることができるものだと考えておきましょう。
今回発表したアプリ
今回OSC広島で発表したアプリは、Twitterのタイムラインからリア充っぽいwordが入っているtweetを検出し、tweetしたuserをミュートするというものです。これからの季節にピッタリのアプリになっています。簡単に説明すると、
1.ミュートするwordを登録しておきます(以下NGword)。
2.Twitter Streaming APIでタイムラインを取得します。
3.取得したタイムラインからNGwordを含むtweetからミュートするuserを検出します。
4.検出したuserをTwitter REST APIを使いuserをミュートします。
5.どのuserをミュートしたかををSlack APIを使いSlackに通知します。
というものです。
以下少し詳しく解説をば。
Twitter API
アプリケーションの登録
まずTwitterのアプリケーション管理画面でアプリケーションの登録をします。
アプリケーションの登録をして、
・Consumer Key
・Consumer Secret
・Access Token Key
・Access Token Secret
を取得します。
もし分からなくなったらこちらを参照してください。Twitter APIの使い方
取得できたら、config.pyにまとめておきましょう。
CONSUMER_KEY = "自分のCONSUMER_KEY"
CONSUMER_SECRET = "自分のCONSUMER_SECRET"
ACCESS_TOKEN = "自分のACCESS_TOKEN"
ACCESS_TOKEN_SECRET = "自分のACCESS_TOKEN_SECRET"
import
今回はPythonを使うので、
・requests
・OAuth1 from requests_oauthlib
をimportします。
pip install requests
pip install requests requests_oauthlib
でそれぞれインストールできます。
確認として、
$ python
>>> from requests_oauthlib import OAuth1Session
>>>
でエラーがでなかったら大丈夫です。
requestsも同じく。
Streaming APIでユーザタイムラインを取得する
それでは、ユーザタイムラインを取得してみましょう。
import json
import config
import requests
from requests_oauthlib import OAuth1
CK = config.CONSUMER_KEY
CS = config.CONSUMER_SECRET
AT = config.ACCESS_TOKEN
ATS = config.ACCESS_TOKEN_SECRET
url = 'https://userstream.twitter.com/1.1/user.json'
auth = OAuth1(CK, CS, AT, ATS)
r = requests.post(url, auth=auth, stream=True)
for line in r.iter_lines():
try:
tweet = json.loads(line)
print(tweet['user']['name']+'::'+tweet['text'])
print(tweet['created_at'])
print('----------------------------------------------------')
except json.decoder.JSONDecodeError:
pass
except:
print('止まるんじゃねぇぞ....')
pass
コードを実行するとこのようにタイムラインを取得することができます。
ただ、よく分からないエラーで止まってしまうことがあります。
今回は不都合がなさそうだったので、握りつぶしてエラーが起きたことだけ分かるようにしています。
Twitter REST APIを使いuserをミュートする
まずは、ミュートするNGwordを設定しましょう。
NG = ['イブ', 'クリスマス', '彼女', '彼氏', 'かれぴっぴ', 'デート']
今回は、このようなwordにしていますが、好きなNGwordを設定して大丈夫です。
では、タイムラインからNGwordを検出してtweetしたユーザをミュートしましょう。
import json
import config, ngList
import requests
from requests_oauthlib import OAuth1
CK = config.CONSUMER_KEY
CS = config.CONSUMER_SECRET
AT = config.ACCESS_TOKEN
ATS = config.ACCESS_TOKEN_SECRET
list1 = ngList.NG
url = 'https://userstream.twitter.com/1.1/user.json'
auth = OAuth1(CK, CS, AT, ATS)
stream_req = requests.post(url, auth=auth, stream=True)
for line in stream_req.iter_lines():
try:
flag = False
tweet = json.loads(line)
mute_get_url = "https://api.twitter.com/1.1/mutes/users/list.json"
mute_get_req = requests.get(mute_get_url,auth=auth)
mute_user_list = json.loads(mute_get_req.text)
print(tweet['user']['name']+'::'+tweet['text'])
print(tweet['created_at'])
print('----------------------------------------------------')
for NGword in list1:
if (NGword in tweet['text']) == True:
params ={'user_id' : tweet['user']['id']}
mute_post_url = "https://api.twitter.com/1.1/mutes/users/create.json"
mute_post_req = requests.post(mute_post_url,auth=auth, params=params)
if mute_post_req.status_code == 200:
print('NGwordを検出しました')
print('----------------------------------------------------')
except json.decoder.JSONDecodeError:
pass
except:
print('止まるんじゃねぇぞ....')
print('----------------------------------------------------')
pass
REST APIでユーザIDをPOSTし、userをミュートします。
このようにするとNGwordを検出し、userをミュートすることができます。
ミュートしたuserをSlackに通知する
では、最後にミュートしたuserをSlack APIを使ってSlackに通知しましょう。
Slack APIを使うための準備については、ここが分かりやすいと思います。
Slack APIの使い方
取得できたURLを使い、
import json
import config, ngList
import requests
from requests_oauthlib import OAuth1
CK = config.CONSUMER_KEY
CS = config.CONSUMER_SECRET
AT = config.ACCESS_TOKEN
ATS = config.ACCESS_TOKEN_SECRET
list1 = ngList.NG
url = 'https://userstream.twitter.com/1.1/user.json'
auth = OAuth1(CK, CS, AT, ATS)
stream_req = requests.post(url, auth=auth, stream=True)
for line in stream_req.iter_lines():
try:
flag = False
tweet = json.loads(line)
mute_get_url = "https://api.twitter.com/1.1/mutes/users/list.json"
mute_get_req = requests.get(mute_get_url,auth=auth)
mute_user_list = json.loads(mute_get_req.text)
for mute_user in mute_user_list['users']:
if tweet['user']['id'] == mute_user['id']:
print('ミュートしています')
print('----------------------------------------------------')
flag = True
if flag == False:
print(tweet['user']['name']+'::'+tweet['text'])
print(tweet['created_at'])
print('----------------------------------------------------')
for NGword in list1:
if (NGword in tweet['text']) == True:
params ={'user_id' : tweet['user']['id']}
mute_post_url = "https://api.twitter.com/1.1/mutes/users/create.json"
mute_post_req = requests.post(mute_post_url,auth=auth, params=params)
if mute_post_req.status_code == 200:
print('NGwordを検出しました')
print('----------------------------------------------------')
slack_url = 'YOUR SLACK URL'
slack_params = 'Content-type: application/json'
mute_user = tweet['user']['name']
mute_account = tweet['user']['screen_name']
mute_tweet = tweet['text']
slack_data = json.dumps({'text': u'{0} (@{1}) をミュートしました tweet内容 {2}'.format(mute_user, mute_account, mute_tweet) })
r = requests.post(slack_url, params=slack_params, data=slack_data)
except json.decoder.JSONDecodeError:
pass
except:
print('止まるんじゃねぇぞ....')
print('----------------------------------------------------')
pass
このようにすると、このようにSlackに通知されます。
まとめ
とりあえず動くものを作ってその後コードを触っていないので見るに耐えないコードになっていますね。
参考程度にしてもらえるとありがたいです。
Web APIを使ってみたかったというのと、OSC広島という機会があったことで作成したアプリです。
色々なデータが取れて面白いですね。
今回はこれだけしかやっていませんが、これより全然遊べます。
しかし、Twitter Streaming APIが使えなくなるかもとの噂があるので、遊べるうちに遊んでおこうと思っています。