このプロジェクトは、Vocalendar.jp からイベント情報を取得し、フィルタリングした上でTwitter (X) に自動投稿するAWS Lambda関数です。
指定した時間帯に開催されるイベントを定期的にチェックし、自動でツイートすることで、イベント情報の告知を効率化します。以前は3つのLambda関数に分かれていた処理(取得、フィルタリング、投稿)を、単一の関数に統合してシンプル化しています。
- イベント取得: Vocalendar APIから当日のイベント情報を取得します。
- 柔軟なフィルタリング:
- 実行時刻を基準とした相対的な時間範囲(例: 3時間後〜4時間後)でイベントを抽出します。
- 環境変数で指定したキーワード(例:
ラジオ,生放送)がタイトルに含まれるイベントを投稿対象から除外します。
- 自動ツイート:
- フィルタリングされたイベントを、1イベント1ツイート形式で投稿します。
- ツイートには、日付(「本日」「明日」など動的に変化)、イベント名、開催時間、詳細リンク、ハッシュタグ(
#vocalendar)が含まれます。
- セキュアな認証情報管理: Twitter APIの認証情報はAWS Secrets Managerを利用して安全に管理します。
以下はリポジトリ内のソースコードで参照されている環境変数の一覧です。 README内の各種設定例と合わせて利用してください。ここに記載している「デフォルト値」は、ソースコード内で環境変数が未設定の場合に使用される値です。
-
OFFSET_START_HOURS
- 説明: フィルタリング期間の開始オフセット(時間)。
- コード上のデフォルト: 3
- 使用箇所:
src/calendar_event_util.py
-
OFFSET_START_MINUTES
- 説明: フィルタリング期間の開始オフセット(分)。
- コード上のデフォルト: 0
- 使用箇所:
src/calendar_event_util.py
-
OFFSET_END_HOURS
- 説明: フィルタリング期間の終了オフセット(時間)。
- コード上のデフォルト: 4
- 使用箇所:
src/calendar_event_util.py
-
OFFSET_END_MINUTES
- 説明: フィルタリング期間の終了オフセット(分)。
- コード上のデフォルト: 0
- 使用箇所:
src/calendar_event_util.py
-
EXCLUDE_TERMS
- 説明: 投稿対象から除外するキーワード(カンマ区切り)。
- コード上のデフォルト: 空文字('')。未設定の場合は除外フィルタ無し。
- 使用箇所:
src/calendar_event_util.py
-
BASE_URL
- 説明: Vocalendar API のエンドポイント。
- コード上のデフォルト:
https://vocalendar.jp/core/events.json - 使用箇所:
src/vocalendar_api.py
-
VOCALENDAR_API_QUERY_LIMIT
- 説明: Vocalendar API から取得するイベント数の上限(クエリパラメータ)。
- コード上のデフォルト: '50'(文字列として扱われます)
- 使用箇所:
src/vocalendar_api.py
-
TWITTER_API_KEY, TWITTER_API_KEY_SECRET, TWITTER_ACCESS_TOKEN, TWITTER_ACCESS_TOKEN_SECRET
- 説明: Tweepy(Twitter API)で OAuth1 を利用する際の API キー/トークン。
- コード上のデフォルト: なし(未設定の場合は Secrets Manager を参照)
- 使用箇所:
src/twitter.py
-
TWITTER_SECRET_NAME
- 説明: Twitter 認証情報を格納した Secrets Manager のシークレット名。
- コード上のデフォルト: なし(指定がないと Secrets Manager から取得できずエラーになります)
- 使用箇所:
src/twitter.py - 備考: Secrets Manager を使う場合、保存するシークレット JSON に以下のキーが必要です:
TWITTER_API_KEY,TWITTER_API_KEY_SECRET,TWITTER_ACCESS_TOKEN,TWITTER_ACCESS_TOKEN_SECRET。
-
TEST_NOW
- 説明: テスト用に現在時刻を上書きする ISO8601 文字列。テスト時のみ使用。
- コード上のデフォルト: 未設定(通常は実行時の現在時刻を使用)
- 使用箇所:
src/time_utils.py
注意:
- このセクションの「デフォルト値」はソースコード内のフォールバック値に合わせて更新しています。運用時は環境変数を明示的に設定してください。
-
Twitter Developer Portal でアプリケーション(プロジェクト/アプリ)を作成し、OAuth1 用の認証情報を取得します。必要なキー/トークンは以下の4つです。
TWITTER_API_KEY(Consumer API Key)TWITTER_API_KEY_SECRET(Consumer API Key Secret)TWITTER_ACCESS_TOKEN(Access Token)TWITTER_ACCESS_TOKEN_SECRET(Access Token Secret)
※コードはこれらの OAuth1 資格情報を使用して
tweepy.Clientを初期化します。環境変数で直接設定するか、Secrets Manager にまとめて保存してTWITTER_SECRET_NAME経由で読み込めます。
- AWSマネジメントコンソールでSecrets Managerを開きます。
- 「新しいシークレットを保存する」を選択します。
- 「その他のシークレットのタイプ」を選択します。
- Secrets Manager に保存する場合は、キー/値ではなくシークレット文字列に JSON を格納するのが便利です。例:
{
"TWITTER_API_KEY": "your_consumer_key",
"TWITTER_API_KEY_SECRET": "your_consumer_secret",
"TWITTER_ACCESS_TOKEN": "your_access_token",
"TWITTER_ACCESS_TOKEN_SECRET": "your_access_token_secret"
}- シークレット名を入力します(例:
twitter/credentials)。Lambdaの環境変数TWITTER_SECRET_NAMEにこの名前を設定してください。 - 設定を保存します。
代替: 簡易的には Lambda の環境変数として上記4つのキーを直接設定しても動作しますが、Secrets Manager を使う方が安全です。
この関数は tweepy などの外部ライブラリに依存しているため、コードと一緒にライブラリをパッケージングしてアップロードする必要があります。
プロジェクトには、このパッケージング作業を自動化する deploy.sh スクリプトが含まれています。
以下のコマンドを実行するだけで、LambdaにアップロードするためのZIPファイルが作成されます。
bash deploy.shスクリプトを実行すると、dist/deployment_package.zip というファイルが生成されます。
- AWS Lambdaコンソールを開き、関数を作成または選択します。
- 「コードソース」セクションで「アップロード元」をクリックし、「.zip ファイル」を選択します。
deploy.shで作成されたdist/deployment_package.zipをアップロードします。- 以下の設定を確認・更新します。
- ランタイム:
Python 3.9以降 - ハンドラ:
lambda_function.lambda_handler
- ランタイム:
注記: deploy.sh は、以前の手動手順(ビルド用ディレクトリの作成、ライブラリのインストール、ZIP圧縮)を自動的に実行します。より高度なデプロイ自動化には、AWS SAM や Serverless Framework などのIaCツールの利用もご検討ください。
Lambda関数にアタッチするIAMロールには、以下の権限が必要です。
AWSLambdaBasicExecutionRole: CloudWatch Logsへのログ書き込みを許可します。- Secrets Managerへのアクセス権限:
{ "Effect": "Allow", "Action": "secretsmanager:GetSecretValue", "Resource": "arn:aws:secretsmanager:<region>:<account-id>:secret:<secret-name>-<random-suffix>" }<region>,<account-id>,<secret-name>はご自身の環境に合わせて書き換えてください。
Lambda関数の設定画面で、以下の環境変数を設定します。
| キー | 説明 | デフォルト値 | 設定例 |
|---|---|---|---|
OFFSET_START_HOURS |
フィルタリング期間の開始オフセット(時間)。 | 3 |
3 |
OFFSET_START_MINUTES |
フィルタリング期間の開始オフセット(分)。 | 0 |
0 |
OFFSET_END_HOURS |
フィルタリング期間の終了オフセット(時間)。 | 4 |
4 |
OFFSET_END_MINUTES |
フィルタリング期間の終了オフセット(分)。 | 0 |
0 |
EXCLUDE_TERMS |
投稿から除外するキーワード(カンマ区切り)。 | (なし) | ラジオ,生放送,テレビ |
BASE_URL |
(任意) Vocalendar APIのURL。 | https://vocalendar.jp/core/events.json |
|
VOCALENDAR_API_QUERY_LIMIT |
(任意) Vocalendar API から取得するイベント数の上限(クエリパラメータ)。 | 50 |
50 |
TWITTER_API_KEY |
(任意) Twitter OAuth1 の Consumer API Key。環境変数で直接指定するか Secrets Manager を使う。 | (なし) | xxxxxxxxxxxxxxxx |
TWITTER_API_KEY_SECRET |
(任意) Twitter OAuth1 の Consumer API Key Secret。 | (なし) | xxxxxxxxxxxxxxxx |
TWITTER_ACCESS_TOKEN |
(任意) Twitter OAuth1 の Access Token。 | (なし) | xxxxxxxxxxxxxxxx |
TWITTER_ACCESS_TOKEN_SECRET |
(任意) Twitter OAuth1 の Access Token Secret。 | (なし) | xxxxxxxxxxxxxxxx |
TWITTER_SECRET_NAME |
(任意) Secrets Managerのシークレット名。環境変数で上記4つを個別に設定する代わりに、JSONでまとめて保存してこの環境変数にシークレット名を設定できます。 | (なし) | twitter/credentials |
TEST_NOW |
(任意) テスト用に現在時刻を上書きする ISO8601 文字列(主にローカルテスト用)。 | (なし) | 2025-11-02T10:00:00+09:00 |
設定例: 上記のデフォルト設定では、Lambdaが実行された時刻の 3時間後 から 4時間0分後 までに開始されるイベントがツイート対象となります。
- Lambda関数のトリガーとして Amazon EventBridge (CloudWatch Events) を追加します。
- 「新しいルールを作成」を選択します。
- ルールタイプとして「スケジュール」を選択します。
- スケジュール式を入力します。例えば、10分ごとに実行する場合は
rate(10 minutes)と設定します。
これでセットアップは完了です。指定したスケジュールでLambdaが実行され、条件に合うイベントがTwitterに投稿されます。
-
lambda_handler(event, context)(src/lambda_function.py)- Lambda 関数のエントリーポイント。処理の流れ(取得 → フィルタリング → 投稿)を orchestrate します。
-
get_vocalendar_events()(src/vocalendar_api.py)- Vocalendar API から当日のイベントデータを取得します。
BASE_URLとVOCALENDAR_API_QUERY_LIMITを参照します。
- Vocalendar API から当日のイベントデータを取得します。
-
filter_events(records)(src/calendar_event_util.py)- 取得したイベントを、
OFFSET_*とEXCLUDE_TERMSに基づく時間範囲・除外キーワードでフィルタリングします。
- 取得したイベントを、
-
format_event_for_tweet(event)(src/calendar_event_util.py)- 個々のイベントからツイート本文を生成します(日時の人間向け表示、リンク、ハッシュタグ等)。
-
post_events_to_twitter(events_data)(src/twitter.py)- フィルタ済みイベントを受け取り、
tweepy.Clientを使ってツイートを投稿します。
- フィルタ済みイベントを受け取り、
-
get_twitter_credentials()(src/twitter.py)- Twitter の OAuth1 資格情報を取得します。優先順は環境変数(
TWITTER_API_KEY等)→TWITTER_SECRET_NAMEで指定した Secrets Manager の JSON です。
- Twitter の OAuth1 資格情報を取得します。優先順は環境変数(
-
get_now()(src/time_utils.py)- 現在時刻を JST で返すユーティリティ。環境変数
TEST_NOWがあればテスト用の時刻に差し替えます。
- 現在時刻を JST で返すユーティリティ。環境変数
-
common(src/common.py)- 共有の logger、JST タイムゾーン定義、Secrets Manager クライアント(
secrets_client)などが定義されています。
- 共有の logger、JST タイムゾーン定義、Secrets Manager クライアント(
注: Secrets Manager に保存する JSON には、TWITTER_API_KEY, TWITTER_API_KEY_SECRET, TWITTER_ACCESS_TOKEN, TWITTER_ACCESS_TOKEN_SECRET のキーを含めてください(TWITTER_SECRET_NAME で参照します)。
以下は、このLambda関数の主要な処理フローを示すチャートです。EventBridgeでスケジュール実行され、Vocalendarからイベントを取得し、フィルタリングしてTwitter(X)に投稿します。Twitterの認証情報は環境変数経由またはSecrets Manager経由で取得されます。
flowchart TD
A[EventBridge スケジュール] --> B(Lambda関数 処理開始)
B --> C[Vocalendar APIでイベント取得]
C --> D[イベントをフィルタリング]
D --> E{投稿対象イベントあり?}
E -->|いいえ| M[終了ログ出力]
E -->|はい| F[ツイート本文を生成]
F --> G{認証情報は環境変数にあるか?}
G -->|はい| H[Twitterに投稿(環境変数認証)]
G -->|いいえ| J[Secrets Managerから認証情報取得]
J --> H
H --> K[Twitter APIでツイート]
K --> M[終了ログ出力]
M --> N(処理終了)
注: 上のフローは実際の認証フローを反映しています。処理は以下の通りです:
- Lambda は Vocalendar API からイベントを取得し、
filter_eventsで投稿対象を抽出します。 - フィルタ結果にイベントがあれば
format_event_for_tweetで本文を生成します。 - 認証情報の取得は優先順で (1) 環境変数
TWITTER_API_KEY等、(2) 環境変数が未設定の場合にTWITTER_SECRET_NAMEで指定した Secrets Manager の JSON を参照します。
簡単な説明:
- EventBridge のスケジュールで Lambda が起動します。
lambda_handlerが Vocalendar API からイベントを取得します(get_vocalendar_events())。
補足: このLambda関数は、処理の各段階(イベント取得・フィルタ・投稿)で状況や結果をCloudWatch Logsに記録します。
- 正常終了時(イベントが投稿された場合・投稿対象が無い場合の両方)には、それぞれの状況に応じたINFOレベルのログが出力されます。
- エラー発生時はERRORレベルのログが記録されます。 ログ内容は運用時のトラブルシュートや動作確認に利用できます。