はじめに
こちらの記事でやったことのTwitter API部分について詳しく書いていきます。
Twitter自動返信Botの開発を進めていく中で、2020年版の「ツイート（リプライ）をリアルタイムで取得する」といった内容の参考記事をなかなか見つけられなかったので、今回いろいろ自分で調べたことについてまとめたいと思います。
こちらが今回作ったBotの仕様です。
-
リアルタイムでタイムラインを監視し、特定のワードを含んだ自分あてのツイートを見つけると、自動で返信する
-
1アカウントにつき自動返信は１日１回まで
-
フォロワー以外のツイートには反応しない
-
リツイートには反応しない
-
非公開アカウントのツイートには反応しない（できない）
-
自分のツイートにも反応できるようにする
Twitter API
Twitter APIキーの取得
2020年8月現在では、APIキーの取得方法は若干異なりますが、利用申請の方法などこちらの記事を参考にさせていただきました。
・2020年度版 Twitter API利用申請の例文からAPIキーの取得まで詳しく解説
利用申請が承認されたら、Developer Portalにて以下を行いました。
-
App permissionsをReadからRead and Writeに変更（プログラムからツイートを行うため）
-
Access Token & SecretをGenerateする（APIの利用に必要）
Twitter Streaming API
どうやら、従来タイムラインのツイートをリアルタイムで取得するためには、User Streams APIというものが使われていたみたいです。
しかし、2018年の8月にこちらのAPIが廃止されたらしく、現在代用品としてはFilter realtime Tweetsのstatuses/filterが使えるそうです。
詳細については、こちらの記事を参考にさせていただきました。
・TwitterのStreaming APIについて - kivantium活動日記
上の記事では、tweepyが使われているみたいですが、個人的にtwitterライブラリの方が使いやすそうだったので、twitterライブラリを使用しました。
以下のコマンドでインストールします。（仮想環境下で実行）
(venv)$ pip install twitter
twitterライブラリについては、以下のサイトが参考になりました。
・twitter 1.18.0
・Python Twitter Tools 利用ノート
・Twitter UserStreamの代用品メモ for Python
twitterライブラリの使い方
Twitter APIのインターフェースオブジェクトの準備
import twitter
# OAuthの認証オブジェクトの作成
oauth = twitter.OAuth('Your Access token',
'Your Access token secret',
'Your API key',
'Your API secret key')
# Twitter REST APIを利用するためのオブジェクト
twitter_api = twitter.Twitter(auth=oauth)
# Twitter Streaming APIを利用するためのオブジェクト
twitter_stream = twitter.TwitterStream(auth=oauth)
リプライを送る
ツイートを送信するAPIのPOST statuses/updateはREST APIなので、上で用意したtwitter_apiオブジェクトを利用します。
text = 'テスト'
# 通常のツイートの場合
twitter_api.statuses.update(status=text)
# 特定のツイートへのリプライの場合
reply_text = '@username' + '\n'
reply_text += text
twitter_api.statuses.update(status=reply_text, in_reply_to_status_id='リプライするツイートのID')
リプライを送る場合は、in_reply_to_status_idを指定するだけではなく、テキスト内に@username（@ + リプライ先のツイートのユーザ名）を含めないといけないみたいです。
タイムラインをリアルタイムで取得する
上記の通り、Streaming APIを利用するので、上で用意したtwitter_streamオブジェクトを使います。
# 監視する対象の文字列
tracking_text = '@my_username テスト' # 半角スペースでAND検索。','区切りでOR検索
for tweet in twitter_stream.statuses.filter(language='ja', track=tracking_text):
print(tweet)
こちらを実行すると、「@my_username」と「テスト」の両方の文字列が含まれているツイートのオブジェクトtweetが取得できます。（公開アカウントのすべてのツイートが対象）
ぜひ、中身を確認してみてください。
このtweetオブジェクトに含まれる情報を利用して、リプライなどを行います。
また、filterの引数にはfollowというものもあり、こちらに「','区切りのユーザID」を渡すと、監視対象のユーザを制限できます。
今回、followを指定しなかった理由は、trackの条件とAND検索になるのではなく、OR検索になってしまうからです。
つまり、followで指定したユーザのツイートは内容に関係なく取得してしまいます。
今回の仕様的に、trackを指定する方がfollowを指定するよりも取得するツイートを絞り込めるので、trackのみを指定しています。
ソースコード
ディレクトリ構成
今回、返信テキストの登録・編集を行う管理画面をflaskで作ったので、構造をflaskに似せています。一つのファイルにまとめてしまっても、まったく問題ないです。
app/
├── .env # Twitter API Keyなどを準備
├── bot.py # 実行ファイル
└── bot/ # 返信テキスト用のDB処理・サービスクラス省略
├── __init__.py
├── config.py
└── tweet.py
実行ファイル
ファイルを読み込んで、関数を実行しているだけです。
import bot
if __name__ == '__main__':
bot.start()
設定ファイル
環境変数を読み込んで、Twitter APIを利用するためのオブジェクトを準備しています。
import os
from os.path import join, dirname
from dotenv import load_dotenv
import twitter
dotenv_path = join(dirname(__file__), '../.env')
load_dotenv(dotenv_path)
oauth = twitter.OAuth(os.environ.get('ACCESS_TOKEN_KEY'),
os.environ.get('ACCESS_TOKEN_SECRET'),
os.environ.get('CONSUMER_KEY'),
os.environ.get('CONSUMER_SECRET'))
twitter_api = twitter.Twitter(auth=oauth)
twitter_stream = twitter.TwitterStream(auth=oauth)
SCREEN_NAME = os.environ.get('TWITTER_SCREEN_NAME') # @usernameの自分のusername
メインの処理
リアルタイムでタイムラインを監視し、特定のワードを含むリプライに対して自動返信します。
このファイルにまとめてしまってもよかったのですが、取得したツイートオブジェクトを扱う別のクラスを用意して、処理を分離しました。
import logging
from bot.config import twitter_stream, SCREEN_NAME
from bot.tweet import Tweet
def start():
# loggingの設定
formatter = '%(levelname)s : %(asctime)s : %(message)s'
logging.basicConfig(level=logging.INFO, format=formatter)
REPLY_TEXT = 'こんにちは'
replied_user_list = [] # 返信したユーザをリストで管理
tracking_text = '@'+ SCREEN_NAME + ' こんにちは'
for tweet in twitter_stream.statuses.filter(language='ja', track=tracking_text):
tweet_obj = Tweet(tweet)
# 必要なキーが含まれていない場合
required_keys_list = [
'id_str',
'text',
'user'
]
if not tweet_obj.has_required_keys(required_keys_list):
logging.warning('FALSE->required key is empty')
print(tweet_obj.tweet)
continue
# リツイートの場合（tracking_textにマッチしていると反応してしまう）
if tweet_obj.is_retweet():
logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->is retweet', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
continue
# 過去に返信したユーザの場合
user_id = tweet_obj.get_user_id()
if user_id in replied_user_list:
logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->has already replied', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
continue
# 自身のツイートの場合
if tweet_obj.is_reply_from_me():
tweet_obj.reply(REPLY_TEXT)
replied_user_list.append(user_id)
logging.info('%s\n [user]: %s\n [tweet]: %s', 'SUCCESS->self tweet', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
continue
# フォロワーのツイートでない場合
if not tweet_obj.is_reply_from_follower():
logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->not follwer', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
continue
# 正常系
tweet_obj.reply(REPLY_TEXT)
replied_user_list.append(user_id)
logging.info('%s\n [user]: %s\n [tweet]: %s', 'SUCCESS', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
今回、返信テキスト用サービスクラスの処理の説明は省略しているので、固定の返信テキストREPLY_TEXTに変えています。
また、1日1リプライについてはこちらを参照ください。
ツイートオブジェクトを扱うクラス
Twitter REST APIを使う処理や条件判定メソッドなどをこちらにまとめました。
from bot.config import twitter_api, SCREEN_NAME
class Tweet():
def __init__(self, tweet):
self.tweet = tweet
def has_required_keys(self, required_keys_list):
for required_key in required_keys_list:
if required_key not in self.tweet:
return False
return True
def reply(self, text):
status = '@' + self.get_user_screenname() + '\n'
status += text
return twitter_api.statuses.update(status=status, in_reply_to_status_id=self.tweet['id_str'])
def is_retweet(self):
return 'retweeted_status' in self.tweet
def is_reply_from_me(self):
return self.get_user_screenname() == SCREEN_NAME
def is_reply_from_follower(self):
followers = twitter_api.followers.ids(screen_name=SCREEN_NAME, count=5000) # int型のフォロワーのID配列
return self.get_user_id() in followers['ids']
def get_user_id(self):
return self.tweet['user']['id']
def get_user_name(self):
return self.tweet['user']['name']
def get_user_screenname(self):
return self.tweet['user']['screen_name']
def get_tweet_text(self):
return self.tweet['text']
まとめ
最初、Twitterで自動返信Botを作りたいと思ってから、いろいろ調べていたのですが、なかなかそれらしき記事にたどり着くことができず、もう作れなくなったのかと思いました。
個人で何かを作るという経験が初めてだったので、いろいろ躓きましたが、とりあえず完成してよかったです。
まだ、プログラミング歴2年目で、しかも普段はPHPを使っている奴のPythonのコードはかなり至らないところがあると思いますが、私のような悩みを持っている人のお役に立てれば幸いです。
冒頭でも紹介していますが、こちらの記事もよろしくお願いします。
・【初めての個人開発】FlaskアプリとTwitterの自動返信BotをHerokuにデプロイした話