お役立ち Twitter Bot を作りながら学ぶ AWS ドリル

~第 4 回 新章突入 ! 気になるワード検索 & 通知 Bot 編

2022-04-04
How to be a Developer

Author : 金澤 圭

 ツイート »
 シェア »
 はてブ »

ソリューションアーキテクト (SA) の金澤 (@ketancho) です。4 月になりましたが、皆さまいかがお過ごしですか ? 今月から新社会人として Developer になった方、あるいは転職し Developer としての活動をスタートされた方、様々な方にとってのスタートになる時期だと思います。おめでとうございます🎉

私はこの builders.flash の How to be a Developer カテゴリに寄稿を続けている人間なので、この時期に記事を書くのが一番ワクワクドキドキします。少しでも皆さまの活動を後押しできる記事を書いていければと思っていますので、ぜひ #AWSウェブマガジン ハッシュタグをつけてご感想や、フィードバックをもらえると嬉しいです🙏さて、前回の AWS ドリル #3 までで、朝イチであいさつする Bot が完成しました。今月からは新章に突入し、少しずつ実用性のある Twitter Bot を作っていきます。

4 月ということで、もしかしたら「この #4 から AWS ドリルを読み始めます !」という方も多いのかなーと思っています。そこで今回は、本記事（ドリル #4）だけを見れば Twitter Bot の開発が完結できるように記事を書き進めていきたいと思います。前提となる準備作業は過去の記事をインデックスしながら進めますので、今月からドリルを始めますという方もご安心して読み進めていただければと思います。(もちろん下の方にある「この連載記事のその他の記事はこちら」からバックナンバーに戻っていただき、#1 からご覧いただくのもおすすめですし、嬉しいです !)

この連載記事のその他の記事はこちら

選択
  • 選択
  • 第 1 回 おはよう Bot 編
  • 第 2 回 昔書いた記事の宣伝 Bot 編
  • 第 3 回 リファクタリング ＆ 曜日ごとのツイート 編
  • 第 4 回 新章突入 ! 気になるワード検索 & 通知 Bot 編
  • 第 5 回 皆さまの代わりに英語でツイートしておくよ Bot 編
  • 第 6 回 AWS Step Functions を使って Well-being Bot を作ろう ! (前編)
  • 第 7 回 AWS Step Functions を使って Well-being Bot を作ろう ! (後編)
  • 第 8 回 - 総集編 & 夏休みの自由研究のすすめ
  • 第 9 回 - IaC 入門しながら作るリマインダー Bot (前編)
  • 第 10 回 - IaC 入門しながら作るリマインダー Bot (中編)
  • 完結回 - IaC & CI/CD 入門しながら作るリマインダー Bot (後編)

目次

1. この記事で作る Bot の機能と、とりあげる AWS サービス・機能

2. 今回から AWS ドリルを始める方の作業

2-1. Twitter API を使う準備をする
2-2. AWS アカウントがない方はアカウントを作る
2-3. AWS Systems Manager の Parameter Store の準備

3. 新しい Lambda 関数 twitter-bot-search を作成し、お試しツイートする

3-1. Lambda 関数を新規で作成する
3-2. Lambda 関数に紐づく IAM ロールの修正
3-3. Lambda 関数のコードを修正する

4. Twitter の検索 API を使ってみる

5. 毎朝自動的にツイート検索するように設定する

5-1. Lambda 関数のコードを修正する
5-2. Amazon EventBridge の設定

6. しきい値を超えたときには Amazon SNS を使ってメール通知もする

6-1. Amazon SNS 側の設定
6-2. Lambda 関数 twitter-bot-search の修正

7. 落ち穂拾い & 参考情報

8. まとめ

ご注意

本記事で紹介する AWS サービスを起動する際には、料金がかかります。builders.flash メールメンバー特典の、クラウドレシピ向けクレジットコードプレゼントの入手をお勧めします。

*ハンズオン記事およびソースコードにおける免責事項 »

このクラウドレシピ (ハンズオン記事) を無料でお試しいただけます »

毎月提供されるクラウドレシピのアップデート情報とともに、クレジットコードを受け取ることができます。 

今すぐ特典を受け取る »

1. この記事で作る Bot の機能と、とりあげる AWS サービス・機能

今回は気になるキーワードを定期的に検索 & 通知する Twitter Botを開発します。いきなりですが、Twitter で定期的に検索しているワードはありますか ? 私は、自分が作った AWS 関連のコンテンツへの感想を調べたり (いわゆるエゴサーチ)、電車の遅れや天気の崩れといった心配事を調べたりすることがよくあります。前者はたくさんヒットすれば嬉しいですし、後者は少ない方がありがたいです。

今回はこのような定点観測するワードを決め、Twitter の検索 API を使って検索し、検索結果がいつもより増えていたら通知する Bot を作ります。

具体的なユースケースを、私の日頃の生活を例にして共有します。私がオフィスに行く日は、目黒線を使って通勤しています。満員電車が苦手なので、朝 6 時に起きて、6 時半ごろに電車に乗ることが多いです。もし電車が遅延しているのであれば、一本でも早い電車に乗りたいと思っています。しかし、朝起きたときに電車の遅延に気付ければ準備を急げるのですが、朝はバタバタしてしまい調べる余裕なんてありません。(天気予報ですら見忘れてよくビニール傘を買っています・・・)

そこで今回、私に代わって Bot にそれをやってもらおうと思います。Twitter 上で「目黒線」というワードで検索し、普段よりもツイート数が多ければ通知で教えてもらうことで、電車の遅延を怪しめるようにしたいと思います。

今回作るもの・お題を整理すると、

  • 毎朝 6:01 に、5:00-6:00 の「目黒線」を含むツイートを検索する。
  • 検索結果をツイートする。
  • 検索に該当する件数が 20 を超えていたら、@ketancho という文字列をつけて通知が飛ぶようにする。
  • また、自身のメールアドレス宛にメールも送るようにする。

となります。太字にした部分は皆さまのライフスタイルや興味によって変更していただける箇所になります。ぜひアレンジしていただき、生活をよくする Twitter Bot を生み出してみてください。

通知するしきい値をどれくらいにすべきかは、平常時のツイート数によって変わってきます。そこでこの記事では、検索 API のお試しも兼ねて、普段どれくらいのツイート数になるかを把握するところからスタートします。また、私は目黒線で目黒に通っていることを隠していないので構わないのですが、ツイートで個人情報を晒さないようにご注意ください。そもそも路線名はツイートに含めない、あるいは Twitter アカウントを非公開にするといった自衛をしていただくようお願いいたします。そして、@ ツイートする先は皆さまのアカウントに変更するようにお願いします。皆さまが上の例のまま実装を進めてしまうと、私にたくさんの通知が飛んできてしまいますので、何卒ご注意いただければと思います😅 (それはそれで皆さまが試してくださっていることを知れて嬉しいのですが。いや、フリではありませんよ。)

さて、今回の検索 & 通知 Bot では、最後のメール通知を送る部分で Amazon Simple Notification Service (Amazon SNS) を利用します。Amazon SNS は、フルマネージド型の pub / sub メッセージングサービスです。アプリケーションの何かの処理が終わったあと、他のアプリケーションや機能と連携するときに、この Amazon SNS がハブの役割をします。間に入ることでアプリケーションや機能間を疎結合にできるメリットがあります。今回は Lambda 関数から Amazon SNS のトピックに対して条件を満たした (しきい値より多い数のツイートがヒットした) ことを Publish し、事前にそのトピックを Subscribe している人のメールアドレスにメール通知する、という使い方をしていきたいと思います。

2. 今回から AWS ドリルを始める方の作業

前述の通り、今月から AWS ドリルを始めます ! という方向けの作業を最初に紹介します。これまでの AWS ドリルは既にやっているよ ! という方は、本記事の「3. 新しい Lambda 関数 twitter-bot-search を作成し、お試しツイートする」まで飛ばしていただければと思います。

2-1. Twitter API を使う準備をする

まずは Twitter 側の初期設定をしていただく必要があります。AWS ドリル #1 の「3. Twitter API を使う準備をする」にある作業を進めてください。この作業の中で取得できる

  • API Key
  • API Key Secret
  • Access Token
  • Access Token Secret

はこの後の工程ですぐに使いますので、お手元のエディタに控えておいていただくようお願いいたします。

2-2. AWS アカウントがない方はアカウントを作る

AWS ドリルで使用する AWS アカウントも用意します。既存の AWS アカウントを使いまわしていただいても構いませんが、AWS ドリルにおける作業が他の AWS リソースに影響を与える可能性があるため、サンドボックス的なアカウントで実施してください。もし、そのような AWS アカウントをお持ちでない方は、アカウント作成ハンズオン を用意していますので、こちらからアカウント作成をお願いします。

その後、操作に必要な IAM 権限のあるユーザーでログインしていただき、マネジメントコンソールの

  • 右上のリージョン選択プルダウンから「アジアパシフィック (東京)」
  • 左下の言語選択プルダウンから「日本語」

を選択してください。

2-3. AWS Systems Manager の Parameter Store の準備

先ほど取得した Twitter API の Key / Token を「安全な格納先」に保存します。こちらは AWS Systems Manager というサービスの Parameter Store という機能を用います。機密度の高い情報をソースコードにハードコーティングするのではなく、「安全な格納先」から取り出して利用する形にします。こちらの作業は、AWS ドリル #3 の「3-1. Parameter Store のパラメータを作成する」をご実施ください。

今回の AWS ドリル #4 で前提とする作業は以上になります。ここから検索＆通知 Bot を作っていきましょう !

3. 新しい Lambda 関数 twitter-bot-search を作成し、お試しツイートする

それでは今回のドリルの本題を進めていきましょう。まずは、Twitter の検索 API を使う前準備を進めます。

3-1. Lambda 関数を新規で作成する

検索＆通知 Bot 用に新しい Lambda 関数を作ります。AWS ドリル #1 から #3 まで使っていた Lambda 関数 twitter-bot-hello とは別の関数として、新規に作成します。AWS Lambda のサービス画面 に遷移し、右上の「関数の作成」をクリックしてください。

関数の基本的な情報には、

  • 関数名 : twitter-bot-search
  • ランタイム : Python 3.9
  • アーキテクチャ : x86_64

を設定し、「関数の作成」をクリックしてください。

クリックすると拡大します

続いて、requests_oauthlib という Python モジュールをインストールします。Twitter API を呼び出すときにこのモジュールを利用するのですが、標準の Lambda 関数には含まれておりません。そのため、requests_oauthlib をインストールし、それを .zip ファイルにまとめ、Lambda 関数にソースコードとしてアップロードします。

今回も AWS CloudShell 上で、作業をします。マネジメントコンソールの右上のコンソールマークをクリックしてください。

クリックすると拡大します

CloudShell のコンソールが起動したら、下記のコマンドを実行してください。(最後のコマンドの後にもエンターを押すのを忘れないように気をつけてください。)

mkdir twitter-bot-search
cd twitter-bot-search/
pip3 install -U pip
pip install requests_oauthlib -t ./
touch lambda_function.py
zip -r twitter-bot-search.zip ./
aws lambda update-function-code --function-name twitter-bot-search --zip-file fileb://twitter-bot-search.zip

無事にコマンドが流れ終わったら、先ほど作成した Lambda 関数 twitter-bot-search を再度表示 (既に表示している人はブラウザをリロード) してください。

エディタの左側に依存パッケージ群が配置され、一番下には lambda_function.py があるはずです。こちらの Python ファイルをダブルクリックして開いてみると、中身が空のファイルが右側に表示されるはずです (現時点で中身は空で大丈夫です)。

クリックすると拡大します

3-2. Lambda 関数に紐づく IAM ロールの修正

早速コードを書いていきたいところなのですが、先に Lambda 関数に紐づく IAM ロールの設定を行います。Lambda 関数から他の AWS サービス・リソースを操作するには、その操作に必要な権限を与える必要があります。その権限付与を IAM ロールで行うのでしたね。(こちらの説明は AWS ドリル #2 の「4. Lambda 関数から DynamoDB テーブルにアクセスする実装を進める」の最後にしていますので、あわせてご覧ください。 )

この時点では、Parameter Store に格納された Twitter API に関する情報にアクセスだけできれば大丈夫です。Lambda 関数に紐づく IAM ロールにその権限を与えるための操作を進めていきましょう。

Lambda 関数 twitter-bot-search の画面を少し上の方にスクロールしていただくと、現在は「コード」タブが選択されている状態だと思います。こちらを「設定」タブに切り替えていただき、その後「アクセス権限」を押してください。

その後、実行ロールにある「ロール名」のリンクをクリックしてください。

クリックすると拡大します

IAM ロールの設定画面に遷移したら、「アクセス許可を追加」から「ポリシーをアタッチ」を選択してください。

クリックすると拡大します

IAM ポリシーは利用者が作成することができ、本来であれば必要最低限の権限に絞るべきなのですが、今回は標準で用意されている IAM ポリシーを使います。上部の検索窓に「SSMRead」と入力し、エンターを押してください。

クリックすると拡大します

すると IAM ロールがひとつに絞られるはずなので、「AmazonSSMReadOnlyAccess」の左側にチェックを入れ、「ポリシーをアタッチ」してください。

クリックすると拡大します

3-3. Lambda 関数のコードを修正する

ここまでで Lambda 関数 のひな形が完成です。続いて、先ほど開いた lambda_function.py に下記のプログラムをコピー＆ペーストしてみてください。今回からドリルを始めた方向けに少し補足すると、init メソッドは Twitter API を呼び出す準備を、tweet メソッドは Twitter API のツイート API の呼び出しを担当しています。前回のドリル #3 で詳細に説明しているので、よければあわせてご覧ください。この記事としては、まずはこのプログラムをひな形として利用します。

import json
import os
from requests_oauthlib import OAuth1Session
from datetime import datetime, timedelta, timezone

import boto3
ssm_client = boto3.client('ssm')

oauth = None

def lambda_handler(event, context): 
    init()
    
    # 同じ文言のツイートができないので、その対策で年月日を入れておきます。
    now_jst = datetime.now(timezone(timedelta(hours=+9), 'JST'))
    now_jst_str = now_jst.strftime('%Y年%-m月%-d日')
    tweet_text = 'おはようございます、今日は ' + now_jst_str + ' です。'
    
    tweet(tweet_text)
    
def init():
    response = ssm_client.get_parameter(
        Name='/credentials/twitter',
        WithDecryption=True
    )
    twitter_parameters = json.loads(response['Parameter']['Value'])

    consumer_key = twitter_parameters['consumer_key']
    client_secret = twitter_parameters['client_secret']
    access_token = twitter_parameters['access_token']
    access_token_secret = twitter_parameters['access_token_secret']
    
    global oauth
    oauth = OAuth1Session(consumer_key, client_secret, access_token, access_token_secret)
        
def tweet(text):
    payload = {'text': text}
    response = oauth.post(
        'https://api.twitter.com/2/tweets',
        json=payload,
    )
    if response.status_code != 201:
        raise Exception(
            '[Error] {} {}'.format(response.status_code, response.text)
        )

Lambda 関数のコードを変更したら忘れずに Deploy しましょう。Deploy 後、Test ボタンをクリックすると「テストイベントの設定」というポップアップが出てくると思います。

Lambda 関数のテスト実行を行う際には、event にどのような情報を入れてテストするかを決めることができます。今回の Lambda 関数は event に格納された情報は使わないので、名前だけ決めていただき (Test のような名前で構いません)、内容はそのままでテストイベントを保存してください。

もう一度 Test ボタンをクリックすると Lambda 関数が実行されます。ご自身の Twitter アカウントをご確認いただき、ツイートできていれば OK です🎉

クリックすると拡大します

4. Twitter の検索 API を使ってみる

ここからが今回の本題で、Twitter の検索 API を使い、キーワードを含むツイートを検索する実装を進めます。Twitter の検索 API に関するドキュメント を見ると、直近 7 日間を対象に検索する /search/recent エンドポイント と、全ての期間を対象にする /serach/all エンドポイント の 2 種類があることがわかります。しかし、後者は Academic Research access でしか利用できないので、今回は前者の /search/recent エンドポイントを使います。まずは API ドキュメント をご一読ください。

様々なクエリパラメータが用意されていますが、今回は下記の 4 つのパラメータを利用します。

  • query - 検索する文字列を入力します。今回はここに「目黒線」のようなシンプルな形で文字列を入れますが、こちらのドキュメント にもあるように、複雑なクエリを書くことも可能です。
  • max_results - 検索結果として取得する最大件数を定義します。今回のドリルではここを 100 としますが、実際は通知のしきい値より少し大きい数にすれば十分です。（しきい値を 20 にするのであれば、この値は 21 で十分。）
  • start_time と end_time - 検索対象とする期間を指定します。YYYY-MM-DDTHH:mm:ssZ のフォーマットにする点、UTC で指定する点に注意が必要です。例えば、4/4 の朝に Bot を動かす際に、その当日の 5:00 から 6:00 を指定したいとします。この場合、日本時間から UTC に変換し、かつフォーマットをあわせる必要があるので、2022-04-03T20:00:00Z から 2022-04-03T21:00:00Z という形で指定することになります。


まずはしきい値を決めるために、キーワードが普段どれくらいツイートされているかを調査しながら検索 API に慣れていきましょう。はじめに、検索処理を行うメソッド count_matching_tweets を作ります。検索対象とするキーワードを search_word, 検索期間を start_time , end_time として受け取り、それを param に格納して検索 API を呼び出す実装にします。先ほどまでのコードに下記の count_matching_tweets メソッドを追加してみてください。

def count_matching_tweets(search_word, start_time, end_time):

    url = 'https://api.twitter.com/2/tweets/search/recent'
    params = {
        'query': search_word,
        'max_results': 100,
        'start_time': start_time,
        'end_time': end_time
    }
    response = oauth.get(
       url, params=params
    )
    json_response = response.json()
    
    print(json_response)
    
    if response.status_code != 200:
        raise Exception(
            '[Error] {} {}'.format(response.status_code, response.text)
        )

    return json_response['meta']['result_count']

そして、lambda_handler からこのメソッドを呼び出してみましょう。現時点では start_time と end_time はハードコーディングしてしまって構いません。UTC で指定する点と直近 7 日間しか対象にできない点に注意し、現在の日時 (皆さまがこのドリルを進めている日時) に合わせて start_time と end_time を変更してください。また、同じ文章をツイートしようとするとエラーになるので、tweet メソッドを呼び出す部分は一時的にコメントアウトしましょう。

例えば、下記の実装であれば、日本時間の 2022/3/3 に投稿されたキーワード (ここでは「目黒線」) を含むツイートが、最大 100 件取得できます。

def lambda_handler(event, context): 
    init()

    now_jst = datetime.now(timezone(timedelta(hours=+9), 'JST'))
    now_jst_str = now_jst.strftime('%Y年%-m月%-d日')
    tweet_text = 'おはようございます、今日は ' + now_jst_str + ' です。'
    
    # ★追加（ここから）
    # Twitter API には UTC で渡す必要があるので、UTC のままにしています。
    now = datetime.now()
    start_time = '2022-03-02T15:00:00Z' # 日付を作業日付2日前にしてください
    end_time = '2022-03-03T15:00:00Z'   # 日付を作業日付前日にしてください
    count = count_matching_tweets('目黒線', start_time, end_time)
    # ★追加（ここまで）
    
    # ★コメントアウト
    #tweet(tweet_text)

ソースコードの修正が終わりましたら、Deploy し、改めて実行してみてください。下記のような JSON が表示されれば成功です。

{'data': [
    {'id': 'XXXXXXXXXXXXXXXXXXX', 'text': '<「目黒線」を含むツイート>'}, 
    {'id': 'XXXXXXXXXXXXXXXXXXX', 'text': '<「目黒線」を含むツイート>'}, 
    {'id': 'XXXXXXXXXXXXXXXXXXX', 'text': '<「目黒線」を含むツイート>'}, 
    {'id': 'XXXXXXXXXXXXXXXXXXX', 'text': '<「目黒線」を含むツイート>'}
  ], 
 'meta': {
   'newest_id': 'XXXXXXXXXXXXXXXXXXX',
   'oldest_id': 'XXXXXXXXXXXXXXXXXXX',
   'result_count': 4}
}

result_count の値は、検索キーワードや期間によって変わってくると思いますが、この例だと 4 つのツイートが検索にヒットしたことになります。せっかくなので、キーワードや期間を変えて、結果がどのように変わるか試してみてください。

検索 API の使い方に慣れてきましたら、通知のしきい値を決めるための調査をしましょう。例えば私の場合は、5:00-6:00 に「目黒線」を含む投稿が平常時はどれくらいあるのかを Bot に検索させたいので、下記のような形で start_time と end_time を変更し調査します (UTC に変換しています)。

  • start_time : '2022-03-03T20:00:00Z'
  • end_time : '2022-03-03T21:00:00Z'

そして日付を変えながら何日分か確認してみましょう。その際、下記の 2 点にお気をつけください。

  • 7 日前より遡る使い方はできない (遡るとエラーになる)
  • end_time に未来日時は入れられない

後者について補足すると、あまりにもギリギリ、例えば end_date に 2022-03-03T21:00:00Z を入れる、つまり日本時間で 3/4 6:00 を指定した上で、API 自体も 3/4 6:00 に呼び出してしまうとエラーになることがあります。(この背景から、後ほど設定する Bot を動かすタイミングを 6:00 ジャストからずらして、6:01 にしています。)

平常時にキーワードに関するツイートがどれくらいあるかを確認できましたでしょうか ? 私の手元だと、5:00-6:00 の「目黒線」に関するツイートは一桁に収まることが多いこと、ただし遅延がある日は 50 件を超えることが分かりました。そこで、通知のしきい値を 20 にすることに決め、後続の実装を進めていこうと思います。

5. 毎朝自動的にツイート検索するように設定する

しきい値を決めることができたので、次は自動的に毎朝検索するように実装を進めましょう。改めて要件を再掲すると、

  • 毎朝 6:01 に、5:00-6:00 の「目黒線」を含むツイートを検索する。
  • 検索結果をツイートする。
  • 検索に該当する件数が 20 を超えていたら、@ketancho という文字列をつけ通知が飛ぶようにする。
  • また、自身のメールアドレス宛にメールも送るようにする。

となります。

最後のメール通知の部分は Amazon SNS を利用して記事の最後に実装するので、先にそれ以外の部分を進めていきましょう。この部分については、既にドリル #1 から #3 に取り組んでいただいた方にとっては、知っている知識だけで実装ができるはずです。慣れてきた方や力試ししたい方は、構築例を見る前にご自身で実装していただき、そのあとで私の例を確認するという流れで進めてみてもいいかもしれません。

5-1. Lambda 関数のコードを修正する

まずはこれまでハードコーディングしていた検索日時 start_time と end_time を動的に変更したいと思います。この部分は (Amazon EventBridge 経由で) 毎朝 6:01 に Lambda 関数が動くことを前提とし、start_time は Lambda 関数実行時間の1時間前、end_time は実行時間のままとし、分と秒は 00 で固定としました。start_time には当日の 5:00、end_time には当日の 6:00 が入る形です (実際の値は UTC なので、-9 時間した値が入ります)。

    start_time = (datetime.now() + timedelta(hours=-1)).strftime('%Y-%m-%dT%H:00:00Z')
    end_time = datetime.now().strftime('%Y-%m-%dT%H:00:00Z')

続いて、しきい値を超えたときだけ @ ツイートするようにツイート文面を組み立てていきます。私は件数がどれくらいだったかも投稿するようにしてみました。また、@ ツイートする先は、ご自身の Twitter アカウントに置き換えてください。

    tweet_text = tweet_text + '今朝の「目黒線」関連のツイートは ' + str(count) + ' 件でした。'
    if 20 < count:
        tweet_text = tweet_text + 'いつもより多めなので気をつけてください。 @ketancho'

最後に、下記の 2 点を修正します。

  • ひとつ前の作業で、ツイートする部分をコメントアウトしていたので元に戻す
  • 逆に count_matching_tweets メソッド内で、検索 API のレスポンスをログ出力していましたが、ログが増えてしまうので削除 or コメントアウトする

以上のポイントを反映させた Lambda 関数全体のコード例がこちらです。

import json
import os
from requests_oauthlib import OAuth1Session
from datetime import datetime, timedelta, timezone

import boto3
ssm_client = boto3.client('ssm')

oauth = None

def lambda_handler(event, context): 
    init()
    
    now_jst = datetime.now(timezone(timedelta(hours=+9), 'JST'))
    now_jst_str = now_jst.strftime('%Y年%-m月%-d日')
    tweet_text = 'おはようございます、今日は ' + now_jst_str + ' です。'
    
    # ★修正（ここから）
    # Twitter API には UTC で渡す必要があるので、UTC のままにしています。
    now = datetime.now()
    start_time = (datetime.now() + timedelta(hours=-1)).strftime('%Y-%m-%dT%H:00:00Z')
    end_time = datetime.now().strftime('%Y-%m-%dT%H:00:00Z')
    count = count_matching_tweets('目黒線', start_time, end_time)
    
    tweet_text = tweet_text + '今朝の「目黒線」関連のツイートは ' + str(count) + ' 件でした。'
    if 20 < count:
        tweet_text = tweet_text + 'いつもより多めなので気をつけてください。 @ketancho'
    # ★修正（ここまで）
    
    # ★少し前にコメントアウトしていたので戻す
    tweet(tweet_text)
    
def init():
    response = ssm_client.get_parameter(
        Name='/credentials/twitter',
        WithDecryption=True
    )
    twitter_parameters = json.loads(response['Parameter']['Value'])

    consumer_key = twitter_parameters['consumer_key']
    client_secret = twitter_parameters['client_secret']
    access_token = twitter_parameters['access_token']
    access_token_secret = twitter_parameters['access_token_secret']
    
    global oauth
    oauth = OAuth1Session(consumer_key, client_secret, access_token, access_token_secret)
        
def count_matching_tweets(search_word, start_time, end_time):
    url = 'https://api.twitter.com/2/tweets/search/recent'
    params = {
        'query': search_word,
        'max_results': 100,
        'start_time': start_time,
        'end_time': end_time
    }
    response = oauth.get(
       url, params=params
    )
    json_response = response.json()
    
    # ★削除 or コメントアウトする
    # print(json_response)
    
    if response.status_code != 200:
        raise Exception(
            '[Error] {} {}'.format(response.status_code, response.text)
        )

    return json_response['meta']['result_count']  

def tweet(text):
    payload = {'text': text}
    response = oauth.post(
        'https://api.twitter.com/2/tweets',
        json=payload,
    )
    if response.status_code != 201:
        raise Exception(
            '[Error] {} {}'.format(response.status_code, response.text)
        )

以上で Lambda 関数の修正は完了です。Deploy し、期待する動きになるか確認してみましょう。

5-2. Amazon EventBridge の設定

続いて、この Lambda 関数が毎朝 6:01 に動くようにしてみましょう。

Lambda 関数の画面を上の方にスクロールしていただくと、「トリガーを追加」というボタンがありますので、こちらをクリックしてください。

クリックすると拡大します

遷移後の画面で、プルダウンから EventBridge (CloudWatch Events) を選択していただき、

  • 新規ルールの作成
  • ルール名 : exec-twitter-bot-search-0601-every-morning
  • ルールタイプ : スケジュール式
  • スケジュール式 : cron(1 21 ? * * *)

として、追加してください。

(こちらの設定方法やスケジュール式の考え方は、AWS ドリル #1 の「4-4. 朝 8 時に自動投稿するようにする」で解説していますので、不明点がございましたらあわせて参照してください。 )

クリックすると拡大します

実装がうまくいっていましたら、毎朝このようなツイートがされるはずです🎉

しきい値を超えるテストがしにくいかもしれませんが、例えばテストのときだけはしきい値を小さい値にするなどして、期待する動きになるかを確認してみてください。

クリックすると拡大します

6. しきい値を超えたときには Amazon SNS を使ってメール通知もする

ここまでで Twitter 上での通知 (@ メンション) が飛んでくるようになりました。同様にメールでも通知されるといいな、という方もいらっしゃると思うので、しきい値を超えたときにはメールでも通知するように変更し、この記事を締めたいと思います。

Amazon SNS 側の設定と、Lambda 関数の修正の 2 点を進めます。

6-1. Amazon SNS 側の設定

まずは、Amazon SNS の「トピック」というリソースを作ります。Amazon SNS の画面 に遷移していただき、右側の「トピックの作成」にトピック名 TwitterBotTopic を入力後、「次のステップ」をクリックしてください。

クリックすると拡大します

次の画面ではトピックに関する様々な設定が行えるのですが、今回のドリルでは特に変更しないので、そのまま下にスクロールし、「トピックの作成」をクリックしてください。

クリックすると拡大します

ここまでの作業でトピックができあがりました。このトピックに対してメッセージを Publish すると、トピックを Subscribe している対象にそのメッセージが届きます。

トピックの詳細欄に ARN という項目がありますが、後ほど Lambda 関数からメッセージを Publish する際にこの ARN を使いますので、お手元のエディタに文字列を控えておいてください。

クリックすると拡大します

続いてサブスクリプションの設定を進めましょう。今回は E メールプロトコルを指定したサブスクリプションを作成し、Lambda 関数から発行するメッセージを受信するようにします。

作成したトピックの画面を少し下にスクロールし、サブスクリプションタブの「サブスクリプションの作成」をクリックしてください。

クリックすると拡大します

サブスクリプションの作成では、

  • プロトコル  : E メール
  • エンドポイント : 皆さまが受信可能なメールアドレス

を指定し、右下の「サブスクリプションの作成」をクリックしてください。

クリックすると拡大します

少し待つと、指定したメールアドレスにこちらのようなメールが届きます。こちらは、サブスクリプションとしてこのメールアドレスが登録されたけど問題ないか ? という確認のためのメールになります。こちらの「Confirm subscription」リンクをクリックすると、サブスクリプションとしての登録が完了します。

クリックすると拡大します

トピックのサブスクリプション一覧で、サブスクリプションが「確認済み」となっていれば登録完了です。

クリックすると拡大します

6-2. Lambda 関数 twitter-bot-search の修正

続いて、Lambda 関数を修正し、メール通知できるように仕上げていきます。プログラムを修正する前に、Amazon SNS の操作権限を Lambda 関数に追加しましょう。手順は本記事の「3. 新しい Lambda 関数 twitter-bot-search を作成し、お試しツイートする」を参照していただければと思うのですが、

  • IAM ロールの検索窓には「SNSFull」と入力する
  • IAM ロールは「AmazonSNSFullAccess」を選択する

の 2 点が先ほどと異なる点になります。(記事の前半でも触れましたが、本来は必要最低限の権限に絞る設定とすべきです。今回は用意されている IAM ポリシーを使いますが、権限を絞った IAM ポリシーの作成方法については後ほど参考情報としてハンズオンを紹介します。)

IAM ロールの修正ができたら、Lambda 関数のコードを修正していきましょう。今回もこれまでのドリルと同じく、AWS SDK for Python のドキュメント を確認し、プログラムから Amazon SNS を利用する方法を確認しながら進めていきます。今回は こちらの Amazon SNS に関する部分 が該当箇所となります。

最初に Amazon SNS のクライアントを作る 方法であったり、

import boto3

client = boto3.client('sns')

トピックに Publish する 方法であったりが示されているので、このあたりを参考にして実装を進めるとよいでしょう。

response = client.publish(
    TopicArn='string',
    TargetArn='string',
    PhoneNumber='string',
    Message='string',
    Subject='string',
    # 省略

Publish する際は、TopicArn に加え、Subject (メールタイトル) と Message (メール本文) の 3 つのパラメータを利用します。

  • TopicArn : 先ほど Amazon SNS を作成したときに控えたトピックの ARN
  • Subject : Twitter Bot からの連絡
  • Message : tweet_text (先ほどツイートするように作った文言そのまま)

としたいと思います。

これまでの AWS ドリルを進められている方は、Amazon SNS を使うのが初めてです ! という方でも、AWS SDK ドキュメントを読みながら新しいサービスを使っていくのに慣れてきていると思いますので、ぜひこの部分も腕試し的に実装を先に進め、そのあと回答例を見る形にしていただければと思います。

 

 

 

それでは、コード例を紹介していきます。lambda_handler メソッドの中しか変更していませんが、このドリルの最終形でもあるので、念のため全てのコードを示したいと思います。

  • TopicArn は皆さまの Amazon SNS トピックの ARN に置き換えていただく必要があります
  • 検索結果が 20 件を超えたときの通知先を皆さまが所有する Twitter アカウントに変えてください (そのままコピペすると私に @ メンションが飛んでしまいます🙏)

あたりが注意点になります。

import json
import os
from requests_oauthlib import OAuth1Session
from datetime import datetime, timedelta, timezone

import boto3
ssm_client = boto3.client('ssm')
# ★追加
sns_client = boto3.client('sns')

oauth = None

def lambda_handler(event, context): 
    init()
    
    now_jst = datetime.now(timezone(timedelta(hours=+9), 'JST'))
    now_jst_str = now_jst.strftime('%Y年%-m月%-d日')
    tweet_text = 'おはようございます、今日は ' + now_jst_str + ' です。'
    
    # Twitter API には UTC で渡す必要があるので、UTC のままにしています。
    now = datetime.now()
    start_time = (now + timedelta(hours=-1)).strftime('%Y-%m-%dT%H:00:00Z')
    end_time = now.strftime('%Y-%m-%dT%H:00:00Z')
    
    count = count_matching_tweets('目黒線', start_time, end_time)
    tweet_text = tweet_text + '今朝の「目黒線」関連のツイートは ' + str(count) + ' 件でした。'
    if 20 < count:
        tweet_text = tweet_text + 'いつもより多めなので気をつけてください。 @ketancho'
        # ★追加（ここから）
        sns_client.publish(
            TopicArn='*Input Your Topic ARN*',
            Subject='Twitter Bot からの連絡',
            Message=tweet_text
        )
        # ★追加（ここまで）
    tweet(tweet_text)
    
def init():
    response = ssm_client.get_parameter(
        Name='/credentials/twitter',
        WithDecryption=True
    )
    twitter_parameters = json.loads(response['Parameter']['Value'])

    consumer_key = twitter_parameters['consumer_key']
    client_secret = twitter_parameters['client_secret']
    access_token = twitter_parameters['access_token']
    access_token_secret = twitter_parameters['access_token_secret']
    
    global oauth
    oauth = OAuth1Session(consumer_key, client_secret, access_token, access_token_secret)
        
def count_matching_tweets(search_word, start_time, end_time):
    url = 'https://api.twitter.com/2/tweets/search/recent'
    params = {
        'query': search_word,
        'max_results': 100,
        'start_time': start_time,
        'end_time': end_time
    }
    response = oauth.get(
       url, params=params
    )
    json_response = response.json()
    
    # print(json_response)
    
    if response.status_code != 200:
        raise Exception(
            '[Error] {} {}'.format(response.status_code, response.text)
        )

    return json_response['meta']['result_count']  

def tweet(text):
    payload = {'text': text}
    response = oauth.post(
        'https://api.twitter.com/2/tweets',
        json=payload,
    )
    if response.status_code != 201:
        raise Exception(
            '[Error] {} {}'.format(response.status_code, response.text)
        )

実装ができたら、Lambda 関数を実行してみてください。しきい値を超えたときには (このときだけテスト用にしきい値を下げてもいいかもしれません)、Twitter 側の @ ツイートとともに、このようなメールが送られてくれば成功です🎉

クリックすると拡大します

7. 落ち穂拾い＆参考情報

今回は新しい AWS サービスとして Amazon SNS を使いました。今回はメールで通知したので「メールを送るサービス」と捉えてしまう方がいるかもしれないのでその点だけ補足します。

Amazon SNS はクラウドからの通知のセットアップ、操作、および送信を容易にするウェブサービスです。アプリケーションからメッセージを発行し、サブスクライバーや他のアプリケーションに迅速に配信するための、スケーラブルかつ柔軟で、費用対効果の高い機能を備えています。

こちらの送信プロトコルのひとつとして今回は「E メール」を選択しましたが、それ以外にも多くのプロトコルが提供されており、皆さまの馴染み深いところですと、「Lambda 関数を呼び出す」こともできたりします。Amazon SNS がよく使われるシナリオとしては、Amazon SNS の開発者ガイド にも記載がありますので、こちらも参照してみてください。また、Amazon SNS の FAQ もご一読いただくと、更に Amazon SNS への理解が深まると思います。こちらもご覧いただければと思います。

最後に、今回の記事に関連するオンデマンドハンズオンを紹介します。AWS ドリル #2 でも紹介しましたが、今回から AWS ドリルを始めました ! という方もいらっしゃると思うので、改めての紹介になります。

こちらのハンズオンでは、アカウントの作成と IAM のハンズオンをご実施いただきます。今回のドリルでは、用意されている IAM ポリシーをお使いいただいていますが、実際の業務では必要最低限の権限に絞っていただくことを推奨します。そのため、用意されているものではなく、ご自身で必要な IAM ポリシーを定義できる必要があります。こちらのハンズオンは、オンデマンド形式でいつでも無償でご覧いただけるものですので、IAM の基本を学ぶ際にはぜひご活用いただければと考えています。

Hands-on for Beginners - ハンズオンはじめの一歩: AWS アカウントの作り方 & IAM 基本のキ

AWS アカウントの作成と、AWS IAM の基本を学べるハンズオンです。今回のドリルでは、IAM の「体験」から入っていただきましたが、こちらの動画ハンズオンでは AWS IAM の座学、また IAM ユーザーや IAM グループといった、IAM のその他のリソースについても手を動かしながら学んでいただけます。ぜひこちらもご覧いただければと思います。

8. まとめ

「お役立ち Twitter Bot を作りながら学ぶ AWS ドリル」連載の第 4 弾として、気になるワードを定期的に検索 & 通知する Twitter Bot を作りました。これまでの AWS ドリルに比べて、実用度が高い Bot に仕上がったのではないかと思います。私は、朝の電車遅延に苦しまないように ! というまじめな (？) テーマにしてしまいましたが、もっとワクワクするテーマを設定いただき、人気の Twitter Bot に育てていただければ嬉しいです🙌

AWS の新しいサービスを使ってみる、という観点では Amazon SNS を利用していただきました。Amazon SNS の機能、あるいは使いどころ (ユースケース) は他にも色々ございますので、ぜひこれをきっかけに学びを深めていただければと思います。そして、お仕事で使う際には、「私、Amazon SNS 使ったことあるので、何でも相談していいですからね ! (最高の笑顔)」と周りの方を引っ張っていただければと思います。

最後に改めてのお願いなのですが、皆さまのご感想・ご要望を #AWSウェブマガジン タグをつけて呟いていただけたら嬉しいです🙏 この記事に +α して、こんなものを作ってみました ! という共有も大歓迎です。今後の連載の参考にさせてもらいたいと思っておりますので、ぜひよろしくお願いしますー !

それでは次回の記事でまたお会いしましょう !

この連載記事のその他の記事はこちら

選択
  • 選択
  • 第 1 回 おはよう Bot 編
  • 第 2 回 昔書いた記事の宣伝 Bot 編
  • 第 3 回 リファクタリング ＆ 曜日ごとのツイート 編
  • 第 4 回 新章突入 ! 気になるワード検索 & 通知 Bot 編
  • 第 5 回 皆さまの代わりに英語でツイートしておくよ Bot 編
  • 第 6 回 AWS Step Functions を使って Well-being Bot を作ろう ! (前編)
  • 第 7 回 AWS Step Functions を使って Well-being Bot を作ろう ! (後編)
  • 第 8 回 - 総集編 & 夏休みの自由研究のすすめ
  • 第 9 回 - IaC 入門しながら作るリマインダー Bot (前編)
  • 第 10 回 - IaC 入門しながら作るリマインダー Bot (中編)
  • 完結回 - IaC & CI/CD 入門しながら作るリマインダー Bot (後編)

builders.flash メールメンバーへ登録することで
AWS のベストプラクティスを毎月無料でお試しいただけます

メンバー特典を受け取る » 

筆者プロフィール

金澤 圭 @ketancho
アマゾン ウェブ サービス ジャパン合同会社
技術統括本部 ソリューションアーキテクト

サーバーレスが好きなソリューションアーキテクト。業種業界問わず、お客様のプロダクト開発をサポートさせていただいています。「AWS Hands-on for Beginners」というオンデマンドで視聴できるハンズオンも企画・推進しており、楽しく学べるコンテンツを日々考えています。好きなサービスは AWS Lambda、AWS Step Functions、Amazon Personalize で、好きな休日の過ごし方は娘ふたりと川の字になって昼寝👧👧と赤ん坊を抱っこしながらの散歩です👶

 ツイート »
 シェア »
 はてブ »

他の記事の一覧はこちら »

さらに最新記事・デベロッパー向けイベントを検索

下記の項目で絞り込む
絞り込みを解除 ≫
1 …
…

AWS を無料でお試しいただけます

AWS 無料利用枠の詳細はこちら ≫
5 ステップでアカウント作成できます
無料サインアップ ≫
ご不明な点がおありですか?
日本担当チームへ相談する