Slack の Incoming Webhook を使って AWS のアップデート情報を自動的に発信する仕組みを作ってみた

## はじめに
みなさん、初めまして！エンジニア 2 年目の hirose です。
私が所属する部署では毎朝担当者がピックアップした AWS の最新アップデート情報を Slack のチャットで発信するというタスクがありました。
手動でチャットを入力し、発信するという典型的な作業を自動化できれば、みんなが楽になる！と思ったので、生成 AI の力を借りて社内の AWS 環境で作ってみました。

## 前提条件
・AWS 環境が使用できる
・Slack に Incoming Webhook アプリをインストールしている
・Slack で Incoming Webhook のURLを発行している

## Slack で発信する内容
私の部署では Amazon Connect を使った案件に携わる社員が多いです。
そのため、今回は AWS のアップデート情報サイトの記事のタイトルに「Amazon Connect」と含まれているものがあれば、毎日決まった時間に以下のフォーマットで送信するようにします。
```
●URL: 記事のURL
●タイトル: 記事のタイトル
●タイトル翻訳：記事の日本語翻訳
●要約:生成AIが記事の内容を150文字以内で要約
```
イメージは以下のようです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3777988/d5de8c73-e584-1c0e-0714-0d9483852d99.png)

## 構成図
構成図は以下となっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3777988/13bd4ce1-2b8b-1fb3-8e0e-dbd93e95a4d2.png)

記事の要約には Amazon Bedrock を使用しました。モデルは Claude 3 Haiku です。

## 一つ目のLambda
https://aws.amazon.com/about-aws/whats-new/recent/feed/
上記の RSS フィードのページから昨日付けのアップデートの URL を取得し、Amazon SQS に送っています。

```Python:send_sqs.py
import boto3
import os
import feedparser
import logging
from datetime import datetime, timedelta, timezone

# 環境変数の取得
LOG_LEVEL = os.environ.get("LOG_LEVEL")
QUEUE_URL = os.environ["QUEUE_URL"]

# ログの設定
logger = logging.getLogger()
logger.setLevel(getattr(logging, LOG_LEVEL))

formatter = logging.Formatter('[%(levelname)s] %(message)s')
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)

for handler in logger.handlers:
    logger.removeHandler(handler)
logger.addHandler(stream_handler)

# キュー情報を設定
sqs = boto3.client("sqs", region_name="ap-northeast-1")
feed_url = "https://aws.amazon.com/about-aws/whats-new/recent/feed/"

def lambda_handler(event, context):
    # UTCタイムゾーンを定義
    UTC = timezone.utc
        
    # UTCで現在の日付と時刻を取得
    today_date = datetime.now(UTC)
    
    # 一日前の日付を計算
    one_day_before = today_date - timedelta(days=1)
    
    # 指定されたフォーマットで一日前の日付を文字列に変換
    target_date = one_day_before.strftime("%a, %d %b %Y") 

    # RSSフィードを取得
    feed = feedparser.parse(feed_url)
    
    # サービス名のリストを作成
    services = [
        'Amazon Connect'
    ]

    # ターゲット日付に該当する記事のみリンクを取得
    for entry in feed.entries:
        article_date = datetime(*entry.published_parsed[:6], tzinfo=UTC).strftime("%a, %d %b %Y")
        
        # 記事の日付とターゲット日付を比較し、タイトルに指定されたサービス名が含まれているかチェック
        if article_date == target_date:
            for service in services:
                if service in entry.title:
                    article_url = entry.link
                    # break  # サービス名が見つかったらループを抜ける

                    # sqsへ記事のURLを送信
                    response = sqs.send_message(
                        QueueUrl=QUEUE_URL,
                        MessageBody=article_url
                    )
            
                    # 一つのサービス名に対してメッセージを送信したら、他のサービス名は無視するためにループを抜ける
                    break

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

## 2つ目のLambda
Amazon SQS から URL を受け取り、フォーマットに合わせて送信しています。
Bedrock の API を使用して、記事タイトルの翻訳と記事内容の要約も処理しています。
```Python:summary_slack.py
import boto3
import json
import os
import requests
from bs4 import BeautifulSoup
import time
import logging
import jpholiday
from datetime import datetime

# 環境変数の取得
LOG_LEVEL = os.environ.get("LOG_LEVEL")
QUEUE_URL = os.environ["QUEUE_URL"]
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]

# ログの設定
logger = logging.getLogger()
logger.setLevel(getattr(logging, LOG_LEVEL))

formatter = logging.Formatter('[%(levelname)s] %(message)s')
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)

for handler in logger.handlers:
    logger.removeHandler(handler)
logger.addHandler(stream_handler)

# AWSクライアントの初期化
sqs = boto3.client("sqs", region_name="ap-northeast-1")
bedrock_runtime = boto3.client("bedrock-runtime", region_name='us-west-2')

# メイン処理
def lambda_handler(event, context):
    # 現在の日付を取得
    today = datetime.now().date()
    
    # 今日が日本の祝日かどうかをチェック
    if jpholiday.is_holiday(today):
        logger.info("今日は日本の祝日です。Slackへの送信をスキップします。")
        return {
            "statusCode": 200,
            "body": "今日は日本の祝日です。Slackへの送信をスキップしました。"
        }
    
    logger.info("今日は平日です。")

    total_processed = 0
    messages_received = False  # メッセージ受信フラグ

    while True:
        # Lambda関数の残り実行時間をチェック
        time_remaining = context.get_remaining_time_in_millis()
        if time_remaining < 10000:  # 10秒未満の場合は処理を終了
            break

        res = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            AttributeNames=["All"],
            MessageAttributeNames=["All"],
            MaxNumberOfMessages=10,
            VisibilityTimeout=15,
            WaitTimeSeconds=20  # ロングポーリングを使用
        )

        # 受信したメッセージがある場合の処理
        if "Messages" in res:
            messages_received = True  # メッセージを受信したのでフラグをTrueに設定
            for message in res["Messages"]:
                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_title = article_title.replace("- AWS", "")
                article_summary = generate_summary(article_text)
                article_translate = generate_title(article_title)
                # "summary" 文字列を取り除く
                article_summary = article_summary.replace("<summary>", "")
                article_summary = article_summary.replace("\n", "")
                time.sleep(1)
                # Slackに送信
                publish_message_to_slack(article_url, article_title, article_translate, article_summary)
                total_processed += 1
        else:
            # キューが空の場合はループを抜ける
            break
        
        # Lambda関数の実行時間を考慮して適宜スリープを挿入
        time.sleep(1)
        
        # メッセージを一つも受信していない場合、Slackに通知
    if not messages_received:
        publish_non_message_to_slack("本日の共有が必要なアップデート情報はありませんでした")

    return {
        "statusCode": 200,
        "body": f"Processed {total_processed} messages"
    }

# 記事本文のスクレイピング
def scraping_article(article_url):
    html = requests.get(article_url).content
    soup = BeautifulSoup(html, "html.parser")
    
    # 記事のタイトルを取得
    article_title = soup.find("title").get_text()
    
    # 記事の本文を含む要素を特定（例: class="article-content"）
    article_body = soup.find("div", class_="wn-content-with-nav")
    
    # 記事の本文を取得
    if article_body:
        article_text = article_body.get_text(strip=True)
    else:
        article_text = "記事の本文が見つかりませんでした。"
    
    return article_title, article_text

# 文章要約
def generate_summary(text):
    prompt_config = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 4096,
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", 
                "text": (
                    "次の記事本文を150字以内で要約してください。"
                    "article_text:{}"
                    "要約内容には以下を含めるようにしてください。"
                    "・記事の内容について（どういうことができるようになったのか）"
                    "・リージョンについて（記載が無ければ、要約には含めないこと）"
                    "・料金について（記載が無ければ、要約には含めないこと）"
                    "また、日本語以外の言語も日本語に翻訳して出力してください。"
                    "丁寧語で統一させてください。"
                    "できるだけ少ない文で要約してください。"
                    "<summary></summary>の中に要約した内容を出力してください。"
                    ).format(text)
                    },
                ],
            }
        ],
    "stop_sequences": [
            "</summary>"
            ],
    "top_p": 0,
    "temperature": 0.5
    }
    
    body = json.dumps(prompt_config)
    
    response = bedrock_runtime.invoke_model(
        modelId="anthropic.claude-3-haiku-20240307-v1:0",
        body=body,
        accept="*/*",
        contentType="application/json"
    )
    response_body = json.loads(response.get("body").read())
    summary_text = response_body.get("content")[0].get("text")

    return summary_text

# タイトル翻訳
def generate_title(text):
    prompt_config = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 4096,
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", 
                "text": (
                    "article_title:{}"
                    "を日本語に翻訳してください。"
                    "翻訳した内容のみを出力してください。"
                    ).format(text)
                    },
                ],
            }
        ],
    "top_p": 0,
    "temperature": 0.5
    }
    
    body = json.dumps(prompt_config)
    
    response = bedrock_runtime.invoke_model(
    modelId="anthropic.claude-3-haiku-20240307-v1:0",
    body=body,
    accept="*/*",
    contentType="application/json"
    )
    response_body = json.loads(response.get("body").read())
    # レスポンスボディから翻訳されたタイトルテキストを抽出
    translated_title_text = response_body.get("content")[0].get("text")

    return translated_title_text

# Slackにメッセージを送信する関数
def publish_message_to_slack(article_url, article_title, article_translate, article_summary):
    # メッセージペイロードを準備
    slack_message = {
        "text": f"●URL: {article_url}\n●タイトル: {article_title}\n●タイトル翻訳：{article_translate}\n●要約:\n{article_summary}",
    }
    
    # SlackのWebhook URLにPOSTリクエストを送信
    response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(slack_message), headers={'Content-Type': 'application/json'})
    
    # 正常なレスポンスを確認
    if response.status_code != 200:
        raise ValueError(f"Slackへのリクエストでエラーが発生しました {response.status_code}, レスポンスは:\n{response.text}")
    
    return response

# Slackにメッセージを送信する関数(アップデート情報がない場合)
def publish_non_message_to_slack(text):
    # メッセージペイロードを準備
    slack_message = {
        "text": f"{text}"
    }
    
    # SlackのWebhook URLにPOSTリクエストを送信
    response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(slack_message), headers={'Content-Type': 'application/json'})
    
    # 正常なレスポンスを確認
    if response.status_code != 200:
        raise ValueError(f"Slackへのリクエストでエラーが発生しました {response.status_code}, レスポンスは:\n{response.text}")
    
    return response
```

## Amazon EventBridge の設定
上記の Lambda 二つを時間差で実行するように設定しました。
一つ目の Lambda は火曜日から土曜日の 9 時に実行します。AWS のアップデートが土日には無いためです。
これで昨日付けのアップデートの URL を取得し、Amazon SQS にて保持できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3777988/329edd3a-570d-56db-d3a1-b4375a6307f8.png)

二つ目の Lambda は月曜日から金曜日の 9 時半に実行します。
これで土日にチャットが来ることを避けられます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3777988/d5026825-f066-a0b1-7fdc-ce96e016f44c.png)

# Amazon SQS の設定
メッセージの保持期間を 3 日に設定しました。
土曜日に取得した URL を月曜日まで保持するためです。

## 終わりに
今回は初めて自動化の仕組みを作り、部署内の作業時間の削減ができたので良かったです。
実は今回コードの生成も ChatGPT-4 の力をお借りました。「~~したい」や「~~教えて」というだけでコードを生成してくれるのは、便利な世の中になりましたね。
この仕組みは AWS アップデート情報以外にも応用が利くと思いますので、誰かの参考になれば幸いです。