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

# はじめに

[こちらの記事](https://qiita.com/YiwaiY/items/ce790838725aba54c035)でやったことのTwitter API部分について詳しく書いていきます。

Twitter自動返信Botの開発を進めていく中で、2020年版の「ツイート（リプライ）をリアルタイムで取得する」といった内容の参考記事をなかなか見つけられなかったので、今回いろいろ自分で調べたことについてまとめたいと思います。

こちらが今回作ったBotの仕様です。

- リアルタイムでタイムラインを監視し、特定のワードを含んだ自分あてのツイートを見つけると、自動で返信する

- 1アカウントにつき自動返信は１日１回まで

- フォロワー以外のツイートには反応しない

- リツイートには反応しない

- 非公開アカウントのツイートには反応しない（できない）

- 自分のツイートにも反応できるようにする

# Twitter API

## Twitter APIキーの取得

2020年8月現在では、APIキーの取得方法は若干異なりますが、利用申請の方法などこちらの記事を参考にさせていただきました。
・[2020年度版 Twitter API利用申請の例文からAPIキーの取得まで詳しく解説](https://www.itti.jp/web-direction/how-to-apply-for-twitter-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](https://developer.twitter.com/en/docs/twitter-api/v1/tweets/filter-realtime/api-reference/post-statuses-filter)が使えるそうです。

詳細については、こちらの記事を参考にさせていただきました。
・[TwitterのStreaming APIについて - kivantium活動日記](https://kivantium.hateblo.jp/entry/2020/04/15/235422)

上の記事では、`tweepy`が使われているみたいですが、個人的に`twitter`ライブラリの方が使いやすそうだったので、`twitter`ライブラリを使用しました。
以下のコマンドでインストールします。（仮想環境下で実行）

```bash
(venv)$ pip install twitter
```

`twitter`ライブラリについては、以下のサイトが参考になりました。
・[twitter 1.18.0](https://pypi.org/project/twitter/)
・[Python Twitter Tools 利用ノート](https://showa-yojyo.github.io/notebook/python-twitter/index.html)
・[Twitter UserStreamの代用品メモ for Python](https://qiita.com/yukinoi/items/ca0d8059fe2a8e72a2c0)

# twitterライブラリの使い方

## Twitter APIのインターフェースオブジェクトの準備

```python
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](https://developer.twitter.com/en/docs/twitter-api/v1/tweets/post-and-engage/api-reference/post-statuses-update)はREST APIなので、上で用意した`twitter_api`オブジェクトを利用します。

```python
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`（@ + リプライ先のツイートのユーザ名）を含めないといけないみたいです。

## タイムラインをリアルタイムで取得する

[上記の通り](#twitter-streaming-api)、Streaming APIを利用するので、上で用意した`twitter_stream`オブジェクトを使います。

```python
# 監視する対象の文字列
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に似せています。一つのファイルにまとめてしまっても、まったく問題ないです。

```bash
app/
├── .env             # Twitter API Keyなどを準備
├── bot.py           # 実行ファイル
└── bot/             # 返信テキスト用のDB処理・サービスクラス省略
    ├── __init__.py
    ├── config.py
    └── tweet.py
```

## 実行ファイル

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

```python:bot.py
import bot

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

## 設定ファイル

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

```python: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
```

## メインの処理

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

```python:__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リプライについては[こちら](https://qiita.com/YiwaiY/items/ce790838725aba54c035#1%E3%82%A2%E3%82%AB%E3%82%A6%E3%83%B3%E3%83%88%E3%81%AB%E3%81%A4%E3%81%8D1%E6%97%A51%E3%83%AA%E3%83%97%E3%83%A9%E3%82%A4%E3%82%92%E3%81%A9%E3%81%86%E5%AE%9F%E8%A3%85%E3%81%99%E3%82%8B%E3%81%8B)を参照ください。

## ツイートオブジェクトを扱うクラス

Twitter REST APIを使う処理や条件判定メソッドなどをこちらにまとめました。

```python: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にデプロイした話](https://qiita.com/YiwaiY/items/ce790838725aba54c035)