👏

最新AWS情報を公式RSSから自動翻訳してLINEに通知

に公開
AWS
LINE
AWS Lambda
LINE Bot
amazontranslate
tech

はじめに

AWSの機能アップデートやブログ更新など日々更新される情報はとても多いです。
しかし、エンジニアとして情報をキャッチアップしていくことはてても大切です。
そこで、LINE BOTを活用して数あるAWS公式RSSから最新情報を通知するシステムを作りましたので、その構成を紹介します。
構成図にはslackも書いてありますが、今回はLINEのみです。

slackを使用した記事は以下を参照ください。
Bot Token + chat.postMessageを使用した記事はこちら。
https://zenn.dev/t_oishi/articles/b1d6e7c36797c9
Incoming Webhooksを使用した記事はこちら。
https://zenn.dev/t_oishi/articles/1a9884072d4b18

構成図

構成図は以下です。

  • Lambda
    • Python 3.13
    • Lambdaレイヤー
      • asyncio（非同期HTTP通信用）
      • aiohttp（RSSフィードの解析用）
      • feedparser（LINE API通信用）
  • EventBridge
    • スケジュール
  • Translate
    • 英語のRSSを日本語化
  • LINE BOT
    • LINE Massaging API使用（LINE Developers,LINE Official Account Manager）

料金

ほぼ０円で実装可能です。

  • Lambda
    実行時間料金
    Lambda無料枠：400,000 GB秒/月
    GB秒単位で課金（メモリ×実行時間）
    例: 128MBのLambdaを1秒実行 → 0.128 GB × 1秒 = 0.128 GB秒
    128MBのLambdaは 1か月で約868時間まで無料。
    今回の実装
    1日 1回で5分として 1,152GB秒/月。（無料枠範囲内）

  • EventBridge
    EventBridge Scheduler
    毎月1,400万回の呼び出しが無料
    今回の実装
    1日 1回　なので31回。（無料枠範囲内）

  • Translate
    基本料金:100万文字あたり$15.00で従量課金。（無料利用枠は例外）
    今回の実装
    タイトルのみなので100万文字には達しない。（無料枠範囲内）

  • LINE Messaging API
    コミュニケーションプラン（無料プラン）: 200通/月まで無料。
    今回の実装
    1日 1通～5通であれば、31回～155回。（無料枠範囲内）

AWS RSSの種類

こちらの記事にわかりやすくまとまっています。

https://dev.classmethod.jp/articles/aws-rss-feeds/#RSS%25E3%2583%2595%25E3%2582%25A3%25E3%2583%25BC%25E3%2583%2589%25E4%25B8%2580%25E8%25A6%25A7

今回こちらだけ補足しました。
週刊AWS
https://aws.amazon.com/jp/blogs/news/tag/週刊aws/
https://aws.amazon.com/jp/blogs/news/tag/週刊aws/feed/

実装方法

LINE Developersへ登録

https://developers.line.biz/console/

ログインできたらプロバイダーを作成します。


プロバイダーはなぜ必要なのか？
LINEミニアプリなどLINE APIを利用する際には、サービスに対応したチャネルを作成する必要があります。
プロバイダーはなぜ必要なのか？
INEで提供するサービスやアプリを管理するための土台がプロバイダーにあたり、具体的なサービスやアプリを動かすためにチャネルがあります。
チャネルは、LINE公式アカウントやLINEミニアプリなどを通して、ユーザーとやり取りをする接点になります。
チャネルは必ずプロバイダーに属する必要があるため、LINEでサービスを利用する際は、まずプロバイダーを作成し、 その中に必要なチャネルを追加していく形になります。この仕組みを通じて、サービスごとに適切な管理や設定が可能になります。

https://lineapiusecase.com/ja/api/provider.html

LINE Messaging APIの有効化

プロバイダを作成し、Messaging APIを選択すると、以下の画面に移ります。

LINE公式アカウントを作成するをクリックして、LINE公式アカウントを作成します。
以下のようにして、アカウントを作成します。

以下になれば作成完了です。LINE Official Account Managerを開きましょう。

次に右上の設定からMessaging APIを開きます。

下記のボタンをクリックして有効化の設定をしていきます。

先ほど作成したプロバイダーを選択します。

ここは何も入力いりません。


これでLINE Messaging APIを有効化できました。

チャネルアクセストークンとユーザーIDの取得

LINE Developersへ戻り、チャネルアクセストークンとユーザーIDを取得します。
後に使うので控えておきましょう。



発行を押して出力します。

これでLINE側の設定は完了です。

Lambda関数の実装

以下が今回のソースコードです。
環境変数として、　以下を指定します。
LINE_TOKEN = チャネルアクセストークン
LINE_USER_ID= ユーザーID

フローとしては、

  1. 並行処理でRSS取得 → 31個のフィードを同時に取得
  2. 当日記事の抽出 → 各フィードの最新10記事から今日の記事のみ選別
  3. タイトル翻訳 → 英語タイトルを日本語に変換（"translate": Trueのもののみ）
  4. LINE送信 → ブログ別にメッセージ送信

下記のソースだけでは動作しません。
asyncio,aiohttp,feedparserをLambdaレイヤーの作成・アップロードをする必要があります。

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"]

# 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 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)

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

Lambdaレイヤーの作成・アップロード

pipがインストールされている前提で、下記を実行します。
このzipファイルをLambdaレイヤーとしてアップロードします。
その後、Lambda関数にアタッチします。

mkdir lambda-layer
cd lambda-layer
mkdir python
cd python

# 必要なライブラリをpythonディレクトリにインストール
pip install asyncio aiohttp feedparser -t .

# ZIPファイルを作成
zip -r lambda-layer.zip python/

Translate用のポリシーをロールにアタッチ

今回はAWSマネージドポリシーを使用します。
LambdaについているIAMロールにTranslateReadOnlyをアタッチしましょう。

テスト

Lambda関数のテストイベントでテストしてみましょう。
RSSの数が多いとタイムアウトになるので、実行時間を5分くらいに設定しておきましょう。
メモリを増やすと時間を短くできます。
テストが成功すると、LINEに通知されます。
失敗する場合はCloudWatchで確認しましょう。

EventBridgeの設定

設定でターゲットとして先ほど作成したLambdaを指定します。
以下の設定では、毎日 午前8時30分にLambdaが実行されます。

おわりに

今回紹介した方法を使えば、AWS公式RSSの最新情報を自動で取得し、LINEに通知できます。
LambdaやEventBridge、Translate、LINE Messaging APIを組み合わせるだけで、ほぼ無料で実装可能です。
情報収集の効率化やチームへの共有にも役立つので、ぜひ活用してみてください。

slackを使用した記事は以下を参照ください。
Bot Token + chat.postMessageを使用した記事はこちら。
https://zenn.dev/t_oishi/articles/b1d6e7c36797c9
Incoming Webhooksを使用した記事はこちら。
https://zenn.dev/t_oishi/articles/1a9884072d4b18

GitHubで編集を提案

Discussion