[AWS] Webサイトの定期更新チェック [AWS Lambda]

目的

みなさん、WEBサイトの更新をできる限り早くキャッチアップしたいときとかありますよね？私はあります。というわけで、今回は、AWS Lambdaを使って自動的にチェックしてメール通知を送るために作ったものを紹介します。
Pythonを利用しますが、Lambdaの環境に含まれる標準パッケージのみで、追加パッケージを利用せずに作ってます。

Disclaimer

基本的に動作保証はしませんが、特に今回作成したものは、HTMLのBodyのcontent部分だけを抜き出して比較するものですので、動的に生成されるようなWEBサイトでは正しく差分を取れない可能性があります。ご注意ください。

構成

下図の構成を取ります。

構成図

(1) 頻度、取得するURLはEventBridgeのルール側で設定し、Lambda実行時にURLを渡す。(複数URLに対して、異なる頻度で実行したかったため)
(2) Lambdaが指定されたURLのコンテンツを取得。
(3) Lambdaが(2)のHTMLのBodyの中の各要素のcontentのみを抽出し、ハッシュ化。ハッシュ値をKeyとしたオブジェクトがS3バケットに存在するか確認、なければ作成。
(4) (3)でオブジェクトが存在していなかった場合、SNSにURLを含むメッセージを通知。

EventBridge

EventBridgeはTargetにLambdaを設定、1時間ごとに起動し、Advanced SettingsでURLを設定します。

EventBridge Ruleの定期実行設定

EventBridge RuleのTarget(Lambda)設定と、URLを渡す設定

Lambda

Pythonで記述しています。追加のPackageは不要となるよう、HTTPのGetリクエストはrullib、ハッシュ生成はhashlib、HTMLの抽出はHTMLParserを利用しています。

import urllib.request
import hashlib
import boto3
import os
from io import StringIO
from html.parser import HTMLParser

s3_client = boto3.client('s3')
sns_client = boto3.client('sns')

class MyHTMLParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.text = StringIO()
        self.is_in_body = False

    def handle_starttag(self, tag, attrs):
        if tag.lower() == "body":
            self.is_in_body = True

    def handle_endtag(self, tag):
        if tag.lower() == "body":
            self.is_in_body = False

    def handle_data(self, d):
        if self.is_in_body == True:
            self.text.write(d)

    def get_data(self):
        text = self.text.getvalue()
        self.text = StringIO()
        return text

def fetch_url_content(url):
    """
    Fetch content from the given URL using urllib.
    """
    req = urllib.request.Request(
        url=url, 
        data=None
    )

    with urllib.request.urlopen(req) as response:
        charset=response.info().get_content_charset()
        content=response.read().decode(charset)
        return content

def create_hash(content):
    sha256_hash = hashlib.sha256()
    sha256_hash.update(content.encode('utf-8'))
    return sha256_hash.hexdigest()

def save_to_s3(bucket_name, key, content):

    s3_client.put_object(Bucket=bucket_name, Key=key, Body=content)

def if_key_exist(bucket_name, key):

    try:
        s3_client.head_object(
            Bucket=bucket_name,
            Key=key
        )
        return True
    except s3_client.exceptions.ClientError as e:
        print(e.response["Error"]["Code"])
        if e.response["Error"]["Code"] == "404":
            return False
        else:
            raise e
    except Exception as e:
        raise e

def send_message_to_sns(topic_arn, message):
    sns_client.publish(
        TopicArn=topic_arn,
        Message=message
    )

def lambda_handler(event, context):
    bucket_name = os.environ.get('BUCKET_NAME')
    topic = os.environ.get('TOPIC')
    url = event["url"]

    if not url or not bucket_name or not topic:
        return {
            'statusCode': 400,
            'body': {
                'message': 'Missing required parameters: url, bucket_name or topic.'
            }
        }

    try:
        print(f"fetch data from url {url}")
        content = fetch_url_content(url)
        s = MyHTMLParser()
        s.feed(content)
        body = s.get_data()
        print(body)

        # Create a hash from the content
        content_hash = create_hash(body)

        if if_key_exist(bucket_name, content_hash):
            message = f"No update on {url}. Skipped..."
            return {
                'statusCode': 200,
                'body': {
                    'message': f'No update.',
                    'hash': content_hash
                }
            }

        else: 
            message = f"There is an update on {url}. New hash {content_hash} will be saved on S3."
            # Save the hash to S3
            save_to_s3(bucket_name, content_hash, body)

            send_message_to_sns(topic, f"there is an update on {url}.\n You must check what the update is!!!\n\n{url}")

            return {
                'statusCode': 200,
                'body': {
                    'message': f'There is an update on {url} and a notification was sent to {topic}',
                    'hash': content_hash
                }
            }
    except Exception as e:
        raise e

SNS Topic

SNSトピックは標準、MAILで設定。

SNSトピック設定

結果

WEBサイトに更新があったときは、下記のような通知を受け取ることができるようになりました。

メール例

メール本文にURLも入れているので、メールを受け取ったらすぐに中身が確認できるようになりました。

今後

DiffもちゃんととってHTMLメールとかで差分を送ったり整形したいところですが、ひとまず通知を受け取るところはできました。