Lambda + CloudWatch Events で Slack に最新情報を通知する RSSリーダーを作ってみた

LambdaとCloudWatch Eventsで好きなRSSの最新情報をSlackに通知する仕組みをサクッと作ってみました

#CloudWatch Events

#AWS Lambda

#AWS

大前諒祐

2020.04.15

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、大前です。

Lambda で RSS リーダー的な物を作ってみましたので、ブログを書いていきます。

ちなみに、Slack には RSS リーダーのアプリケーションがあるので、自己満足的な要素が大きいです。。

RSS を Lambda で取得して、AWS 側で何か処理をさせたい時の参考などにして頂ければ幸いです。

Slack に RSS フィードを追加する

構成

大まかな処理の流れは以下です。

CloudWatch Events で Lambda 関数を 1時間毎に実行
Lambda から RSS フィードを取得
1時間以内に更新があれば Slack に通知

今回は、厚生労働省が提供している RSS を取得してみる事にしてみます。

厚生労働省 - 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 インストール

Documentation - feedparser

$ pip install feedparser -t ./

pytz インストール

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 e