トレンドに応じたWikipediaページを投稿するTwitter BotをHerokuでデプロイ
なんとか作った「相づちくん」Twitter Botが実質的にBANされてしまいました。しかし、せっかく作ったTwitterと Herokuのアカウントを放っておくのももったいないので、お手軽Botを作って走らせてみました。
日本のトレンド「 ティンさん 」に関係するかもしれないWikipediaページ ： パリクティン山https://t.co/zGxlB9COCz
— 週末bot (@weekendlog_bot) 2021年8月25日
何を作るか
前々からWikipediaのAPIについても興味があったので、Twitter APIと組み合わせて何かできないかと考えました。
Twitter APIでは、Twitter上のトレンドのキーワードが取得できます。このキーワードでWikipediaを検索し、検索結果の一番目の項目のWikipediaページへのリンクをツイートする、というBotを作ってみることにしました。
キーワード取得 -> Wikipedia検索 -> ツイート、だと、ごく簡単な処理の流れですが、1点、Twitter APIの使用上の制約を回避する必要があることがわかりました。その制約とは、「まったく同内容を繰り返してツイートするとTwitter APIがエラーとなる」ものです。
Botからのツイートは時間を置いて実行するとしても、それでも取得するトレンド・キーワードが前回のものと同じものになる可能性はあります。プログラムの開発やテストのためローカル環境で何度かツイートしてみる場合は、何もしなければ同じキーワードを連続して使うことになります。同じキーワードでWikipediaを検索すれば同じ結果となりますので、投稿するツイートも同内容となり、Twitter APIでエラーとなります。
この制約を回避するために「一度検索したキーワードを記録しておき、次回以降は使用を避ける」という処理が必要ということになりました。
どう作るか
「相づちくん」Twitter Botは、「メンションされたら相づちを返信する」というコンセプトですでの、メンション・ツイートを待ってリアルタイムに返信処理を行うため TwitterのStream APIを使って実装しました。
今回の「トレンドWIkipediaページをツイート」は、毎日決まった時間（もしくは一定の時間後）にツイートすることとして、Twitter REST APIをPythonのtweepyパッケージ経由で使うことにしました。「決まった時間にツイート」の部分はHeroku上でcronにより指定時刻に起動することで実現したいと思いました。
例によって、ネットで参考になる情報を検索したところ、heroku + Python で Twitterトレンド bot 作る - Qiitaに、今回やりたいことが書かれており、こちらのコードをお手本として使用させていただきました。
作ったBotのコード
今回作成したコードは、下記の2つのファイルになります。
- main.py : タイマー設定を行う
- trendwiki.py : トレンド・キーワード取得、Wikipedia検索、結果のツイートを行う。
タイマー設定
main.pyでタイマーの設定を行います。ほぼお手本のままです。
from apscheduler.schedulers.blocking import BlockingScheduler from trendwiki import trend_wiki_tweet sched = BlockingScheduler() # trend_wiki_tweetを起動 @sched.scheduled_job('cron', hour='7, 12, 18, 23') def main_trendwiki(): try: trend_wiki_tweet() except Exception as e: print('ERROR on trend_wiki_tweet()') print(e) if __name__ == '__main__': sched.start()
タイマー設定用のパッケージとしてAPschedulerを使っています。このパッケージは、わかりやすい記法で様々なパターンのタイマー設定を記述できて便利そうです。
今回も、一定インターバルではなく、睡眠時間帯をはずして1日4回ツイートするよう指定しています。
Twitterの認証処理
タイマー設定以外の処理はすべてtrendwiki.pyに書いています。
このファイルでは、冒頭でtweepyによるTwitter APIの認証処理を記述しています。
import tweepy import wikipedia import pickle import os import datetime # Twitter APIの認証処理 def twitter_api(): CONSUMER_KEY = os.environ['API_KEY'] CONSUMER_SECRET = os.environ['API_SECRET_KEY'] ACCESS_TOKEN = os.environ['ACCESS_TOKEN'] ACCESS_SECRET = os.environ['ACCESS_TOKEN_SECRET'] auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET) auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET) api = tweepy.API(auth) return api
お手本では、コマンドを使用してHerokuの環境変数にAPIキーなどを設定して使用する方法が書かれていました。この方法を使うと、Heroku側にキーやトークンを記述したファイルを置く必要がないので、うまいやり方だと思いました。
トレンドの取得〜Wikipedia検索〜ツイート
TwitterとWikipediaへのアクセスはtrendwiki.pyのtrend_wiki_tweet()に書いています。
関数名や変数名の付け方は全く自己流なんで、よいやり方を身に着けないと...orz。また、エラー処理をちょこちょこ書いていますが、これも正統なやり方を調べないと。
Twitterからトレンドのリストを取得
最初にTwitterからトレンドのリストを取得します。
def trend_wiki_tweet(): # Twitterに接続 api = twitter_api() # Twitterからトレンド・キーワードを取得 woeid_japan = 23424856 try: trends = api.trends_place(woeid_japan) except Exception as e: response_string = "トレンド取得時にエラー" print(response_string) print(e) return
検索済みキーワードのリストをファイルから取得
続いて前回までに検索したキーワードを保存したファイルから取得します。
リストの保存のため、初めて標準ライブラリのpickleを使ってみました。これは便利ですね。昔、C言語で構造体の配列やリストをダンプするのに結構苦労した記憶があります。こういう標準ライブラリがあればどれだけ楽だったことか……
# 前回チェックしたトレンド・キーワードをファイルから取得 if os.path.exists("./last_trend_name.tmp"): with open('./last_trend_name.tmp', 'rb') as f: last_trend_list = pickle.load(f) else: last_trend_list = []
Wikipeidaの検索
つぎのブロックでは、Twitterから得たトレンド・キーワード毎に（検索済みでなければ）Wikipediaを検索し、最初の項目のタイトルとURLを取得します。検索したキーワードは、検索済みキーワードリストに追加します。キーワードリストはファイルに保存して引き継いで行くのでサイズ過大になるとこまりますので、とりあえず上限30件に設定しています。
# wikipedia I/Fの設定 wikipedia.set_lang("ja") search_response = '' response_string = '' # Wikipediaをトレンド・キーワードで検索し、最初のコンテンツの情報を取得 for trend in trends: for trend_id in trend['trends']: trend_n = trend_id['name'] # 過去に検索済みのキーワードであれば飛ばす if trend_n in last_trend_list: continue # キーワードリストに追加 last_trend_list.append(trend_n) while len(last_trend_list) > 30: last_trend_list.pop(0) # Wikipediaを検索 search_string = '"' + trend_n + '"' try: search_response = wikipedia.search(search_string) except Exception as e: response_string = "Wikipediaページ検索時にエラー" print(response_string) print(e) break if not search_response: # 検索結果なしなら次のキーワードへ continue else: # 検索結果1番目のページのタイトル、URLをWikipediaから取得 try: response_string = wikipedia.page(search_response[0]).title response_url = wikipedia.page(search_response[0]).url except Exception as e: response_string = "Wikipediaページ情報取得時にエラー" print(response_string) print(e) break
検索済みキーワードリストの保存
ここで検索済みキーワードのリストをファイルに書き込みます。
# 今回チェックしたトレンド・キーワードをファイルに保存 with open('./last_trend_name.tmp', 'wb') as f: pickle.dump(last_trend_list, f)
検索結果をツイート
検索結果がセットされている場合は、検索に使ったキーワードと検索結果のページタイトル、URLをツイートします。
# 検索したWikipediaページのURLをツイート if (not search_response) or (not response_string): now = datetime.datetime.now() tweet = '新しいトレンド・キーワードがありませんでした' \ + '（' + str(now.year) + '年' \ + str(now.month) + '月' \ + str(now.day) + '日' \ + str(now.hour) + '時' \ + str(now.minute) + '分' \ + '）' else: tweet = "日本のトレンド「 {} 」に".format(trend_n) \ + "関係するかもしれないWikipediaページ" \ + " ： " \ + response_string \ + '\n' \ + response_url try: api.update_status(tweet) except Exception as e: response_string = "Tweet時にエラー" print(response_string) print(e) print("キーワード : " + search_string)
Herokuへのデプロイ
お手本の記事は、Herokuへのデプロイ方法もシンプルにまとめられていて、大変わかりやすく参考になりました。
一旦デプロイした後の細かな更新の際に使用するスクリプトが紹介されていまして、小技として活用させてもらっています。
APschedulerでタイマー設定した場合のHerokuの利用時間
pythonのAPschdulerなどでタイマー(cron)による自動実行 処理を実行させると、カスタムクロック・プロセスとして実行されます。このプロセスは、タイマーで指定した時刻に処理が走るわけですが、そのタイマー監視処理自体もHerokuの利用時間としてカウントされます。
$ heroku ps -a weekendlog-bot Free dyno hours quota remaining this month: 975h 35m (97%) Free dyno usage for this app: 11h 56m (1%)
これは、今回のTwitter Botをデプロイして実行を始めてからだいたい12時間後の無料利用時間の使用状況です。どうやら、プログラムをHerokuにデプロイし、ジョブとして実行指示した時点からリアルタイムの経過時間が利用時間としてカウントされるように見えます。
1日は24時間ですので、1ヶ月31日とすると、744時間。Herokuのデフォルトの無料利用時間は550時間ですので、Botを動かしたままだと不足してしまいます。ただ、Herokuは、アカウントにクレジットカード情報を登録すると1000時間に増えるとのことなので、このプラスアルファを反映すれば若干余裕が生じることになります。上記の使用状況は、このカード登録で無料時間を増やした後の状況です。
