この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
こんにちは、大前です。
Lambda で RSS リーダー的な物を作ってみましたので、ブログを書いていきます。
ちなみに、Slack には RSS リーダーのアプリケーションがあるので、自己満足的な要素が大きいです。。
RSS を Lambda で取得して、AWS 側で何か処理をさせたい時の参考などにして頂ければ幸いです。
構成
大まかな処理の流れは以下です。
- CloudWatch Events で Lambda 関数を 1時間毎に実行
- Lambda から RSS フィードを取得
- 1時間以内に更新があれば Slack に通知
今回は、厚生労働省が提供している RSS を取得してみる事にしてみます。
やってみた
Slack Incoming WebHooks 設定
Lambda から Slack に通知を行うために、Slack 側にアプリを追加する必要があります。
Slack のメニューから「App」を選択し、"Incoming WebHooks" を検索するなどして表示します。
表示されたアプリを選択するとブラウザなどに遷移するかと思いますので、「Slack に追加」を押下します。
投稿したいチャンネルを選択する項目がありますので、チャンネルを選択して、「Incoming Webhook インテグレーションの追加」を押下します。
今回は "#webhook-test" に通知することにします。
すると、「Webhook URL」というものが表示されます。
これを Lambda にて使用しますので、メモしておきます。
Slack 側の設定は以上です。
ソースの準備
今回は外部ライブラリを使用するので、ローカルでソースコードやライブラリを含めた zip を作成します。
zip 作成にあたっては、下記ブログを参考にさせて頂きました。
Pythonで実装したAWS Lambdaをスケジュール化！
作業ディレクトリ作成
$ cd ~
$ mkdir rssfeed
$ cd rssfeed
feedparser インストール
$ pip install feedparser -t ./pytz インストール
$ pip install pytz -t ./lambda_function.py 作成
$ vi lambda_function.py以下の関数を作成しました。
- RSS_URL
- 指定する RSS を変更する事で、取得する対象を変えられます。
- 今回は対象が RSS 1.0 であるため、RSS 2.0 でもうまく動作するかは確認していません。
- SLACK_CHANNEL
- 通知先のチャンネル。「#」からはじまる必要あり
- SLACK_USERNAME
- 通知する際に表示されるユーザー名。
- SLACK_URL
- Slack Incoming WebHooks にて取得した URL。
- THRESHOLD
- 更新対象をみなす期間の設定。（単位は時間）
- 今回は 1時間毎に関数を実行させるので、1としています。
import json
from datetime import datetime,timezone,timedelta
from urllib.request import Request, urlopen
import feedparser
import pytz
RSS_URL = "https://www.mhlw.go.jp/stf/news.rdf"
SLACK_CHANNEL = "#webhook-test"
SLACK_USERNAME = "RSSリーダー"
SLACK_URL = "https://hooks.slack.com/services/XXXXXXXXXX"
THRESHOLD = 1
def post_slack(text):
send_data = {
"channel": SLACK_CHANNEL,
"username": SLACK_USERNAME,
"text": text
}
payload = "payload=" + json.dumps(send_data)
request = Request(
SLACK_URL,
data=payload.encode("utf-8"),
method="POST"
)
with urlopen(request) as response:
response_body = response.read().decode("utf-8")
def lambda_handler(event, context):
try:
# 対象のRSSフィードを取得
feed = feedparser.parse(RSS_URL)
target_list = []
for entry in feed.entries:
now = datetime.now(pytz.timezone('Asia/Tokyo'))
entry_date = datetime.fromisoformat(entry.updated)
threshold = now - timedelta(hours=THRESHOLD)
if threshold <= entry_date < now:
tmp = entry.title + '\n' + entry.link + '\n' + entry.updated + '\n' target_list.append(tmp) if len(target_list) > 0:
# 更新されていればSlackに投稿
print('post slack')
post_slack('\n'.join(target_list))
except Exception as e:
print(e)
raise ezip 作成
必要なライブラリのインストールとソースコードを作成したら、zip に固めます。
$ cd ~/rssfeed
$ zip -r ~/rssfeed.zip .Lambda 関数の作成
zip が出来たら、AWS 側で Lambda 関数を作成していきます。
Lambda 関数の作成より、関数名とランタイムを指定して作成。
ロールについては、今回は何もしなくて大丈夫です。
Lambda 関数が作成されたら、「関数コード」の「コードエントリタイプ」にて ".zip ファイルをアップロード" を選択し、先ほど作成した zip ファイルをアップロードして保存します。
保存が完了すると、こんな感じでライブラリ含めてソースがアップロードされている事が確認出来ます。
この時点で、エラーが起きないかテストしてみましょう。
関数の画面上部より、「テスト」を選択します。
「テストイベントの設定」より、「イベント名」を適当に入力して「作成」します。
作成後に、「テスト」を押下すると関数が実行されます。
何事もなければ、成功となるはずです。
また、1時間以内に更新された RSS が存在していれば、この時点で Slack に通知が飛びます。
Slack にちゃんと通知される事も確認したければ、THRESHOLD 値を一時的に大きくしましょう。
CloudWatch Events の設定
Lambda 関数は出来たので、関数を実行する CloudWatch Events を作成します。
同じく関数画面の上部より、「+トリガーを追加」を押下。
以下の設定を行い、作成します。
- トリガーの種類
- CloudWatch Events/EventBridge
- ルール
- 新規ルールの作成（既存で使えるものがあればそれを使っても問題ありません）
- ルール名
- ご自由に
- ルールタイプ
- スケジュール式
- スケジュール式
- rate(1 hour)
- 1時間毎に実行されれば良いので、cron式でも問題ありません
- トリガーの有効化
- すぐにトリガーを有効にしたくない場合は、チェックを外しましょう。
ここまでで、全ての設定は完了です。
あとは放置して待つのみです。
確認
私はこの状態で一晩放置してみましたが、問題なく 1時間毎に更新された情報のみ Slack に通知が来ていました！
おわりに
Lambda + CloudWatch Events で Slack に通知を行う RSS リーダーを作成してみました。
簡単に作成できる上、RSS の URL を変更すれば好きな RSS を Slack に通知可能ですので、是非お試しください。
以上、AWS 事業本部の大前でした。
11
