PythonでSlackBot開発⑧「Heroku Schedulerでプログラム定期実行」

PythonでSlackBotを開発しよう企画の第8回目。今回はプログラムを定期的に自動実行してくれるようにしていきます。本格的にBotっぽくなってきましたが、リクエストを受け付けて処理を始めるようでは、処理を自動でしてくれている感がなくて辛いので、自動で働いてくれるようにしていきます。

Heroku Schedulerとは？

今回利用するのは、Herokuの提供しているHeroku Schedulerです。

Scheduler is a free add-on for running jobs on your app at scheduled time intervals, much like cron in a traditional server environment. - スケジューラーは、従来のサーバー環境のcronと同様、アプリケーション上でスケジュール設定された時間間隔でジョブを実行するための無償のアドオンです。
（引用：Heroku Scheduler | Heroku Dev Center）

Heroku Schedulerはアプリケーションを動かしているDynoとは別で利用時間換算されるようなので、使いすぎると課金される可能性はありますが、普通に個人で利用している分には無料で利用できるアドオン機能です。

それでは、早速利用していきたいと思います。

Heroku Schedulerの設定

まずは、定期実行したい処理を記述したファイルを用意します。今回は、定期的に自動でTwitterでつぶやきを検索して、指定した文字列でつぶやかれているつぶやきの中で「いいね」がついている上位5個のつぶやきを教えてくれるように実装します。定期的に呼び出す「tweet_post_slack_with_search.py」ファイルに以下のように処理を実装します。

tweet_post_slack_with_search.py

# coding=utf-8
from service import twitter_service
from service import slackbot_service

def tweet_post_slack():
    api = twitter_service.prepare_twitter_api()
    search_results = twitter_service.search_tweet(api, 'ethereum', 'popular', 100)
    if len(search_results) == 0:
        slackbot_service.send_to_slack(message, 'ツイート見つからなかったす。')
        return

    # いいね数の多い順のつぶやき一覧取得
    statuses = twitter_service.sort_by_favorite_count(search_results, False)
    # いいね数の多い5つのつぶやき辞書取得
    top_five_statuses_dictionary = twitter_service.select_statuses(statuses, 5)
    top_five_statuses = top_five_statuses_dictionary.values()
    # slackにポストする形式に整えた文章作成
    text = twitter_service.create_tweet_list_text(top_five_statuses)
    slackbot_service.post_text('general', '自動お知らせっすわ！\n{}'.format(text))

tweet_post_slack()

「tweet_post_slack_with_search.py」から呼び出す「slackbot_service.py」と「twitter_service.py」も以下のように実装します。

slackbot_service.py

# coding=utf-8
import os
import requests

from service import docomo_dialogue_service
from service import twitter_service

def send_to_slack(message, text):
    """
    Slackで呼び出されたチャンネル上でtext返信
    """
    message.send(text)

def post_text(channel, text):
    """
    Slackの特定のチャンネルにtextを送信する（未検証メソッド）
    """
    url_slack_api = 'https://slack.com/api/chat.postMessage'
    slack_api_params = {
        'token': os.environ['SLACKBOT_API_TOKEN'],
        'channel': channel,
        'text': text,  # 投稿するテキスト
        'username': 'surfer-bot',  # 投稿のユーザー名
        'icon_emoji': ':sunglasses:',  # 投稿のプロフィール画像に入れる絵文字
        'link_names': 1,  # メンションを有効にする
    }
    requests.post(url_slack_api, data=slack_api_params)

twitter_service.py

# coding=utf-8
import tweepy
import os

def prepare_twitter_api():
    """
    TwitterのAPIアクセスキーを取得
    """
    auth = tweepy.OAuthHandler(os.environ['CONSUMER_KEY'], os.environ['CONSUMER_SECRET'])
    auth.set_access_token(os.environ['ACCESS_TOKEN'], os.environ['ACCESS_TOKEN_SECRET'])
    return tweepy.API(auth)

def search_tweet(api, word, result_type, count):
    """
    Twitterでwordで一致する日本語つぶやき情報を取得
    result_type: popular->人気のツイート。recent->最新のツイート。mixed->全てのツイート。
    """
    search_results = api.search(q=word, lang='ja', result_type=result_type, count=count)
    return search_results

def create_tweet_list_text(statuses):
    """
    与えたつぶやきのリストを元に、以下のようなテキストを作成する
    --------------------------
    太郎(@taro)
    つぶやきだよ
    (2いいね)https://twitter.com/taro/status/randomalphabet
    --------------------------
    ...
    """
    post_text = ''
    for status in statuses:
        user = status.user
        fav_count = status.favorite_count
        tweet_link = 'https://twitter.com/{}/status/{}'.format(user.screen_name, str(status.id))
        result_text = '\n{}@({})\n{}\n({}いいね){}\n'.format(user.name, user.screen_name, status.text, str(fav_count), tweet_link)
        post_text = '{}-------------------------------------------------------{}'.format(post_text, result_text)
    return post_text

def sort_by_favorite_count(statuses, is_asc):
    """
    引数で与えられたつぶやきを「いいね」の少ない順で並び替えてstatusリストを返す。(is_ascがfalseなら多い順)
    """
    # 辞書{つぶやき, いいね数}に詰めていく
    if is_asc:
        return sorted(statuses, key=lambda status: status.favorite_count, reverse=False)
    else:
        return sorted(statuses, key=lambda status: status.favorite_count, reverse=True)

def select_statuses(statuses, count):
    """
    つぶやきの最初のものからcountの数の{id, つぶやき}辞書を取得。(countが0以下の場合は空の辞書を返す)
    """
    result_dictionary = {}
    if count < 1:
        return result_dictionary
    loop_count = 0
    for status in statuses:
        loop_count += 1
        result_dictionary.setdefault(loop_count, status)
        if loop_count >= count:
            break
    return result_dictionary

これで、「tweet_post_slack_with_search.py」を実行すれば、Twitterのつぶやきをいい感じに検索してSlackにpostしてくれます。試しにterminalでファイルを実行してみると...

$ forego run python tweet_post_slack_with_search.py

↑いい感じにpostしてくれました！

ファイルの変更が完了したらGitに反映して、「$ git push heroku master」してHerokuにデプロイします。ファイルの変更内容が反映できたら、Heroku Schedulerの設定をしていきます。以下のコマンドでHeroku Schedulerをアドオンします。

$ heroku addons:create scheduler:standard

アドオンできたら、以下のコマンドでHeroku Schedulerの設定画面を開きます。

$ heroku addons:open scheduler

コマンドを打つと、↑ 画面が開くので、ここでスケジュール設定をしていきます。

21:30にファイルが実行されるように設定しておきます。「これで21:30になったら来るはず！」と思って待っていましたが、21:30になってもSlack通知が飛んでこず...。

何か設定が足りてないかな？？と思ってましたが、次の日確認すると、6:30にファイルが起動してSlackにメッセージがpostされていました。

完全に見落としていましたが、UTC（協定世界時）での21:30。つまり日本時間でいうと9時間先の6:30に実行されるということですね。日本時間で設定できるように変更できるのかなと思って調べてみましたが、パッとは見つかりませんでした。

とはいえ、これでプログラムの自動実行をすることができるようになりました！

まとめ

これでTwitterに定期的につぶやくBot機能も実装できるし、1時間に1回仮想通貨チャートを確認して、値上がり率が一定以上になっていたらSlackで教えてくれる機能なども実装できます。ほんとプログラムって便利だ...。

普通のWebアプリケーションでは、ブラウザからHTTPリクエストを受け付けて、処理を実行する・画面を表示するという処理の流れが普通だと思います。ですが今回開発しているBotでは、Slack上で期待している入力をしたらそれをリクエストとして受け付けて処理を実行するというものにしているので、アクセス経路が限定されます。Heroku Schedulerなどを利用して定期実行できるようにしておくことで、Botに自動かつ定期的に働いてもらうようにして生活の諸々を自動化してきたいっすね。

Pythonでのプログラミング学習中の方向けに、Noteでより詳細なプログラミングチュートリアルを配信しているので、そちらもチェックしてみてください( ・v・)/

note.mu

では！