概要
今さら言うまでもないですが、Twitterって便利ですよね。
全体的に匿名性が高くフランクな雰囲気が漂っているからなのか、面白いつぶやきがそこら中に転がっていますし、リアルタイム性に優れる事から最新のニュースなどがとても簡単に手に入ります。
実際、Twitter発信で何かがトレンド化する事も多いため、自分も情報収集ツールとして日頃から非常に重宝しています。
今回は、そんなTwitterのAPIを利用して
とあるユーザーのTwitterアカウントを監視しながら特定のキーワードを含むツイートに反応して何らかのアクションを実行する
コードを書いてみました。
一応、Twitterの標準アプリには特定のユーザーを指定してその行動（ツイート、リツイート、リプライ、いいねなど）を通知してくれる機能があった気がするのですが、特定のキーワードに対して何か反応をするみたいな細かな設定はできなかった気がするのでプログラミングの力で何とかしたいと思ったのが主な背景です。
仕様
- Python3
- Docker
自分はなるべくローカルの環境を汚したくないのでDockerを使用していますが、この辺のやり方には各自の判断に任せます。最終的にPython3が動く環境であれば何でもOK。
実装
前置きはそこそこに実装を行います。
Twitter APIのキー、トークンを取得
Twitter APIを利用するためには各認証情報（キーやトークンなど）が必要になるため、事前に入手しておきましょう。
2021年度版 Twitter API利用申請の例文からAPIキーの取得まで詳しく解説
↑の記事が参考になります。
環境構築
$ mkdir tweet-watcher && cd tweet-watcher
$ touch Dockerfile
$ touch docker-compose.yml
$ touch requirements.txt
$ touch test.py
./Dockerfile
FROM python:3
USER root
RUN apt-get update
RUN apt-get -y install locales && \
localedef -f UTF-8 -i ja_JP ja_JP.UTF-8
ENV LANG ja_JP.UTF-8
ENV LANGUAGE ja_JP:ja
ENV LC_ALL ja_JP.UTF-8
ENV TZ JST-9
ENV TERM xterm
ADD . /app
WORKDIR /app
RUN apt-get install -y vim less
RUN pip install --upgrade pip
RUN pip install --upgrade setuptools
RUN pip install -r requirements.txt
./docker-compose.yml
version: '3'
services:
python3:
restart: always
build: .
container_name: python3
working_dir: /app
tty: true
volumes:
- .:/app
./requirements.txt
tweepy
python-dotenv
- tweepy
- Twitter APIのラッパー
- python-dotenv
- APIキー・シークレットなどを環境変数で扱うためのライブラリ
./test.py
print('test')
動作確認
$ docker-compose up -d
$ docker-compose run python3 python test.py
test
ターミナル上で「test」と返ってくれば、Python3が動く環境の構築に成功です。
コード
$ touch main.py
$ touch .env
./main.py
import tweepy
import textwrap
import os
from dotenv import load_dotenv
load_dotenv()
# 反応したいキーワード（複数指定可）
keywords = [
'python',
'docker',
'twitter',
'api'
]
class StreamListener(tweepy.StreamListener):
# 対象のユーザーが新規にツイートをするたびにこの関数が走る
def on_status(self, status):
tweet_type = self.check_tweet_type(status)
# 通常のツイート以外（リツイートやリプライ）だった場合はここで終了
if tweet_type != 'normal_tweet': return
# ツイートの本文を取得（大文字/小文字の区別が面倒なのでとりあえず小文字に変換）
text = status.text.lower()
# ツイートの本文にキーワードが含まれているかどうかチェック
if any([keyword in text for keyword in keywords]):
# 以下に行いたい処理を記述
# 今回はサンプルなのでツイートの本文を出力してみる
heredoc = textwrap.dedent('''
-------------------------
{user_name} tweeted!
{text}
''').format(user_name = status.user.name, text = status.text).strip()
print(heredoc)
else:
# キーワードが含まれていなかった場合は何もしない
pass
# ツイートの種類をチェック（リツイート or リプライ or 通常のツイート）
def check_tweet_type(self, status):
# JSON内のキーに「retweeted_status」があればリツイート
if 'retweeted_status' in status._json.keys():
return 'retweet'
# 「in_reply_to_user_id」がNoneでなかった場合はリプライ
elif status.in_reply_to_user_id != None:
return 'reply'
# それ以外は通常のツイート
else:
return 'normal_tweet'
# 各認証情報を準備
api_key = os.getenv('TWITTER_API_KEY')
api_secret = os.getenv('TWITTER_API_SECRET')
access_token = os.getenv('TWITTER_ACCESS_TOKEN')
access_token_secret = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')
# API認証
auth = tweepy.OAuthHandler(api_key, api_secret)
auth.set_access_token(access_token, access_token_secret)
# リスナーを作成
stream_listener = StreamListener()
stream = tweepy.Stream(auth = auth, listener = stream_listener)
# 監視対象のユーザーID（https://idtwi.com/ ←で調べられる）
twitter_user_id = os.getenv('TWITTER_USER_ID')
# 監視スタート
print('Start watching tweets')
# ユーザーIDは配列で複数渡す事が可能
# もし別のスレッドで非同期処理を行わせたい場合はfilterの引数に「is_async = True」を渡す
stream.filter(follow=[twitter_user_id])
./.env
TWITTER_API_KEY=APIキー
TWITTER_API_SECRET=APIシークレット
TWITTER_ACCESS_TOKEN=アクセストークン
TWITTER_ACCESS_TOKEN_SECRET=アクセストークンシークレット
TWITTER_USER_ID=監視対象のユーザーID
【ポイント】
- 今回はよりリアルタイムにツイートを検知するためにREST APIではなくStreaming APIを使用。
- keywords（配列）に拾いたいキーワードを指定する（複数可）。
- ユーザーIDを調べるには以下のサイトを参照。
- 対象のユーザーが新規にツイートをするたびに「StreamListener」クラスの「on_status」関数が実行されるので、この中に主な処理を書いていく。
- デフォルトの状態だとリツイートやリプライなども拾ってしまうため、通常のツイートだけに絞りたい場合は条件文などで制御する必要あり。今回は「check_tweet_type」関数でそれを行っている。
- ツイートの本文にキーワードが含まれているかどうかは組み込み関数のanyでチェック。
- 「on_status」関数の背後では「while True〜文」が動いているため、安全に終了させたい場合はどこかでFalseを返してやればOK。
なお、「on_status」関数の引数に渡している「status」内には次のようなjsonデータが格納されているので、欲しい情報をキーで指定して適宜取り出してやりましょう。
statusの中身
{
"created_at": ,
"id": ,
"id_str": ,
"text": "api",
"source": ,
"truncated": ,
"in_reply_to_status_id": ,
"in_reply_to_status_id_str": ,
"in_reply_to_user_id": ,
"in_reply_to_user_id_str": ,
"in_reply_to_screen_name": ,
"user": {
"id": ,
"id_str": ,
"name": ,
"screen_name": ,
"location": ,
"url": ,
"description": ,
"translator_type": ,
"protected": ,
"verified": ,
"followers_count": ,
"friends_count": ,
"listed_count": ,
"favourites_count": ,
"statuses_count": ,
"created_at": ,
"utc_offset": ,
"time_zone": ,
"geo_enabled": ,
"lang": ,
"contributors_enabled": ,
"is_translator": ,
"profile_background_color": ,
"profile_background_image_url": ,
"profile_background_image_url_https": ,
"profile_background_tile": ,
"profile_link_color": ,
"profile_sidebar_border_color": ,
"profile_sidebar_fill_color": ,
"profile_text_color": ,
"profile_use_background_image": ,
"profile_image_url": ,
"profile_image_url_https": ,
"default_profile": ,
"default_profile_image": ,
"following": ,
"follow_request_sent": ,
"notifications": ,
"withheld_in_countries":
},
"geo": ,
"coordinates": ,
"place": ,
"contributors": ,
"is_quote_status": ,
"quote_count": ,
"reply_count": ,
"retweet_count": ,
"favorite_count": ,
"entities": {
"hashtags": ,
"urls": ,
"user_mentions": ,
"symbols":
},
"favorited": ,
"retweeted": ,
"filter_level": ,
"lang": ,
"timestamp_ms":
}
実行
$ docker-compose run python3 python main.py
あとは監視対象のユーザーが特定のキーワードを含むツイートを行なった場合、そのツイート内容がターミナルの出力されるはず。
今回は試しにテスト用のTwitterアカウントを自分で作成して試してみました。
Start watching tweets
-------------------------
Test Bot tweeted!
今日はPythonの勉強をしました！
-------------------------
Test Bot tweeted!
Dockerでの環境構築めっちゃ楽！
-------------------------
Test Bot tweeted!
Twitter APIの使い方がわかってきました！
こんな感じで出力されれば上手くプログラムが動いています。
あとがき
以上、とあるユーザーのTwitterアカウントを監視しながら特定のキーワードを含むツイートに反応して何らかのアクションを実行してみました。
今回はサンプルという事もあり簡単にツイート内容をターミナルに出力するだけでしたが、ちょっとコードを書き換えれば別にどんな処理でも行えるようになってます。
パッと思いつく限りでは、LINE Notify で自分のLINEアカウントに通知を飛ばしたり、仮想通貨取引所が公開しているAPIを叩いてビットコインの自動売買を行うなどですかね。（最近話題のイーロン・マスク砲に備えたり）
かなり応用の効く内容だと思うので、各自それぞれ工夫してみてください。