😸
最新AWS情報を【Bot Token + chat.postMessage】でslackに通知

2025/08/31に公開
 はじめに前回のslack通知BOTは、Incoming Webhooksを使用しましたが、今回はBot Token + chat.postMessageで実装していきたいと思います。
Incoming Webhooksを使用した記事はこちら。

https://zenn.dev/t_oishi/articles/1a9884072d4b18

 Incoming WebhooksとBot Token + chat.postMessageの違い
 Incoming Webhooks仕組み

slack が発行する URL (Webhook URL) に対して JSON を POST すると、指定されたチャンネルに投稿される。
メリット
設定がシンプル（Webhook URL に投げるだけ）
短期的・小規模な通知用途ならすぐ使える
デメリット
投稿先チャンネルは固定されがち（基本は発行時に指定）
権限管理が柔軟でない（WebHook URL が漏れると誰でも投稿できる）
slack 公式的にはレガシー扱い（今後は Bot Token 推奨）

 Bot Token + chat.postMessage API仕組み

投稿先チャンネルは API 呼び出し時に指定できる
https://docs.slack.dev/reference/methods/chat.postMessage/

https://qiita.com/MakiMatsu/items/d47f351139866025825c
メリット
チャンネルを柔軟に指定できる（固定されない）
他の Slack API も利用可能（スレッド返信、ファイルアップロード、リッチメッセージなど）
権限管理が厳密（必要な Scope のみ付与）
デメリット
初期設定が少し手間（App 作成、権限付与、トークン管理が必要）
Bot を対象チャンネルに招待しないと投稿できない

 実装方法アプリ管理画面にいきます。

https://api.slack.com/apps

OAuth & PermissionsのScopesで、chat:writeとchat:write.publicを許可しておきます。

chat:write（メッセージ送信に必須）

chat:write.public（任意：Bot が未参加チャンネルにも投稿できる）

アプリを再インストールします。

Bot User OAuth Tokenは使用しますので控えておいてください。

インストールができたら、テストとして以下をリクエストしてみましょう。
curl -X POST https://slack.com/api/chat.postMessage \
  -H "Authorization: Bearer Bot User OAuth Token" \
  -H "Content-type: application/json" \
  --data '{
    "channel": "#指定したチャンネル名",
    "text": "Hello from Slack Bot :tada:"
  }'
指定したチャンネルに通知が来れば成功です！

 LambdaSLACK_BOT_TOKENとSLACK_CHANNEL の環境変数に先ほどのトークンとチャンネル名（# 付き）は適宜変更してください。
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_BOT_TOKEN = os.environ.get("SLACK_BOT_TOKEN")
SLACK_CHANNEL = os.environ.get("SLACK_CHANNEL")

# 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_BOT_TOKEN:
        print("Slack Bot Token が設定されていません、送信スキップ")
        return

    url = "https://slack.com/api/chat.postMessage"
    headers = {
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
        "Content-Type": "application/json"
    }
    data = {
        "channel": SLACK_CHANNEL,
        "text": message
    }

    resp = requests.post(url, headers=headers, json=data)
    result = resp.json()
    if not result.get("ok"):
        print("Slack送信失敗:", result)
        resp.raise_for_status()
    else:
        print("Slack送信成功:", result)

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": "通知完了"}
GitHubで編集を提案
はじめに

Incoming WebhooksとBot Token + chat.postMessageの違い

Incoming Webhooks

Bot Token + chat.postMessage API

実装方法

Lambda

Discussion
