0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

AWS更新情報を自動でSlack通知するシステムを作ってみた 〜GitHub ActionsとOpenAI APIで実現する週次通知Bot〜

Posted at

はじめに

AWSの新機能リリースやアップデート情報、追いきれていませんか?

特にAmazon Connectのような特定サービスに注目している場合、関連する更新だけを効率的にキャッチアップしたいですよね。そこで、AWS公式の更新情報を自動収集し、AIで日本語要約してSlackに通知するシステムを作りました。

本記事では、GitHub ActionsとOpenAI APIを活用した自動通知システムの実装方法を詳しく解説します。

🎯 作ったもの

システム概要

  • AWS公式RSSフィードから最新情報を自動収集
  • OpenAI APIで英語記事を日本語3行要約に変換
  • 毎週月曜日9時Slackへ自動通知
  • 重複防止機能で同じ記事の再通知を防止

実際の通知例

📢 Amazon Connect 週次更新(1件/一次情報リンク)
📅 2025年1月20日 | AI要約: ON
----------------------------------------
1. Amazon Connect launches new agent workspace
   ・エージェント向け新ワークスペース機能を提供開始
   ・複数チャネルの統合管理とリアルタイム分析が可能に
   ・生産性向上と顧客満足度改善への寄与が期待される
----------------------------------------

🏗️ システムアーキテクチャ

📦 必要な技術スタック

  • Python 3.10: メイン処理言語
  • GitHub Actions: 定期実行の自動化
  • OpenAI API: GPT-4による要約生成
  • Slack Incoming Webhooks: メッセージ投稿

🚀 実装方法

1. プロジェクト構成

aws-notifier/
├── .github/
│   └── workflows/
│       └── notify.yml        # GitHub Actions定義
├── scripts/
│   ├── notify.py             # メイン処理
│   ├── ai_providers.py       # AI要約処理
│   └── config.py             # 設定管理
├── data/
│   └── seen_links.json       # 通知済みリンク履歴
├── requirements.txt          # 依存関係
└── README.md

2. メイン処理の実装(notify.py)

import feedparser
import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict
import os

class AWSUpdateNotifier:
    def __init__(self):
        self.slack_webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
        self.openai_api_key = os.environ.get('OPENAI_API_KEY')
        self.seen_links_file = 'data/seen_links.json'
        self.time_window_days = 7
        
    def fetch_aws_feeds(self) -> List[Dict]:
        """AWS RSSフィードから更新情報を取得"""
        feeds = [
            'https://aws.amazon.com/new/feed/',
            'https://aws.amazon.com/about-aws/whats-new/recent/feed/'
        ]
        
        entries = []
        cutoff_date = datetime.now() - timedelta(days=self.time_window_days)
        
        for feed_url in feeds:
            feed = feedparser.parse(feed_url)
            for entry in feed.entries:
                # 日付のパースと絞り込み
                published = datetime.strptime(
                    entry.published, '%a, %d %b %Y %H:%M:%S %Z'
                )
                if published > cutoff_date:
                    entries.append({
                        'title': entry.title,
                        'link': entry.link,
                        'summary': entry.summary,
                        'published': published
                    })
                    
        return entries
    
    def filter_connect_updates(self, entries: List[Dict]) -> List[Dict]:
        """Amazon Connect関連の更新のみをフィルタリング"""
        keywords = [
            'amazon connect', 'contact center', 'contact lens',
            'voice id', 'customer profiles', 'contact flow'
        ]
        
        filtered = []
        for entry in entries:
            text = f"{entry['title']} {entry['summary']}".lower()
            if any(keyword in text for keyword in keywords):
                filtered.append(entry)
                
        return filtered
    
    def remove_duplicates(self, entries: List[Dict]) -> List[Dict]:
        """通知済みのリンクを除外"""
        seen_links = self.load_seen_links()
        new_entries = [
            entry for entry in entries 
            if entry['link'] not in seen_links
        ]
        return new_entries
    
    def generate_summary(self, entry: Dict) -> str:
        """OpenAI APIで日本語要約を生成"""
        if not self.openai_api_key:
            # APIキーがない場合は簡易要約
            return self.fallback_summary(entry)
            
        from openai import OpenAI
        client = OpenAI(api_key=self.openai_api_key)
        
        prompt = f"""
        以下のAWS更新情報を日本語で3行に要約してください。
        各行は80文字以内で、箇条書き形式(・で開始)としてください。
        
        タイトル: {entry['title']}
        内容: {entry['summary'][:500]}
        """
        
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=300,
                temperature=0.3
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"OpenAI API error: {e}")
            return self.fallback_summary(entry)
    
    def send_to_slack(self, entries: List[Dict]):
        """Slackに通知を送信"""
        if not entries:
            print("No new updates to notify")
            return
            
        blocks = self.create_slack_blocks(entries)
        
        response = requests.post(
            self.slack_webhook_url,
            json={"blocks": blocks}
        )
        
        if response.status_code == 200:
            self.save_seen_links(entries)
            print(f"Successfully notified {len(entries)} updates")
        else:
            print(f"Slack notification failed: {response.status_code}")
    
    def create_slack_blocks(self, entries: List[Dict]) -> List[Dict]:
        """Slack用のブロックメッセージを作成"""
        blocks = []
        
        # ヘッダー
        date_str = datetime.now().strftime('%Y年%m月%d日')
        ai_status = "ON" if self.openai_api_key else "OFF"
        
        blocks.append({
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"📢 Amazon Connect 週次更新({len(entries)}件)"
            }
        })
        
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"📅 {date_str} | AI要約: {ai_status}"
            }
        })
        
        # 各更新情報
        for i, entry in enumerate(entries, 1):
            summary = self.generate_summary(entry)
            blocks.append({
                "type": "divider"
            })
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*{i}. <{entry['link']}|{entry['title']}>*\n{summary}"
                }
            })
            
        return blocks
    
    def run(self):
        """メイン処理の実行"""
        print("Starting AWS update notification...")
        
        # 1. フィード取得
        entries = self.fetch_aws_feeds()
        print(f"Fetched {len(entries)} entries")
        
        # 2. フィルタリング
        entries = self.filter_connect_updates(entries)
        print(f"Filtered to {len(entries)} Connect-related entries")
        
        # 3. 重複除去
        entries = self.remove_duplicates(entries)
        print(f"New entries: {len(entries)}")
        
        # 4. Slack通知
        self.send_to_slack(entries)

if __name__ == "__main__":
    notifier = AWSUpdateNotifier()
    notifier.run()

3. GitHub Actions の設定(.github/workflows/notify.yml)

name: AWS Update Notifier

on:
  schedule:
    # 毎週月曜日 9:00 JST (0:00 UTC)
    - cron: '0 0 * * 1'
  workflow_dispatch:  # 手動実行も可能

jobs:
  notify:
    runs-on: ubuntu-latest
    
    steps:
    - name: Checkout repository
      uses: actions/checkout@v3
      with:
        token: ${{ secrets.GITHUB_TOKEN }}
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
    
    - name: Run notification script
      env:
        SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
        OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      run: |
        python scripts/notify.py
    
    - name: Commit seen links
      run: |
        git config user.name "github-actions[bot]"
        git config user.email "github-actions[bot]@users.noreply.github.com"
        git add data/seen_links.json
        if git diff --staged --quiet; then
          echo "No changes to commit"
        else
          git commit -m "Update seen links [skip ci]"
          git push
        fi

4. 環境変数の設定

GitHub リポジトリの Settings > Secrets and variables > Actions で以下を設定:

  • SLACK_WEBHOOK_URL: Slack Incoming Webhook URL
  • OPENAI_API_KEY: OpenAI API キー(オプション)

🔧 カスタマイズのポイント

監視対象サービスの変更

config.pyでキーワードを変更することで、他のAWSサービスも監視可能:

SERVICE_CONFIGS = {
    'lambda': {
        'keywords': ['aws lambda', 'serverless', 'function'],
        'emoji': ''
    },
    'bedrock': {
        'keywords': ['amazon bedrock', 'generative ai', 'foundation model'],
        'emoji': '🤖'
    },
    's3': {
        'keywords': ['amazon s3', 'simple storage', 'object storage'],
        'emoji': '🗄️'
    }
}

通知頻度の変更

GitHub Actionsのcron式を変更:

# 毎日 10:00 JST
- cron: '0 1 * * *'

# 平日のみ 9:00 JST
- cron: '0 0 * * 1-5'

# 月水金 15:00 JST
- cron: '0 6 * * 1,3,5'

AI要約プロバイダーの切り替え

OpenAI以外にもClaude APIなどに対応:

class ClaudeProvider:
    def generate_summary(self, text: str) -> str:
        # Claude API実装
        pass

class AzureOpenAIProvider:
    def generate_summary(self, text: str) -> str:
        # Azure OpenAI実装
        pass

📊 運用してみた結果

効果

  • 情報収集時間を週30分→0分に削減
  • 重要な更新の見逃しがゼロに
  • チーム全体での情報共有が自動化

課題と改善点

  1. 要約品質のばらつき

    • プロンプトエンジニアリングで改善中
    • Few-shot学習の導入を検討
  2. API利用料金

    • GPT-4o-miniで月額約$5程度
    • キャッシュ機能の実装を検討
  3. 通知のノイズ

    • 重要度スコアリング機能を追加予定
    • ユーザーごとのフィルタリング機能を検討

🎯 今後の展望

  • 複数チャンネル対応: サービスごとに異なるチャンネルへ通知
  • インタラクティブ機能: Slackボタンで詳細表示/非表示
  • トレンド分析: 更新頻度やカテゴリの可視化
  • カスタムアラート: 特定キーワードの即時通知

まとめ

GitHub ActionsとOpenAI APIを活用することで、AWS更新情報の自動通知システムを簡単に構築できました。特定サービスに特化した情報収集が自動化され、チーム全体の情報共有が大幅に改善されました。

参考リンク

0
0
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
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?