📚
最新AWS情報を【Incoming Webhooks】でslackに通知

2025/08/31に公開
 はじめに前回、AWS公式RSSからLINEに通知する実装をしましたが、今回はIncoming Webhooksを使用してslackへも通知をしていこうと思います。

pythonコードやAWSの構成ほぼ同じです。
https://zenn.dev/t_oishi/articles/10007f6fa2397b
Bot Token + chat.postMessageを使用した記事はこちら。

https://zenn.dev/t_oishi/articles/b1d6e7c36797c9

 実装方法
 slackの設定事前にslackにログインできるようにしておいてください。
まず、BOTを通知したいチャンネルを作成しておきます。

今回は# aws-botというチャンネルを作成しておきました。

その後、以下のURLでアプリを作成していきます。

https://api.slack.com/apps




アプリの名前とワークスペースの名前を選択します。







以下の設定で全然作成したチャンネルが出てこず、１時間くらい調べていました。

→Xを見ていたらslackの不具合があったようでした。チャンネルが出てこないときはslack公式で不具合情報が出てないか見てみるのもいいかもしれません。



Webhook URLを控えておきましょう。


 Lambda以下はLINE通知も含まれています。
環境変数に先ほどのWebhook URLを設定して下さい。
import os
import asyncio
import aiohttp
import feedparser
from datetime import datetime, timezone
import requests
import boto3
from botocore.exceptions import BotoCoreError, ClientError

# 環境変数
LINE_API_URL = "https://api.line.me/v2/bot/message/push"
LINE_TOKEN = os.environ["LINE_TOKEN"]
LINE_USER_ID = os.environ["LINE_USER_ID"]
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")

# Amazon Translateクライアント
translate = boto3.client("translate", region_name="ap-northeast-1")

# 送った日づけ
today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")

# RSSフィード一覧
RSS_FEEDS = {
    "What's New with AWS": {"url": "https://aws.amazon.com/jp/about-aws/whats-new/recent/feed/", "translate": False},
    "AWS Blog": {"url": "https://aws.amazon.com/blogs/aws/feed/", "translate": True},
    "AWS Architecture Blog": {"url": "https://aws.amazon.com/jp/blogs/architecture/feed/", "translate": True},
    "AWS Partner Network (APN) Blog": {"url": "https://aws.amazon.com/jp/blogs/apn/feed/", "translate": True},
    "AWS Big Data Blog": {"url": "https://aws.amazon.com/jp/blogs/big-data/feed/", "translate": True},
    "AWS Compute Blog": {"url": "https://aws.amazon.com/jp/blogs/compute/feed/", "translate": True},
    "AWS Database Blog": {"url": "https://aws.amazon.com/jp/blogs/database/feed/", "translate": True},
    "AWS Desktop and Application Streaming Blog": {"url": "https://aws.amazon.com/jp/blogs/desktop-and-application-streaming/feed/", "translate": True},
    "AWS Developer Blog": {"url": "https://aws.amazon.com/jp/blogs/developer/feed/", "translate": True},
    "AWS DevOps Blog": {"url": "https://aws.amazon.com/jp/blogs/devops/feed/", "translate": True},
    "AWS Cloud Enterprise Strategy Blog": {"url": "https://aws.amazon.com/jp/blogs/enterprise-strategy/feed/", "translate": True},
    "Amazon Game Tech Blog": {"url": "https://aws.amazon.com/jp/blogs/gametech/feed/", "translate": True},
    "The Internet of Things on AWS – Official Blog": {"url": "https://aws.amazon.com/jp/blogs/iot/feed/", "translate": True},
    "AWS Machine Learning Blog": {"url": "https://aws.amazon.com/jp/blogs/machine-learning/feed/", "translate": True},
    "AWS Management Tools Blog": {"url": "https://aws.amazon.com/jp/blogs/mt/feed/", "translate": True},
    "AWS Media Blog": {"url": "https://aws.amazon.com/jp/blogs/media/feed/", "translate": True},
    "AWS Messaging & Targeting Blog": {"url": "https://aws.amazon.com/jp/blogs/messaging-and-targeting/feed/", "translate": True},
    "AWS Mobile Blog": {"url": "https://aws.amazon.com/jp/blogs/mobile/feed/", "translate": True},
    "Networking & Content Delivery": {"url": "https://aws.amazon.com/jp/blogs/networking-and-content-delivery/feed/", "translate": True},
    "AWS Open Source Blog": {"url": "https://aws.amazon.com/jp/blogs/opensource/feed/", "translate": True},
    "AWS Government, Education, & Nonprofits Blog": {"url": "https://aws.amazon.com/jp/blogs/publicsector/feed/", "translate": True},
    "AWS for SAP": {"url": "https://aws.amazon.com/jp/blogs/awsforsap/feed/", "translate": True},
    "AWS Security Blog": {"url": "https://aws.amazon.com/jp/blogs/security/feed/", "translate": True},
    "AWS Startups Blog": {"url": "https://aws.amazon.com/jp/blogs/startups/feed/", "translate": True},
    "AWS Japan Blog": {"url": "https://aws.amazon.com/jp/blogs/news/feed/", "translate": False},
    "AWS Marketplace": {"url": "https://aws.amazon.com/blogs/awsmarketplace/feed/", "translate": True},
    "AWS Fundamentals Blog": {"url": "https://aws.amazon.com/blogs/training-and-certification/feed/", "translate": True},
    "AWS IoT Blog": {"url": "https://aws.amazon.com/blogs/iot/feed/", "translate": True},
    "週刊AWS": {"url": "https://aws.amazon.com/jp/blogs/news/tag/%E9%80%B1%E5%88%8Aaws/feed/", "translate": False},
}

import re

def translate_title_if_needed(title, should_translate):
    """
    フィード設定に基づいてタイトルを翻訳する
    """
    if not should_translate:
        print(f"翻訳設定OFF、スキップ: {title}")
        return title
        
    try:
        print(f"翻訳設定ON、翻訳実行: {title}")
        result = translate.translate_text(
            Text=title,
            SourceLanguageCode="en",
            TargetLanguageCode="ja"
        )
        translated_title = result.get("TranslatedText", title)
        print(f"翻訳完了: {title} -> {translated_title}")
        return translated_title
        
    except (BotoCoreError, ClientError) as e:
        print(f"Translate error: {e}")
        return title

async def fetch_feed(session, name, feed_config):
    url = feed_config["url"]
    should_translate = feed_config["translate"]
    
    async with session.get(url) as resp:
        text = await resp.text()
        feed = feedparser.parse(text)
        today = datetime.now(timezone.utc).date()
        entries = []

        for entry in feed.entries[:10]:
            pub_date = None
            if hasattr(entry, "published_parsed") and entry.published_parsed:
                pub_date = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc).date()
            elif hasattr(entry, "updated_parsed") and entry.updated_parsed:
                pub_date = datetime(*entry.updated_parsed[:6], tzinfo=timezone.utc).date()
            
            if pub_date == today:
                # フィード設定に基づいて翻訳
                title = translate_title_if_needed(entry.title, should_translate)
                entries.append(f"・{title}\n{entry.link}")
        return name, entries

async def fetch_all_feeds():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_feed(session, name, config) for name, config in RSS_FEEDS.items()]
        return await asyncio.gather(*tasks)

def send_line_message(message):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LINE_TOKEN}"
    }
    data = {
        "to": LINE_USER_ID,
        "messages": [{"type": "text", "text": message}]
    }
    resp = requests.post(LINE_API_URL, headers=headers, json=data)
    resp.raise_for_status()
    print("LINE送信成功:", resp.status_code)

def send_slack_message(message):
    if not SLACK_WEBHOOK_URL:
        print("Slack Webhook URL が設定されていません、送信スキップ")
        return
    headers = {"Content-Type": "application/json"}
    data = {"text": message}
    resp = requests.post(SLACK_WEBHOOK_URL, headers=headers, json=data)
    resp.raise_for_status()
    print("Slack送信成功:", resp.status_code)

def lambda_handler(event, context):
    feeds = asyncio.run(fetch_all_feeds())

    for name, entries in feeds:
        if entries:
            text = f"【{name}】 ({today_str})\n" + "\n".join(entries)
            send_line_message(text)
            send_slack_message(text)

    return {"statusCode": 200, "body": "通知完了"}

LINEに通知したくない場合、下記を削除してください。
def send_line_message(message):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LINE_TOKEN}"
    }
    data = {
        "to": LINE_USER_ID,
        "messages": [{"type": "text", "text": message}]
    }
    resp = requests.post(LINE_API_URL, headers=headers, json=data)
    resp.raise_for_status()
    print("LINE送信成功:", resp.status_code)
これで完了です！
無料プランは 直近90日分のメッセージしか閲覧できないので、90日を超えると検索・表示できなくなります。
GitHubで編集を提案
はじめに

実装方法

slackの設定

Lambda

Discussion
