Twitter Streaming API を使ってリアルタイム自動返信Botを作る

はじめに

こちらの記事でやったことの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

実行ファイル

ファイルを読み込んで、関数を実行しているだけです。

bot.py

import bot

if __name__ == '__main__':
    bot.start()

設定ファイル

環境変数を読み込んで、Twitter APIを利用するためのオブジェクトを準備しています。

config.py

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

メインの処理

リアルタイムでタイムラインを監視し、特定のワードを含むリプライに対して自動返信します。
このファイルにまとめてしまってもよかったのですが、取得したツイートオブジェクトを扱う別のクラスを用意して、処理を分離しました。

__init__.py

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を使う処理や条件判定メソッドなどをこちらにまとめました。

tweet.py

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にデプロイした話