heroku + Python で Twitterトレンド bot 作る

目的

　ブラウザ版、アプリ版ともに、Twitterのトレンドはトップ30ぐらいしか見ることができません。
　トレンドのページにずっと張り付くのも面倒ですし、任意の地域のトレンドを見るのも難しいです。

　そこで、Twitter APIを利用してトレンドを取得し、任意のハッシュタグがトレンドにあったときに通知をするTwitter botを作りましょう。
　Twitter APIを利用すれば、任意の地域のトレンドをトップ50取得できます。
　たぶん無料で運用できるはず。
　記事で扱っているコードの一部を書き換えるだけで動くと思います。

　ローカルで試したいだけの方は適宜読み飛ばしてください。

本文

準備

botのtwitterアカウント作成

　つくってください。

ディレクトリ準備

　botのファイルを保存するディレクトリを作成してください。
　以下で作成するファイルは全てここに保存すること。
　ここではvtuber-twitter-botとします。最終的にはこんな感じの構成。

vtuber-twitter-bot
├── dics.py
├── tools.py
├── main.py
├── index.py
├── runtime.txt
├── requirements.txt
├── Procfile

heroku

　こちらの記事が詳しいです。登録とCLIのインストールをしましょう。
　無料で運用するのでクレカの登録は不要。

• Heroku入門

Python 3.x

　Python 3.xでbotのコードを書いていきます。インストールするライブラリは以下。抜けがあったら教えてください。たぶん全部pipでインストールできます。

• APScheduler
  • herokuの機能ではできない細かい定期実行の設定のため
• tweepy
  • Twitter APIのため

Twitter API

　こちらの記事が詳しいです。
　API key, Consumer_secret, Access_token, Access_secretを控えておいてください。

• Twitter API 登録 (アカウント申請方法) から承認されるまでの手順まとめ

まずはローカルで

　以下のコードを実行してください。
　トレンドを取得できる都市の一覧が出てきます。
　取得したい都市をcountry(国名)、name（都市名）を頼りに探し、そのwoeidを控えてください。

import tweepy


def twitter_api():
    CONSUMER_KEY    = 'YOUR API KEY'
    CONSUMER_SECRET = 'YOUR CONSUMER SECRET'
    ACCESS_TOKEN    = 'YOUR ACCESS TOKEN'
    ACCESS_SECRET   = 'YOUR ACCESS SECRET'
    auth            = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
    auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
    api             = tweepy.API(auth)

    return api


api = twitter_api()
api.trends_available()

#>>[{'country': '',
#>>  'countryCode': None,
#>>  'name': 'Worldwide',
#>>  'parentid': 0,
#>>  'placeType': {'code': 19, 'name': 'Supername'},
#>>  'url': 'http://where.yahooapis.com/v1/place/1',
#>>  'woeid': 1},
#>> {'country': 'Canada',
#>>  'countryCode': 'CA',
#>>  'name': 'Winnipeg',
#>>  'parentid': 23424775,
#>>  'placeType': {'code': 7, 'name': 'Town'},
#>>  'url': 'http://where.yahooapis.com/v1/place/2972',
#>>  'woeid': 2972},
#>>     .
#>>     .
#>>     .

　世界トレンド、日本トレンド、日本の主要都市のトレンドを取得するには以下のwoeidがあれば十分だと思います。漏れがあったら上の方法で探してください。

woeid_dic = {'世界': 1, '日本': 23424856,
             '東京': 1118370, '京都': 15015372, '大阪': 15015370,
             '札幌': 1118108, '仙台': 1118129, '名古屋': 1117817, 
             '神戸': 1117545, '広島': 1117227, '福岡': 1117099,
             '埼玉': 1116753, '千葉': 1117034, '横浜': 1118550,
             '川崎': 1117502, '相模原': 1118072, '北九州': 1110809,
             '岡山': 90036018, '新潟': 1117881, '高松': 1118285,
             '浜松': 1117155, '熊本': 1117605, '沖縄': 2345896}

woeidを控えたらいよいよトレンド取得です。

city   = '東京'
woeid  = woeid_dic[city]
trends = api.trends_place(woeid)

#トレンドの順位と内容だけプリント
for trend in trends:
    trend_l = 0
    for trend_id in trend['trends']:
        trend_n  = trend_id['name']
        trend_l += 1
        print(city, trend_l, trend_n)

実装

　以下では、例として適当なハッシュタグがトレンドに入ったときに通知するbotを作ります。
　設計としては以下の通りです。取得頻度や通知頻度は適宜変更してください。その際、API limitを意識すること。（GET trends/place は 75リクエスト / 15分）

• トレンドを5分毎に取得
• 15分毎に通知
• 過去の通知より順位が低ければキャンセル

　dics.pyだけ書き換えればお好みのbotになるはず。

　理解のため、それぞれのコードをローカルで実行してみることをオススメします。
　os.environ['ENV_NAME']はENV_NAMEという環境変数の値を受け取るコードなので、ローカル実行の際は注意。

dics.py

dics.py

#トレンドにあったら通知したいハッシュタグのリスト
check_trend  = ['#ハッシュタグ', '#トレンド']

#トレンドを取得する地域のwoeid
woeid_dic = {'世界': 1, '日本': 23424856}

tools.py

tools.py

import pickle
import os
import tweepy

from dics import check_trend, woeid_dic


#Twitterでツイートしたりデータを取得したりする準備
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


def twitter_trend_notification(tw_switch):
    api       = twitter_api()
    trend_log = './trend_log.pickle'

    now_trend     = {}
    tmp_record    = {}
    recent_trends = {}
    for city in woeid_dic.keys():
        now_trend.update({city: {}})
        tmp_record.update({city: {}})
        recent_trends.update({city: {}})

    #トレンドを取得し、地域・順位・トレンドワードを抽出
    for city, woeid in woeid_dic.items():
        trends  = api.trends_place(woeid)
        for trend in trends:
            trend_l = 0
            for trend_id in trend['trends']:
                trend_l   += 1
                trend_name = trend_id['name']
                if trend_name in check_trend:
                    print(city, trend_l, trend_name)
                now_trend[city].update({trend_name: trend_l})

    #取得したトレンドのうち、通知したいワードについてのみ抽出
    for city in now_trend.keys():
        for c_trend in check_trend:
            if c_trend in now_trend[city].keys():
                in_dic_tmp = {c_trend: now_trend[city][c_trend]}
                tmp_record[city].update(in_dic_tmp)
    if not os.path.exists(trend_log):
        with open(trend_log, 'wb') as pi:
            pickle.dump(tmp_record, pi)
    with open(trend_log, 'rb') as pi:
        old_record = pickle.load(pi)
    #print(tmp_record)
    #print(old_record)

    #過去の記録と比べて、順位が上がっていれば更新する
    #過去の記録にないトレンドがあれば追加
    new_record = old_record
    for city in now_trend.keys():
        for t_trend in tmp_record[city].keys():
            if t_trend in old_record[city].keys():
                if tmp_record[city][t_trend] < old_record[city][t_trend]:
                    new_record[city][t_trend] = tmp_record[city][t_trend]
            else:
                in_dic_tmp = {t_trend: tmp_record[city][t_trend]}
                new_record[city].update(in_dic_tmp)
    with open(trend_log, 'wb') as pi:
        pickle.dump(new_record, pi)
    #if new_record != {'世界': {}, '日本': {}}:#, '東京': {}}:
    #    print('trend : ', new_record)

    #過去の通知10件での順位を抽出する
    recent_tweets = api.user_timeline(count=10, tweet_mode='extended')
    #recent_tweets = [tweet.full_text for tweet in recent_tweets]
    for tweet in recent_tweets:
        sp_tw = tweet.full_text.split('\n')
        if '【トレンド通知】' in sp_tw[0]:
            hashtag = '#'+ tweet.entities['hashtags'][0]['text']
            for key_city in recent_trends.keys():
                try:
                    re_lev  = sp_tw[2].replace('直近15分で「{}」が{}のトレンド'.format(hashtag, key_city), '').replace('位になったようです。', '')
                    re_lev  = int(re_lev)
                    try:
                        if recent_trends[key_city][hashtag] > re_lev:
                            recent_trends[key_city][hashtag] = re_lev
                    except:
                        recent_trends[key_city].update({hashtag: re_lev})
                except:
                    pass

    #通知を実行する場合、過去の通知より順位が上ならばツイートする
    if tw_switch:
        for city in new_record.keys():
            for trend_name in new_record[city].keys():
                if not trend_name in recent_trends[city].keys():
                    tw_flag = 1
                elif recent_trends[city][trend_name] > new_record[city][trend_name]:
                    tw_flag = 1
                else:
                    tw_flag = 0

                if tw_flag and (trend_name in check_trend):
                    tw_str  = '【トレンド通知】\n\n'
                    #tw_str  = '【トレンド通知】試験運用中\n\n'
                    tw_str += '直近15分で「{}」が{}のトレンド{}位になったようです。'.format(trend_name, city, new_record[city][trend_name])
                    print(tw_str)
                    api.update_status(tw_str)
        #ツイートしたら記録を消す
        os.remove(trend_log)

main.py

main.py

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()

from tools import twitetr_trend_notification


#ツイートしない
@sched.scheduled_job('cron', minute='5,10,20,25,35,40,50,55', hour='*')
def main_wo_tw():
    try:
        tw_switch = 0
        twitter_trend_notification(tw_switch)
    except Exception as e:
        print('ERROR on twitter_trend_notification')
        print(e)


#ツイートする
@sched.scheduled_job('cron', minute='0,15,30,45', hour='*')
def main_w_tw():
    try:
        tw_switch = 1
        twitter_trend_notification(tw_switch)
    except Exception as e:
        print('ERROR on twitter_trend_notification')
        print(e)



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

index.py

#空っぽ。要らない気もする。

デプロイ

ファイル作成

　runtime.txt, requirements.txt, Procfileを作成。

runtime.txt

python-3.6.2

requirements.txt

tweepy==3.6.0
APScheduler==3.0.3

Procfile

web: python index.py
clock: python main.py

初デプロイ

　app-nameはコードを保存するディレクトリの名前にするとわかりやすいです。
　以下のコマンドはこのディレクトリ内で実行すること。

heroku login
heroku create app-name

環境変数設定

　heroku config:set ENV_NAME="VALUE"で環境変数を設定できます。最後の4つはそのまま実行。
　app-nameはさっきのやつ。

appname=app-name

heroku config:set ACCESS_TOKEN="YOUR TWITTER ACCESS TOKEN" --app $appname
heroku config:set ACCESS_TOKEN_SECRET="YOUR TWITTER ACCESS TOKEN SECRET" --app $appname
heroku config:set API_KEY="YOUR TWITTER API KEY" --app $appname
heroku config:set API_SECRET_KEY="YOUR TWITTER API SECRET KEY" --app $appname
heroku config:set TZ="Asia/Tokyo" --app $appname

デプロイ

app-nameはさっきのやつ。

appname=app-name

git init
git add .
git commit -m "new commit"
git remote add heroku https://git.heroku.com/$appname.git
git push heroku master

メンテナンス

　app-nameディレクトリでheroku logs -tとコマンドを実行すれば、printされたログを確認できます。

　デバッグにはapp-nameディレクトリの1つ上のディレクトリに以下のシェルスクリプトをおいとくと楽です。
　コード修正などが終わったら、app-nameディレクトリで../heroku_deploy.shを実行。失敗したらchmod u+x ../heroku_deploy.shとかで実行可能に設定。

heroku_deploy.sh

git add .
git commit -m "new commit"
git push heroku master
heroku ps:scale web=0 clock=1
heroku ps
heroku logs --tail