Go to Qiita Advent Calendar 2024 Top
LoginSignup
2
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@hirose0929inＡＲアドバンストテクノロジ株式会社（ARI）

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

Posted at

はじめに

みなさん、初めまして！エンジニア 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

構成図

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

image.png

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

一つ目のLambda

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

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 を使用して、記事タイトルの翻訳と記事内容の要約も処理しています。

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

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

image.png

Amazon SQS の設定

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

終わりに

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

2
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?