はじめに
- AWS EventBridgeを利用して、特定のサイトとキーワードをクロールし、SNSにクロールされたデータを送信する方法を紹介します。
- API Gatewayを接続することで、キーワードを直接修正でき、データをDynamoDBに保存すれば、クロールされたデータをウェブサイトなどでも閲覧することができます。
- 本記事では、最も簡単な方法として、トリガーにて毎朝9時にメールで送信します。
結果物
図式化
事前準備
- SNSの受信には、メールアカウントが必要です。
（または、SlackやTeamsのチャンネルメールアドレスを利用することも可能です） - Lambdaで使用するPythonライブラリをインストールする必要があるため、pipのインストールが必要です。
SNS設定
- タイプ：スタンダード
- 名前：任意
- その他：デフォルト
Lambdaからクローリングされたデータを送信するためのAmazon SNSトピックを作成します。
作成されたSNSトピックにサブスクリプションを登録します。
右上の「サブスクリプションの作成」をクリックします。
- ※トピックARN：作成した Amazon SNSのトピック
- プロトコル：Eメールアドレス
- エンドポイント：通知先のメールアドレス
※ トピックARNはLambdaコードで使用するため、メモ帳などにコピーしておきます。
作成したサブスクリプションのステータスが「保留中確認」になったら、エンドポイントのメールアドレスに認証メールが届きます。
メール内の「Confirm subscription」をクリックすると、ステータスが「確認済み」になったことを確認できます。
Pythonライブラリー作成（Lambdaレイヤー用）
Lambdaは内蔵ライブラリが限られているため、外部ライブラリを使用するにはレイヤーの追加が必要です。レイヤーでライブラリを追加するためには、特定のディレクトリ構造に従って圧縮する必要があります。
参考： レイヤーコンテンツのパッケージング - AWS Lambda (amazon.com)
pip install requests beautifulsoup4 -t .
zip -r9 myLayer.zip .
上記のコマンドを使用して、ライブラリを圧縮ファイルにまとめます。
環境構築
Lambda関数作成
- 関数名：任意
- ランタイム：Python 3.12
- アーキテクチャ：x86_64
- その他：デフォルト
先ほど作成したSNSトピックにクローリングデータを送信するためのLambda関数を作成します。
本記事ではPythonを利用してコードを作成します。
Lambdaレイヤー作成
Lambdaダッシュボードの「その他のリソース」ー「レイヤー」から、右上の「レイヤーの作成」をクリックします。
- 名前：任意
- .zipファイルをアップロード：myLayer.zip（先ほど作成したPythonライブラリ）
- 互換性のあるアーキテクチャ：x86_64
- 互換性のあるアーキテクチャ：Python 3.12
上記の内容でLambdaレイヤーを作成します。
Lambda関数設定
Lambda設定変更
Lambda関数へ切り替えて作業を進めます。
「設定」→「一般設定」で「タイムアウト」を「30秒」に変更し、保存します。
Lambdaレイヤー追加
「コード」タブー⇒「レイヤー」⇒「レイヤーの追加」をクリックします。
- レイヤーソース：カスタムレイヤー
- カスタムレイヤー：追加したレイヤー
- バージョン：最新のもの
上記の内容で保存します。
Lambda送信先追加
「送信先を追加」をクリックし、データの送信先になるAmazon SNSを追加します。
- ソース：非同期呼び出し
- 条件：失敗時
- 送信先タイプ：SNSトピック
- 送信先：作成したAmazon SNSのARN
上記の内容で保存します。
Lambdaトリガー追加
「トリガーを追加」をクリックし、Lambdaコードを実行するためのEventBridgeを追加します。
- ルール：新規ルールを作成
- ルールタイプ：スケジュール式
- ※ スケジュール式：cron(0 0 ? * MON-FRI *)
※ 日本時間毎朝9時に実行されます。(JST⇒UTC)
参考：cron 式のリファレンス - Amazon EventBridge
上記の内容で保存します。
Lambdaコードを作成
Lambdaのコードソースに以下のコードを貼り付けます。
※右上の「COPY」を押して簡単にコピーできます。
import json
import requests
from bs4 import BeautifulSoup
import boto3
from datetime import datetime
def lambda_handler(event, context):
sns_client = boto3.client('sns')
sns_topic_arn = 'arn:aws:sns:ap-northeast-1:' #作成したSNSトピックARNに変更
# デフォルトのURLとキーワードリスト
default_urls = ['example.com'] #クローラー対象のWEBサイト
default_keywords = ['AI'] #クローラー対象のWEBキーワード
# Eventから受け取ったURLとキーワードを追加または更新
urls = event.get('urls', default_urls)
keywords = event.get('keywords', default_keywords)
# URLとキーワードを重複なくマージ
urls = list(set(default_urls + urls))
keywords = list(set(default_keywords + keywords))
all_results = []
for url in urls:
response = requests.get(url)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
for keyword in keywords:
articles = soup.find_all('a', href=True)
for article in articles:
title = article.get_text().strip()
link = article['href']
# タイトルがキーワードと完全に一致する場合は除外
if title.lower() == keyword.lower():
continue
if keyword.lower() in title.lower():
full_link = link if link.startswith('http') else f"https://www.itmedia.co.jp{link}"
all_results.append({'title': title, 'link': full_link})
# 今日の日付を取得
today = datetime.today().strftime('%Y-%m-%d')
sns_subject = f"{today}の最新ニュース"
# SNSメッセージのフォーマット
sns_message = "\n".join([f"タイトル: {result['title']}\nリンク: {result['link']}" for result in all_results])
# SNSを通じてフォーマットされたメッセージを送信
sns_client.publish(
TopicArn=sns_topic_arn,
Message=sns_message,
Subject=sns_subject
)
return {
'statusCode': 200,
'body': 'クローリングの結果がSNSに送信されました。'
}
コード内部にある一部の箇所を変更する必要があります。
sns_topic_arn= 'arn:aws:sns:ap-northeast-1:xxx'
作成したSNSトピックARNに変更します。
default_urls = ['example.com']
サイトを追加する場合は、 ['example.com', 'example2.com'] のように記述します。
default_keywords = ['AI']
キーワードを追加する場合は、 ['AI', 'Tech'] のように記述します。
コードを修正したら、「Deploy」後に「Test」を押します。
テストイベントのテンプレートはデフォルトのSNS値で問題ありません。
設定したメールで内容が届くことを確認できます。
まとめ
Lambdaを使って簡単にコードを作成し、クロールしたデータをメールに転送してみました。
サイトによってはクロールがうまくいかない場合もありますので、コードは適宜修正してください。
また、許可されていない商用利用は禁止されている場合もあるので、ご注意ください。
最後までご覧いただきありがとうございました！
