【2025年最新版】X Bot を作りながら学ぶ AWS ドリル - #2 Amazon EventBridge を活用して Year Progress Bot を作ろう

AWS Lambda, AWS CloudShell, Amazon EventBridge

こんにちは、@ketancho です。

先日、3年ぶりに『X Bot を作りながら学ぶ AWS ドリル』連載を再開しました。

www.ketancho.net

多くの方にご覧いただけ、大変ありがたかったです。また、はてなブログさんにもおすすめエントリーとしてご紹介いただくことができました🎉(ありがたやー!)

AWS を使えるようになりたい!と思っている方が、座学に加えて『手を動かす』ことは非常に重要だと考えています。では何から手を動かせばいいの?という方々にとって、このハンズオン形式の AWS ドリル連載が一助になれば大変嬉しいです。

さて、この記事はそんな AWS ドリル連載の2記事目として、新しい AWS サービスを活用しながら、ボットに新しい機能を追加していこうと思います。ぜひ実際に手を動かしながら進めていただければと考えているのですが、前回の記事の

の2箇所は、今回の記事の前提条件となるので、まだ前回の記事を試されていない方は、あわせてご覧いただければと思います。

▼ この記事の目次

この記事で行うこと&追加する X Bot の機能

この記事では、その年の何%が経過したかをポストする、いわゆる Year Progress Bot の機能を追加していこうと思います。

  • Amazon EventBridge を活用し、毎時00分に AWS Lambda を定期実行する
  • 1年が0.1%進捗するたびに「進捗どうですか?」と聞いてくる うるさい 親切な機能を追加する

この記事で利用する AWS サービス

  • AWS Lambda
  • AWS CloudShell
  • Amazon EventBridge ← New!!

AWS Lambda 関数のひな形を作成する

AWS Lambda の画面に遷移し、下記の条件で Lambda 関数を作成してください。

  • 関数名 : x-bot-25-year-progress
  • ランタイム : Python 3.13
  • アーキテクチャ : x86_64

(Lambda 関数の作成手順がいまいち分からん、という方は 前回の記事の AWS Lambda 関数のひな形を作成する をご確認ください。)

続いて、AWS CloudShell を起動し、下記のコマンドを実行してみてください。(今回も Tweetpy モジュールをインストールします。)

mkdir x-bot-25-year-progress
cd x-bot-25-year-progress/
pip3 install -U pip
pip install tweepy -t ./
touch lambda_function.py
zip -r x-bot-25-year-progress.zip ./
aws lambda update-function-code --function-name x-bot-25-year-progress --zip-file fileb://x-bot-25-year-progress.zip

条件を満たしたときに X にポストする AWS Lambda 関数を実装する

作成した Lambda 関数において、下記の実装を行います。もし、実装の部分にチャレンジしてみたい!という方は、このあと掲載する参考ソースコードを見ずに、まずは実装してみてもいいかもしれません。

  • Lambda 関数が呼び出された(後ほど毎時0分に呼び出されるようにします)ときに、その年の1/1 0時を起点に何時間が経過したかを計算する
  • その1年の総時間を計算する(365 日/年 x 24 時間/日 ※うるう年であれば366日として計算する)
  • ↑の2つの値から何%が経過したかを小数点第1位まで求める(小数点第2位を切り捨て)
  • 同じ計算を1時間前に行い、この1時間でこの値が変化していたら(例: 1時間前 18.5% → 今 18.6%)、%をツイートをする
  • その年の1/1 0時だけは異なるメッセージをポストする

それでは、私が作った参考ソースコードを載せておきます。(なお、生成 AI 先生に簡単なコードレビューはしてもらっていますが、変数名など拙い部分はご容赦ください。また、この条件下だと動き変にならない?ということにお気づきになられたら、こっそり教えて下さい🙏)

こちらのソースコードを先ほどの Lambda 関数に貼り付けていただき、忘れずに Deploy しておきましょう。(X の API Key も忘れずに皆さまのものに置き換えてくださいね!)

import json
import os
import tweepy
from datetime import datetime, timedelta, timezone
import calendar

# API Key, Access Token (本来、ハードコーディングするのは望ましくありません。連載の中で修正していきます。)
consumer_key = '*Input Your API Key*'
client_secret = '*Input Your API Key Secret*'
access_token = '*Input Your Access Token*'
access_token_secret = '*Input Your Access Token Secret*'

def lambda_handler(event, context):
    """
    - Lambda 関数が呼び出された(後ほど毎時0分に呼び出されるようにします)ときに、その1年間で何時間経過したかを計算する
    - その1年の総時間を計算する(例:365 日/年 x 24 時間/日 ※うるう年でなければ)
    - ↑の2つの値から何%が経過したかを小数点第1位まで(小数点第2位を切り捨て)求める
    - 同じ計算を1時間前に行い、この1時間でこの値がインクリメントされていたら(例: 1時間前 18.5% → 今 18.6%)、進捗確認するツイートをする
    """
    now = datetime.now(timezone(timedelta(hours=+9), 'JST'))
    
    # その年の経過日(1月1日を0日目としたときに何日目か)
    days_passed_in_the_year = int(now.strftime('%j'))-1

    # その年の経過時(↑の経過時間版)
    hours_passed_in_the_year = 24 * days_passed_in_the_year + now.hour

    # その年の合計時間(うるう年考慮)
    year_total_hours = 366 * 24 if calendar.isleap(now.year) else 365 * 24

    # ①実行されたタイミングのその年の経過時間%を小数点第1位まで求める(小数点第2位を切り捨て)
    percentage_hours_passed_in_the_year = math.floor(hours_passed_in_the_year / year_total_hours * 1000)/10

    # ②実行されたタイミングの1時間前の、経過時間%を小数点第1位まで求める(小数点第2位を切り捨て)
    percentage_hours_passed_in_the_year_1_hour_age = math.floor((hours_passed_in_the_year-1) / year_total_hours * 1000)/10

    # ①≠② であれば経過時間%が進んだことになるので、ツイートする。そうでなければしない。
    if percentage_hours_passed_in_the_year == percentage_hours_passed_in_the_year_1_hour_age:
        return

    if hours_passed_in_the_year != 0:
        tweet_text='\ {0}年の約{1}%が経過しました。今年の進捗どうですか?😎 /'.format(now.year, percentage_hours_passed_in_the_year)
    else:
        # 新年最初は前の年の100%をツイートする
        tweet_text='\ {0}年の100%が経過しました!1年間お疲れ様でした!新しい年もやっていきましょう!💪 /'.format(now.year-1)
    
    tweepy_client = tweepy.Client(
        consumer_key=consumer_key,
        consumer_secret=client_secret,
        access_token=access_token,
        access_token_secret=access_token_secret
    )
    
    tweepy_client.create_tweet(
        text=tweet_text
    )

Amazon EventBridge を活用し、毎時0分に Lambda 関数が起動するようにする

先ほど作成した Lambda 関数を上の方にスクロールすると、左側に「トリガーを追加」というボタンがあるので、それをクリックしてください。

AWS Lambda は様々な AWS サービスや外部サービスをトリガーにし、起動することができます。今回は Amazon EventBridge と連携していきますので、検索窓に "eventbridge" と入力し、「EventBridge(CloudWatch Events)」を選択しましょう。

EventBridge を使うことで、Cron 式でトリガータイミングを起動することができます。

  • ルール: 「新規ルールを作成」を選択
  • ルール名: every-hour-at-0-minutes
  • スケジュール式: cron(0 * ? * * *)

と設定し、右下の「追加」をクリックしてください。

以上で今回の実装は完了です。

X Bot の動きを確認する

さて、実際にうまく動いているかを確認するには、少し待つ必要があります。 1年が 0.1% 進んだらポストするということは、1日に2-3回はポストされるということになるので、8-9時間ほど待っていただけば最初のポストが行われるはずです。

このようにポストされたら成功です🎉

落ち穂拾い:トラシュー Tips

今回ものづくりした Year Progress Bot は8-9時間に1度の Post になるので、うまく動かなかったときのデバッグがしにくいかもしれません。私が手元でやったトラシュー手順を幾つか Tips として紹介しておきます。

ポストしないようにしておく

X API の制限で、短い時間に何度もポストしてしまうと、429 のエラーが出ることがあります。デバッグの際は、X API を呼び出している部分をコメントアウトしたうえで、print 文で投稿内容だけ(あるいは投稿されるか)を確認するといいかもしれません。

    if hours_passed_in_the_year != 0:
        tweet_text='\ {0}年の約{1}%が経過しました。今年の進捗どうですか?😎 /'.format(now.year, percentage_hours_passed_in_the_year)
    else:
        # 新年最初は前の年の100%をツイートする
        tweet_text='\ {0}年の100%が経過しました!1年間お疲れ様でした!新しい年もやっていきましょう!💪 /'.format(now.year-1)
    
    print(tweet_text) # 追加して、これ以降をコメントアウトする
    '''
    tweepy_client = tweepy.Client(
        consumer_key=consumer_key,
        consumer_secret=client_secret,
        access_token=access_token,
        access_token_secret=access_token_secret
    )
    
    tweepy_client.create_tweet(
        text=tweet_text
    )
    '''

日時をエミュレートする

境界値のテストのために(例: 1/1の最初のポストが正しく行われるか)、'datetime' を使って日時をエミュレートすると楽に検証が進められると思います。例えば、下記のように 2025/3/23 19:00:00 を指定すると、条件を満たすはずなので

    #now = datetime.now(timezone(timedelta(hours=+9), 'JST'))
    now = datetime(2025, 3, 23, 19, 0, 0)

こちらで実行すると、

\ 2025年の約22.4%が経過しました。今年の進捗どうですか?😎 /

というメッセージが出力されるはずです。

Next Challenge のススメ

最後に、今回の記事に関連した +α の学び・チャレンジを紹介させていただきます。+α に取り組むことで、学習の効果が2倍にも3倍にもなると思いますので、ぜひ取り組んでいただければと思います。もちろん、私が挙げていないような内容でも、皆さまの関心や業務に応じた内容にしていただくのも Welcome、むしろその方がいいまであるかもしれません。

  • Amazon EventBridge の他の使い方について調べてみましょう。今回は cron 的に「時間になったら」をトリガーしましたが、他の AWS サービスや SaaS サービスとの連携など、Amazon EventBridge は繋ぎのサービスとしての機能を豊富に揃えています。何ができるのかをマネジメントコンソールや、公式 Doc を見ながら確認してみることをおすすめします。
  • もう少し Year Progress Bot を静かにさせてみましょう。今の実装だと、1日に2回も3回も急かしてくるので鬱陶しいかもしれません。今回は小数点第1位まで計算後に比較していますが、これを整数部分まで(小数点第1位を切り捨て)にすることで、3-4日に1度くらいの適度なおせっかい Bot にすることができます。(私もこの記事を公開したらこの実装に変えようと思います。うるさいので🤫)

まとめ

X Bot を作りながら学ぶ AWS ドリル の第2弾として、

  • Amazon EventBridge を活用し、毎時00分に AWS Lambda を定期実行する
  • 1年が0.1%進捗するたびに「進捗どうですか?」と聞いてくる うるさい 親切な機能を追加する

を実装し、Year Progress Bot を作ってみました!ぜひ、今回の学びを活かした面白い X Bot を作っていただき、私までご紹介いただければとても嬉しいです!

次回はまた別の AWS サービスを使って、X Bot を成長させていこうと思います。このようなハンズオン形式で、実際に動くものを作りながら AWS を学ぶ記事を、今後も不定期連載していこうと思います。もし気に入っていただけましたら こちらからこのブログの読者 になっていただく and/or 私の X(Twitter) をフォロー していただけたら嬉しいです🙌

それでは!