AWS Lambdaで自作RSSリーダー開発

はじめに

こちらは

• Alexa×BedrockでAWS最新ニュースキャッチアップアプリ コンセプト・設計
• AWS Lambdaで自作RSSリーダー開発 (本記事)
• Self hosted AWS LambdaによるAlexaスキル開発
• AlexaでBedrockに質問をするスキル開発

の連載記事です。本記事では以下のアーキテクチャの作成を目指して解説をしていきます。

内容としては以下になりますので、部分的に活用することも可能です。

• RSS (Rich Site Summary)
• RSSリーダーの実装 (Python)
• Amazon Translateを使用した翻訳
• Amazon DynamoDBへの格納
• DynamoDBでのTTL (Time To Live)設定

RRSとは

RSSとはwebサイトの新着・更新情報を配信する技術を指します。Webページ内の見出しや本文の要約、更新情報等をXMLベースのフォーマットに従って配信されており、ユーザーは効率的にサイトの情報を取得できるようになります。

AWSが公式に提供しているブログやニュースなどもRSSに対応しています。
例えばAWS NEWSの場合、トップページにはRSSへのリンクが添付されており、<item>...</item>に記事の内容が記載されます。こちらが随時更新されるので、定期的に取得をすることで最新情報を取得することが可能になります。

AWS NEWS RSS (抜粋)

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<rss version="2.0">
  <channel>
      :
      :
    <item>
      <link>https://aws.amazon.com/about-aws/whats-new/2023/11/aws-iot-sitewise-supports-asset-model-components/</link>
      <guid isPermaLink="false">72911903d948cd8c25e6dd58be997b727e7b1c20</guid>
      <title>AWS IoT SiteWise now supports Asset model components</title> 
      <description>&lt;p&gt;AWS IoT SiteWise now supports Asset model components to help industrial customers create reusable components to derive new asset models and express variations easily. &lt;/p&gt;</description>
      <pubDate>Wed, 22 Nov 2023 19:06:00 +0000</pubDate>
      <category>general:products/aws-iot-sitewise,marketing:marchitecture/internet-of-things,general:products/aws-govcloud-us</category>
      <author>[e-mail]</author>
    </item>
    <item>
    :
    </item>
    :
  </channel>
</rss>

AWS関連のRSSは以下の記事にわかりやすくまとまっていたので、ご参考ください。
https://dev.classmethod.jp/articles/aws-rss-feeds/

開発

AWS Lambda構築

Lambda レイヤー

• feedparser
  • RSSフィードを扱いやすくするために使用

Lambdaコードのデプロイ

コード全行

lambda_function.py

import re
import feedparser
import boto3
from datetime import datetime
from dateutil import tz
from dateutil import parser

# DynamoDB
# boto3.resourceは非推奨のためclientを使用
dynamodb = boto3.client('dynamodb')
TABLE_NAME = 'rss_feeds'
# 何日後にDynamoDBから削除するのか設定
day = 3
TTL = 60 * 60 * 24 * day

# Amazon translate
translate = boto3.client('translate')
SRC_LANG = 'en'
TRG_LANG = 'ja'

# timezone
timezone = tz.gettz('Asia/Tokyo')

# RSS URL
RSS_URL = 'https://aws.amazon.com/jp/blogs/aws/feed/'

def lambda_handler(event, context):
    # RSS取得
    data = feedparser.parse(RSS_URL)

    # 1件ずつitemを取得
    for entry in data.entries:
        title = entry.title
        link = entry.link
        # descriptionにはHTMLタグが含まれるので削除
        description = remove_http_tag(entry.description)
        # DynamoDBのTTLで使用するため、unixtimeも算出
        published, unixtime =  get_time_info(entry.published)
        limit_unixtime = unixtime + TTL
        category = entry.category
        
        # 既存判断。既存の場合はスキップ
        if not is_item_exist(link):
            # タイトルと要約はAmazon Translateで翻訳
            ja_title = get_translate_text(title)
            ja_description = get_translate_text(description)
            # DynamoDBへ格納
            put_item(title, link, description, published, category, ja_title, ja_description, limit_unixtime)
            

def remove_http_tag(str):
    return re.sub(re.compile('<.*?>'), '', str)
    

def is_item_exist(link) -> bool:
    options = {
        'TableName': TABLE_NAME,
        'Key': {
            'link': {'S': link},
        }
    }
    
    try:
        response = dynamodb.get_item(**options)
        response['Item']
        print("The Item Already Exist")
        return True
    except KeyError:
        print("New Item : " + link)
        return False

def put_item(title, link, description, published, category, ja_title, ja_description, limit_unixtime):
    options = {
        'TableName': TABLE_NAME,
        'Item': {
            'title': {'S': title},
            'link': {'S': link},
            'description': {'S': description},
            'published': {'S': published},
            'category': {'S': category},
            'ja_title': {'S': ja_title},
            'ja_description': {'S': ja_description},
            'limit_unixtime': {'N': str(limit_unixtime)}
        }
    }
    # DynamoDB格納
    dynamodb.put_item(**options)
    
def get_translate_text(text):
    # 英語→日本語訳
    response = translate.translate_text(
        Text=text,
        SourceLanguageCode=SRC_LANG,
        TargetLanguageCode=TRG_LANG
    )

    return response.get('TranslatedText')
    
def get_time_info(rss_pub_date):
    # datetimeに変換
    datetime = parser.parse(rss_pub_date).astimezone(timezone)
    
    # iso8601に変換
    iso_datetime = datetime.isoformat()
    
    # unixtime取得
    unixtime = datetime.timestamp()
    
    return iso_datetime, unixtime
    

タイムアウト設定
Lambdaのデフォルトは3秒であり、今回の処理は3秒を上回ることが予想されるため1分程度にします

cf. 検証ではRSS100件-20秒程度で実行された

ロール設定
以下のポリシーを設定する。

• Lambda作成時にデフォルトで作成されるポリシー
• TranslateReadOnly (マネージドポリシー)
• DynamoDB用のポリシー

dynamodb-policy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "dynamodb:PutItem",
                "dynamodb:GetItem"
            ],
            "Resource": "arn:aws:dynamodb:<YOUR REGION>:<ACCOUNT ID>:table/rss_feeds"
        }
    ]
}

EventBridge設定

日次で実行できるようにLambdaのトリガ設定から、EventBridgeを設定します。毎日実行する場合にはcron式 (0 15 * * ? *) を設定します。(UTCであることに注意)

DynamoDB設定

テーブルの準備
Lambdaのコードに合わせて、rss_feedsテーブルを準備しておきます。
パーティションキーはlinkと設定しました。また、日次でしか動かないのでオンデマンドキャパシティモードを選択しました。

TTL設定
TTLとはDynamoDBで古くなったデータを自動で削除する機能です。
TTLはunixtimeの属性に対して有効にすることができ、DynamoDBはその属性値を監視し続けます。DynamoDBは現在時刻のunixtimeを常に持っており、その値と比較して対象カラムのunixtimeが小さい時、そのデータを削除対象として削除します。
https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/howitworks-ttl.html

Lambdaのコードに合わせてlimit_unixtime項目に対してTTLを設定します。

設定が完了すると以下のようなDynamoDBテーブルが出来上がります。

Amazon translate 設定

特に必要がないです。

おわりに

Eventbridgeで設定した時刻でDynamoDBに最新情報が格納されました。
本記事ではAWS最新ニュースを取得するアプリケーションを構築しましたが、RSSに対応する記事であればどの記事でも対応可能です。また、部分的な技術も他に活用できますので、ぜひご参考ください。