【Notion API × AWS】RSSで取得した記事をNotionに自動登録する仕組みをLambdaで作ってみた

僕はFeadlyやはてなブックマークなどで情報取集していて、

気になる記事とかはNotion Web Clipperでクリップして、BookMark用のデータベースに登録する運用をしています。

度々、見逃したり埋もれたりするので、自分が確認したいサイトについては、最初からNotionに自動登録して、

あとから確認できるようにし、登録された記事を確認して、BookMarkとして振り分ける運用にしたいと思って作ってみました。

成果物

  • RSSから取得した記事のタイトルと記事URLを下記のようにデータベースに登録されます。

構成図

  1. EventBrideで1時間毎に、Lambdaを起動する
  2. LambdaがAWS Systems ManagerからRSS情報を取得する
  3. 取得したRSSから前回取得した記事以降に、更新された記事タイトルと記事URLを取得する
  4. 取得した記事URLにアクセスして、スクレイピングを行う
  5. AWS Sercrets ManagerからNotionの接続情報を取得する
  6. 記事タイトル・記事URL・記事内容・記事のタグをNotionに送信
  7. 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

参考

タイトルとURLをコピーしました