DevelopersIOに投稿された新規記事を要約してメール通知してみた

こんにちは。クラメソへjoinして1週間、膨大なインプット情報に殴られている戸川です。
先日会社からAWSの個人用検証環境をいただき喜び勇んで遊んできたので、内容をこちらでアウトプットしたいと思います。

作ったもの

タイトルの通りです。
LambdaでDevelopersIOの記事をスクレイピングし、Bedrockで要約した結果をSNSで通知しました。

構成図

通知結果

作った経緯

・DevelopersIOに投稿される記事を（要約レベルでいいので）全部把握しておきたかった

私は大体1日に数回DevelopersIOを巡回していますが、それでも「えっ！？　知らん間になんかおもろい記事増えとるやんけ！」と思うことがこれまで何度かありました。
そういった悲しい出来事の発生を抑制するには、見落としに気づくための何らかの仕組みが必要でした。
そんな中、今回は「取り敢えず前日分の記事全部要約して目を通そう」という雑な結論に至りました。

・Bedrockを使ってみたかった

Amazon Bedrockは2023年9月末にGAされた、AWSの生成AIサービスです。
以前からOpenAIのChatGPTなどは使っていてBedrockにも惹かれていましたが、リリース当時は転職活動などで忙しく遊ぶ余裕がなかったため、ずっと隙を見計らっていました。

・何でもいいからサーバレスの気分だった

人にはそういう瞬間があると思います。

ソースコード

1つ目のLambda関数です。
今回はDevelopersIOのトップページから前日日付分の記事URLを取得しSQSへ送信しています。

import boto3
import os
import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta, timezone


# キュー情報を設定
queue_url = os.environ["QUEUE_URL"]
sqs = boto3.client("sqs", region_name="ap-northeast-1")
main_url = "https://dev.classmethod.jp"


def lambda_handler(event, context):
    # 昨日の日付をターゲットに設定
    t_delta = timedelta(hours=9)
    JST = timezone(t_delta, 'JST')
    yesterday_date = datetime.now(JST) - timedelta(1)
    target_date = yesterday_date.strftime("%Y.%m.%d")

    # トップページから記事リンク一覧を取得
    html = requests.get(main_url).content
    soup = BeautifulSoup(html, "html.parser")
    article_links = soup.find_all("p", class_="date")

    # ターゲット日付に該当する記事のみリンクを取得
    for article_link in article_links:
        article_date = article_link.text.strip()
        if article_date == target_date:
            link = article_link.find_parent("div", class_="post-container").find("a", class_="link")
            if link:
                article_url = main_url + link["href"]

                # sqsへ記事のURLを送信
                response = sqs.send_message(
                    QueueUrl=queue_url,
                    MessageBody=article_url
                )

    return {
        'statusCode': 200,
        'body': "OK"
    }

2つ目のLambda関数です。
SQSキューから記事URLを取得し本文をスクレイピングした後、Bedrock（今回はClaude Instantモデルを使用）に要約タスクをリクエストしています。
要約結果はSNS経由でメール通知します。

import boto3
import json
import os
import requests
from bs4 import BeautifulSoup


# キュー情報を設定
queue_url = os.environ["QUEUE_URL"]
topic_arn = os.environ["TOPIC_ARN"]
sqs = boto3.client("sqs", region_name="ap-northeast-1")
sns = boto3.client("sns", region_name="ap-northeast-1")
bedrock_runtime = boto3.client("bedrock-runtime",region_name="ap-northeast-1")


# メイン処理
def lambda_handler(event, context):
    res = sqs.receive_message(
        QueueUrl=queue_url,
        AttributeNames=["All"],
        MessageAttributeNames=["All"],
        MaxNumberOfMessages=1,
        VisibilityTimeout=30,
        WaitTimeSeconds=0
    )

    if "Messages" in res:
        message = res["Messages"][0]
        article_url = message["Body"]
        # メッサージをキューから削除
        receipt_handle = message["ReceiptHandle"]
        sqs.delete_message(
            QueueUrl=queue_url,
            ReceiptHandle=receipt_handle
        )
        # スクレイピング処理に記事URLを連携
        article_title, article_text = scraping_article(article_url)
        article_summary = generate_summary(article_text)
        response = publish_message(article_url, article_title, article_summary)

        return {
            "statusCode": 200,
            "body": "OK"
        }


# 記事本文のスクレイピング
def scraping_article(article_url):
    html = requests.get(article_url).content
    soup = BeautifulSoup(html, "html.parser")
    # 記事のタイトルを取得
    article_title = soup.find("title").get_text()
    # 不要な文章をクラスで指定
    exclude_classes = [
        "blocks", "copyright", "events", "posts", "post-content",
        "related", "share-navigation", "sub-content"
    ]
    exclude_divs = soup.find_all("div", class_=exclude_classes)
    if exclude_divs:
        for exclude_div in exclude_divs:
            exclude_div.extract()
    article_text = soup.body.get_text()

    return article_title, article_text


# 文章要約
def generate_summary(text):
    input_text = (
        "\n\nHuman: You are an IT engineer."
        "Summarize the following article_text and write up to 5 sentences in the form of a response example."
        "In addition, please translate language other than Japanese to Japanese and output."
        "\n\narticle_text: {}\n\nresponse example:"
        "- first sentence\n- second sentence\n- third sentence\n- forth sentense\n- fifth sentence\n\nAssistant:"
    ).format(text)

    request_body = json.dumps(
        {
            "prompt": input_text,
            "max_tokens_to_sample": 300,
            "temperature": 0.5,
            "top_k": 250,
            "top_p": 1,
            "anthropic_version": "bedrock-2023-05-31"
        }
    )
    response = bedrock_runtime.invoke_model(
        modelId="anthropic.claude-instant-v1",
        body=request_body,
        accept="*/*",
        contentType="application/json"
    )
    response_body = json.loads(response.get("body").read())

    return response_body


# 要約結果をEメール送信
def publish_message(article_url, article_title, article_summary):
    message = (
        "article_url: {}\narticle_title: {}\narticle_summary: {}"
    ).format(article_url, article_title, article_summary["completion"])
    response = sns.publish(
        TopicArn=topic_arn,
        Message=message,
        Subject="dev-io-summary"
    )

    return response

今回Lambdaに許可したポリシーは以下になります。

"bedrock:InvokeModel"
"sns:Publish"
"sqs:SendMessage"
"sqs:ReceiveMessage"
"sqs:DeleteMessage"
"sqs:GetQueueAttributes"
"logs:CreateLogGroup"
"logs:CreateLogStream"
"logs:PutLogEvents"

注意点

• LLMによる文章要約はその仕組み上、発信者の意図に沿った要約が生成されるとは限らないという点には注意が必要です。あくまでも参考程度の情報として認識することが重要でしょう。
• Webスクレイピングで情報収集を行う際はサイトの利用規約やrobots.txtなどからスクレイピングが禁止されていないことを確認した上で常識的な利用に留めましょう。

感想

今回初めてBedrockを利用しましたが、思っていたより簡単に使うことができました。東京リージョンでも更に多くのモデルが使えるようになると嬉しいです。
また今回の構成はDevelopersIO以外からの情報収集にも転用可能だと思うので、隙を見て追加開発していければと思います。

参考

Amazon BedrockをBoto3から使ってみた