僕はFeadlyやはてなブックマークなどで情報取集していて、
気になる記事とかはNotion Web Clipperでクリップして、BookMark用のデータベースに登録する運用をしています。
度々、見逃したり埋もれたりするので、自分が確認したいサイトについては、最初からNotionに自動登録して、
あとから確認できるようにし、登録された記事を確認して、BookMarkとして振り分ける運用にしたいと思って作ってみました。
成果物
- RSSから取得した記事のタイトルと記事URLを下記のようにデータベースに登録されます。
構成図
- EventBrideで1時間毎に、Lambdaを起動する
- LambdaがAWS Systems ManagerからRSS情報を取得する
- 取得したRSSから前回取得した記事以降に、更新された記事タイトルと記事URLを取得する
- 取得した記事URLにアクセスして、スクレイピングを行う
- AWS Sercrets ManagerからNotionの接続情報を取得する
- 記事タイトル・記事URL・記事内容・記事のタグをNotionに送信
- Notionに記事が登録される
使うための準備
AWS SAM
当機能では、AWS SAMで構築しています。
SAM CLIを使用するには、次のツールをインストールしておく必要があります。
Notion integrationを作成
Notion APIでNotionのデータベースに記事を登録するには、インテグレーションを作成する必要があります。
下記リンクにアクセスし、「New Integration」からインテグレーションを作成します。
インテグレーションを作成したらInternal Integration Tokenをコピーして控えてください。
後程、AWS Secrets Managerに登録する際に必要になります。
https://www.notion.so/my-integrations
ワークスペースにインテグレーションを招待
Notionページにインテグレーションを招待する必要があります。
招待することで、Notion APIと連携が可能になります。
データベースIDを取得する
当機能は、データベースに書き込みを行いますので、データベースIDを取得する必要があります。
ページ内に作成したデータベースしたら共有でリンクをコピーしてください。
URLの「?v」の前にある文字列部分がデータベースIDになりますので、控えてください。
https://www.notion.so/{workspace_name}/{database_id}?v={view_id}
後程、AWS Secrets Managerに登録する際に必要になります。
AWS System Managerの設定
下記を作成する
- パラメータ名：RSSURLList
- 利用枠：標準
- タイプ：文字列
- データ型：text
- 値 (下記json参照)
- url：記事の取得先URL
- tag：Notionに記事登録時のタグ名称
{
"rsslist": [
{
"url": "https://aws.amazon.com/jp/about-aws/whats-new/recent/feed/",
"tag": "AWS",
"name":"AWSの最新情報"
},
{
"url": "https://aws.amazon.com/jp/blogs/news/feed/",
"tag": "AWS",
"name":"Amazon Web Services ブログ"
},
{
"url": "https://qiita.com/tags/aws/feed",
"tag": "AWS",
"name":"Qiita - AWS"
},
{
"url": "https://qiita.com/tags/docker/feed",
"tag": "Docker",
"name":"Qiita - Docker"
},
{
"url": "https://qiita.com/tags/python/feed",
"tag": "Python",
"name":"Qiita - Python"
},
{
"url": "https://dev.classmethod.jp/feed/",
"tag": "技術ブログ",
"name":"クラスメソッド"
},
{
"url": "https://zenn.dev/feed",
"tag": "技術ブログ",
"name":"Zenn - トレンド"
},
{
"url": "https://zenn.dev/topics/aws/feed",
"tag": "AWS",
"name":"Zenn - AWS"
},
{
"url": "https://zenn.dev/topics/docker/feed",
"tag": "Docker",
"name":"Zenn - Docker"
},
{
"url": "https://zenn.dev/topics/python/feed",
"tag": "Python",
"name":"Zenn - Python"
}
]
}
Secret Manegerの設定
- シークレットのタイプ：その他のシークレットのタイプ
- キー/値のペア
- NOTION_TOKEN：Integration Token
- DATABASE_ID：データベースID
- 暗号化キー：DefaultEncryptionKey
実装の解説
ソースコードはこちら
RSS取得
RSS取得と解析は、下記の記事を参考にしました。
AWS System ManagerからRSS取得先を取得します。
def get_target_url() -> List[str]:
"""
取得対象にするrssのURLを返却する
"""
region_name = "ap-northeast-1"
ssm = boto3.client('ssm',region_name=region_name)
url_param: str = ssm.get_parameter(
Name='RSSURLList'
)['Parameter']['Value']
json_dict = json.loads(url_param)
return json_dict
feedparserを使い記事を解析します。
現在時刻と記事の公開日を比較し、1時間経過した記事を取得対象としています。
def get_rss(endpoint: str, tag: str, interval: int = 60) -> List[RssContent]:
"""
rssのxmlを返すendpoint(url)からrss情報を取得し、必要な情報だけ抜き出す
interval分以内の記事だけを返す。定期実行はinterval分と同じ間隔にすればよい
intervalを負数にすると全記事返す(デバッグ用)
"""
nowtime = datetime.now(timezone(timedelta(hours=+9), 'JST'))
feed = feedparser.parse(endpoint)
rss_list: List[RssContent] = []
for entry in feed.entries:
if not entry.get("link"):
continue
published = convert_time(entry.published_parsed)
if (nowtime - published).total_seconds() // 60 <= interval or interval < 0:
rss_content = RssContent(
title=entry.title,
url=entry.link,
tag=tag,
published_date=published
)
rss_list.append(rss_content)
return rss_list
シークレットの取得
Secret Manegerからの取得ロジックについては、下記の記事を参考にしました。
取得したいシークレットキーを引数に指定することで、シークレットの値を取得しています。
import boto3
import base64
from botocore.exceptions import ClientError
import ast
def get_secrets_manager_key_value(secret_name: str, secret_key: str) -> str:
"""AWS Secrets Managerからシークレットキーの値を取得する."""
value = ''
secrets_dict = get_secrets_manager_dict(secret_name)
if secrets_dict:
if secret_key in secrets_dict:
# secrets_dictが設定されていてsecret_keyがキーとして存在する場合
value = secrets_dict[secret_key]
else:
print('シークレットキーの値取得失敗：シークレットの名前={}、シークレットキー={}'.format(secret_name, secret_key))
return value
def get_secrets_manager_dict(secret_name: str) -> dict:
"""Secrets Managerからシークレットのセットを辞書型で取得する"""
region_name = "ap-northeast-1"
secrets_dict = {}
if not secret_name:
print('シークレットの名前未設定')
else:
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
print('シークレット取得失敗：シークレットの名前={}'.format(secret_name))
print(e.response['Error'])
else:
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
secret = base64.b64decode(get_secret_value_response['SecretBinary'])
secrets_dict = ast.literal_eval(secret)
return secrets_dict
Notion登録
PythonでNotion APIを扱うため、「Notion SDK for Python」と
Notion登録前に記事の内容をスクレイピングするため、「BeautifulSoup」を利用しています。
targetTagsに設定されている値がスクレイピングの対象になります。
対象のタグにあったNotionのリクエストBodyを作り上げ、リクエストしています。
NotionへのリクエストBodyは、公式リファレンスを参照
import os
import traceback
from typing import List
from get_secrets import get_secrets_manager_key_value
from bs4 import BeautifulSoup
import requests
from notion_client import Client, APIResponseError
import urllib.request, urllib.error
def register_notion(rss_list:dict) -> None:
"""
Notionの指定したデータベースに記事のタイトル、タグ、URL、記事内容を登録する
"""
try:
## AWS Secrets Managerに設定しているNotionのシークレットを取得する
notion_token = get_secrets_manager_key_value('notion_rss', 'NOTION_TOKEN')
## AWS Secrets Managerに設定しているNotionのシークレットを取得する
database_id = get_secrets_manager_key_value('notion_rss', 'DATABASE_ID')
## 認証を行う
notion = Client(auth=notion_token)
## 取得したRSS数文登録を行う
for rss in rss_list:
block = []
if not checkURL(rss.url):
continue
soup = BeautifulSoup(requests.get(rss.url).content, 'html.parser')
articleTag = soup.find_all("div", {"class": "content"})
if not articleTag:
articleTag = soup.find_all(['section','article'])
for article in articleTag:
targetTags = article.find_all(['h1','h2','h3','p','span','img','li','pre','blockquote'])
for tag in targetTags:
if tag.name == 'h1':
if tag.get_text(strip=True):
block.append(append_message("heading_1",tag.get_text(strip=True)))
if tag.name == 'h2':
if tag.get_text(strip=True):
block.append(append_message("heading_2",tag.get_text(strip=True)))
if tag.name == 'h3':
if tag.get_text(strip=True):
block.append(append_message("heading_3",tag.get_text(strip=True)))
if tag.name == 'li':
if tag.get_text(strip=True):
block.append(append_message("bulleted_list_item",tag.get_text(strip=True)))
if tag.name == 'p':
if tag.parent.name == 'blockquote':
continue
if tag.get_text(strip=True):
block.append(append_message("paragraph",tag.get_text(strip=True)))
if tag.name == 'img':
if checkURL(tag['src']) and checkExtension(tag['src']):
block.append(append_image(tag['src']))
if tag.name == 'pre':
if tag.get_text(strip=True):
block.append(append_code(tag.get_text(strip=True)))
if tag.name == 'blockquote':
if tag.get_text(strip=True):
block.append(append_message("quote",tag.get_text(strip=True)))
## Notionに登録を行う
notion.pages.create(
parent={'database_id': database_id},
properties=property_data(rss),
children=block
)
Lambda・ポリシー・ロール
起動時間とポリシーとロールは、template.yamlに記述しています。
NotionRegisterFunction:
Type: AWS::Serverless::Function
Properties:
FunctionName: "lambda-notion-rss-register"
CodeUri: ./lambda_notion_rss_register
Handler: app.lambda_handler
Runtime: python3.8
Architectures:
- x86_64
Role: !GetAtt NotionRegisterFunctionRole.Arn
Events:
RSSGetSchedule:
Type: Schedule
Properties:
Schedule: rate(1 hour) # 1時間毎
Input: |
{
"region": "tokyo"
}
NotionRegisterFunctionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: "2012-10-17"
Statement:
- Effect: "Allow"
Action: "sts:AssumeRole"
Principal:
Service: lambda.amazonaws.com
ManagedPolicyArns:
- !Ref NotionRegisterFunctionPolicy
NotionRegisterFunctionPolicy:
Type: AWS::IAM::ManagedPolicy
Properties:
PolicyDocument:
Version: "2012-10-17"
Statement:
- Effect: "Allow"
Action:
- "logs:CreateLogStream"
- "logs:CreateLogGroup"
- "logs:PutLogEvents"
Resource: "arn:aws:logs:ap-northeast-1:*:*"
- Effect: "Allow"
Action:
- "secretsmanager:GetSecretValue"
- "ssm:GetParameters"
- "ssm:GetParameter"
Resource: "*"
ビルド・実行・デプロイ
コマンド
コマンドは下記
sam build
sam local invoke NotionRegisterFunction --event events/event.json
sam deploy --guided
デプロイ結果
Lambda
EventBridge
Role
Policy
