はじめに
にわのわです。
時代の逆光かはたまた先取りかRSS feedをAWSでホストしていこうと思います。
RSSというプロトコルを利用しつつも、特定のwebサイトの更新を取得するのではなく、
インターネットをプレイしている際に見つけた面白いサイトを自分のRSS feedに追加していきたい。というのが今回の目的です。
余談ですが、RSSは0.9x系だとRich Site Summary、1.x系だとRDF Site Summary、
2.0系だとReally Simple Syndicationの略語となっています。
ちなみに今回はAtomを使用します。
各バージョンについてはwikipediaがよくまとまっていると思います。
作ったもの
今回作ったものはこちらで公開しています。
AWS SAMで構築しているので
sam build
sam deploy -g
であなただけのRSSをホストできます。
仕様
/
GET
RSSFeedを取得する
中身
get.py
"""
get:
summary: 'RSSFeedを取得するAPI'
responses:
'200':
description: 'Successful response'
content:
application/xml:
example: |
<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<title>Your RSS Feed</title>
<link>http://yourwebsite.com</link>
<description>Your feed description</description>
</channel>
</rss>
x-amazon-apigateway-integration:
httpMethod: post
type: aws_proxy
uri:
Fn::Sub: arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${GetRSSFeedFunction.Arn}/invocations
"""
import json
import boto3
import os
def lambda_handler(event, context):
# S3クライアント作成
s3 = boto3.client('s3')
BUCKET_NAME = os.environ['BUCKET_NAME']
# S3からrss.xmlを取得。
# 対象ファイルがない場合はエラーを出力する
try:
response = s3.get_object(
Bucket=BUCKET_NAME,
Key='rss.xml'
)
except Exception as e:
print(e)
raise e
# Lambdaプロキシ統合に対応したレスポンスを返す
response = {
"statusCode": 200,
"headers": {
"Content-Type": "application/xml"
},
"body": response['Body'].read().decode('utf-8')
}
return response
POST
RSSfeedの追加を行う。
bodyにRSSFeedに追加したいURLを添える必要がある。
リクエストボディ例.json
{
"link": "https://qiita.com/niwanowa"
}
中身
post.py
"""
post:
summary: 'RSSFeedを登録するAPI'
requestBody:
required: true
content:
application/json:
schema:
type: object
properties:
link:
type: string
example: 'http://yourwebsite.com'
required:
- link
responses:
'200':
description: 'Content added to the RSS feed'
"""
import json
import boto3
import os
import feedgenerator
import feedparser
from datetime import datetime, timezone
import requests
from bs4 import BeautifulSoup
import time
os.environ['TZ'] = "UTC"
def lambda_handler(event, context):
print(event)
# S3クライアント作成
s3 = boto3.client('s3')
# 環境変数からS3のバケット名とファイル名を取得
BUCKET_NAME = os.environ['BUCKET_NAME']
FILE_NAME = os.environ['FILE_NAME']
# rssファイル読み込み
obj = s3.get_object(Bucket=BUCKET_NAME, Key=FILE_NAME)
rss_xml_string = obj['Body'].read().decode('utf-8')
# feedparserでパース
feed_parser = feedparser.parse(rss_xml_string)
print(feed_parser)
# FeedParserDictからfeedgeneratorのAtom用Feedオブジェクトを作成
feed_generator = feedgenerator.Atom1Feed(
title=feed_parser.feed.title,
link=feed_parser.feed.link,
description=feed_parser.feed.subtitle,
language=feed_parser.feed.language,
author_name=feed_parser.feed.author,
)
# Feedオブジェクトにパーサーのエントリーを追加
for entry in feed_parser.entries:
print(entry)
feed_generator.add_item(
title=entry.get('title'),
link=entry.get('link'),
description=entry.get('description', 'description is empty'),
pubdate=datetime.fromtimestamp(time.mktime(entry.updated_parsed)),
)
print(feed_generator.writeString('utf-8'))
# リクエストボディからlinkを取得
link = json.loads(event['body']).get('link')
# linkからページタイトルを取得
page_title = get_page_title(link)
# feedにエントリー追加
feed_generator.add_item(
title=page_title,
link=link,
description=page_title,
pubdate=datetime.now(),
)
# Stringに書き出し
rss_xml_string = feed_generator.writeString('utf-8')
# #s3にアップロード
s3.put_object(
Bucket=BUCKET_NAME,
Body=rss_xml_string.encode('utf-8'),
Key=FILE_NAME,
ContentType='application/xml'
)
# Lambdaプロキシ統合に対応したレスポンスを返す
response = {
"statusCode": 200,
"headers": {
"Content-Type": "application/json"
},
"body": json.dumps({"message": "Content added to the RSS feed"})
}
return response
def get_page_title(link):
# ページタイトルを取得する。
# ページタイトルが取得できない場合は、空文字を返す。
try:
response = requests.get(link)
soup = BeautifulSoup(response.text, "html.parser")
page_title = soup.title.string if soup.title else None
return page_title
except Exception as e:
print(e)
return None
/reset
GET
RSSFeedの初期化を行う。
ファイルがなければ作成し、ある場合は置換する。
中身
reset.py
"""
/reset:
get:
summary: 'RSSFeedをリセット(初回生成)するAPI'
responses:
'200':
description: 'Initialized RSSFeed'
x-amazon-apigateway-integration:
httpMethod: post
type: aws_proxy
uri:
Fn::Sub: arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${ResetRSSFeedFunction.Arn}/invocations
"""
import json
import boto3
import os
import feedgenerator
from datetime import datetime, timezone
def lambda_handler(event, context):
# S3クライアント作成
s3 = boto3.client('s3')
# 環境変数からS3のバケット名とファイル名を取得
BUCKET_NAME = os.environ['BUCKET_NAME']
FILE_NAME = os.environ['FILE_NAME']
# rssファイルの要素を初期化する。
feed_title = "にわのわのRSS"
feed_link = "https://example.com/rss"
feed_description = "にわのわさんのRSSFeed。"
feed_language = "ja"
feed_author = "にわのわ"
# エントリーの要素を変数で代入（例として2つのエントリーを生成）
entry_title = "にわのわさんのブログ"
entry_link = "https://hugo.niwanowa.tips/"
entry_summary = "にわのわさんがやってるブログだよ。"
# RSSフィードの基本構造を生成
feed = feedgenerator.Atom1Feed(
title=feed_title,
link=feed_link,
description=feed_description,
language=feed_language,
author_name=feed_author,
)
# フィードにエントリーを追加
feed.add_item(
title=entry_title,
link=entry_link,
description=entry_summary,
pubdate=datetime.now(timezone.utc)
)
# RSSフィードのXMLを生成
rss_xml_string = feed.writeString('utf-8')
# S3のrss.xmlを置換する。
s3.put_object(
Bucket=BUCKET_NAME,
Body=rss_xml_string.encode('utf-8'),
Key=FILE_NAME,
ContentType='application/xml'
)
# Lambdaプロキシ統合に対応したレスポンスを返す
response = {
"statusCode": 200,
"headers": {
"Content-Type": "application/json"
},
"body": json.dumps({"message": "Initialized RSSFeed"})
}
return response
おわりに
「にわのわさんの良いコンテンツ」みたいな感じでインターネットをプレイしていた際に、
良いものを見つけたら気軽に共有していきたい。という思いで作りました。