初めに

自分のポートフォリオをNotionで作っている中で執筆した記事をまとめたいと考えました.
そこで、普段記事を書いているはてなブログとZennの投稿記事を以下の画像のように Notionのデータベースに自動でまとめてくれるようにしました

環境

GitHub Actions(自動化環境)
Python 3.11
Docker
各種API

準備

環境構築

今回の開発では、Dockerを使用してPythonの開発環境を作成します。

以下が Dockerfile 及び compose.yaml です

Dockerfile

FROM python:3.11
USER root

# Python依存関係のインストール
COPY requirements.txt /tmp/
RUN pip install --upgrade pip setuptools \
    && pip install --requirement /tmp/requirements.txt

# スクリプトのコピー
COPY . /opt

compose.yaml

version: '3'
services:
  article-aggregator:
    restart: always
    build: .
    container_name: 'article-aggregator'
    working_dir: '/root/'
    environment:
      - NOTION_API_KEY=${NOTION_API_KEY}
      - NOTION_DATABASE_ID=${NOTION_DATABASE_ID}
    tty: true
    env_file:
      - .env
    volumes:
      - ./opt:/root/opt

以下のコマンドで、ビルド及び実行をします。

docker-compose up --build

Notion APIの取得

my-integrationsから新しいインテグレーションを作成します。この時のAPI Keyはこの後に使用するので、メモしておきます。

作成したインテグレーションを更新したいNotion データベースのコネクトに追加します

はてなブログの一覧を取得する

まずは自分のブログの AtomPubの ルートエンドポイント と API Key を取得します。

以下の要件で、記事の一覧を取得します。

下書きの記事は取得しない
NotionのDBに入れる形(タイトル、URL、日付) でまとめる
一回で取得できる上限があるので、nextがあるかどうかを確認する

これを満たすために以下のようなコードを作成しました

import requests
import requests
import feedparser
from requests.auth import HTTPBasicAuth  # HTTPBasicAuthのインポートを追加

class HatenaScraper:
    def __init__(self, baseurl_, user_, password_):
        self.baseurl = baseurl_
        self.user = user_
        self.password = password_
        self.entries = []

    def load_entry(self):
        ret = self.__load_entry(self.baseurl + "entry")
        self.__show_titles(ret.entries)

        next = self.__get_next_data(ret)
        while next != "":
            ret = self.__load_entry(next)
            self.__show_titles(ret.entries)
            next = self.__get_next_data(ret)

    def __load_entry(self, url):
        response = requests.get(url, auth=HTTPBasicAuth(self.user, self.password))  # self.userとself.passwordを使用
        return feedparser.parse(response.content)

    def __get_next_data(self, response):
        try:
            return response.feed.links[1].href
        except:
            return ""

    def __show_titles(self, entries):
        for entry in entries:
            if entry.app_draft == "no":
                title = entry.title
                published = entry.published  # 'published' 属性を使用して日時を取得
                # print(f"タイトル: {title}, 投稿日時: {published}")
                article_info = {
                    'title': entry.title,
                    'url': entry.link,
                    'created_at': entry.published
                }
                self.entries.append(article_info)

Zennの記事一覧を取得する

https://zenn.dev/api/articlesがAPIとしてあるので、こちらを使用します。

またZennの記事一覧取得は、以下の作者様のコメントにあるように https://zenn.dev/api/articleshttps://zenn.dev/ユーザー名/feed?all=1でも取得ができるみたいです

zenn.dev

Zennの記事も以下の要件を満たすようにします。

NotionのDBに入れる形(タイトル、URL、日付) でまとめる

これらの要件を満たして、上記APIを使って一覧を取得するコードは以下です。

import requests
from datetime import datetime


class ZennScraper:
    def __init__(self, url):
        """
        ZennScraperクラスのコンストラクタ。
        指定したURLを初期化し、空の記事リストを作成する。

        Args:
            url (str): スクレイピングするZennのURL。
        """
        self.url = url
        self.articles = []

    def get_articles(self, username):
        base_url = "https://zenn.dev/api/articles"

        response = requests.get(base_url, params={'username': username, 'order': 'latest'})
        if response.status_code == 200:
            data = response.json()
            for article in data["articles"]:
                # 日付の解析とフォーマット
                published_at = article.get("published_at")
                if published_at:
                    date_obj = datetime.strptime(published_at, '%Y-%m-%dT%H:%M:%S.%f%z')
                    formatted_date = date_obj.strftime('%Y-%m-%d')
                article_info = {
                    'title': article["title"],
                    'url': f"https://zenn.dev{article['path']}",
                    'created_at': formatted_date
                }
                self.articles.append(article_info)

NotionのDBに記事を追加する

PythonからNotionのAPIを操作するときに便利なライブラリのnotion-sdk-pyがあるので、こちらを使用します。

Notionに記事を追加する場合は、以下の流れで行います。 1. 以前に更新した記事を一度削除(アーカイブ)する 2. 新しく取得した記事を追加する

この時に1の処理を直列で処理をすると時間がかかるので、Pythonの ThreadPoolExecutor を使って並列にします。

NotionDBの記事を全て削除する

    def delete_page(self, page_id):
        """
        指定された`page_id`のページをNotionデータベースから削除（アーカイブ）する。
        """
        self.notion.pages.update(page_id=page_id, archived=True)

    def delete_all_pages(self):
        """
        データベース内の全ページを削除する。`ThreadPoolExecutor`を使用して並列削除を行う。
        """
        start_cursor = None
        has_more = True

        while has_more:
            response = self.notion.databases.query(database_id=self.database_id, start_cursor=start_cursor)
            with ThreadPoolExecutor() as executor:
                [executor.submit(self.delete_page, page['id']) for page in response['results']]

            has_more = response['has_more']
            start_cursor = response.get('next_cursor')

ページを追加する

    def add_article(self, title, url, date):
        """
        新しい記事をNotionデータベースに追加する。
        """
        new_page = {
            "Title": {"title": [{"text": {"content": title}}]},
            "Link": {"url": url},
            "Date": {"date": {"start": date}}
        }
        self.notion.pages.create(parent={"database_id": self.database_id}, properties=new_page)

コードの全容は以下になります。

from concurrent.futures import ThreadPoolExecutor
from notion_client import Client


class NotionManager:
    def __init__(self, api_key, database_id):
        """
        Notion APIのクライアントを初期化し、データベースIDを設定する。
        """
        self.notion = Client(auth=api_key)
        self.database_id = database_id

    def add_article(self, title, url, date):
        """
        新しい記事をNotionデータベースに追加する。
        """
        new_page = {
            "Title": {"title": [{"text": {"content": title}}]},
            "Link": {"url": url},
            "Date": {"date": {"start": date}}
        }
        self.notion.pages.create(parent={"database_id": self.database_id}, properties=new_page)

    def delete_page(self, page_id):
        """
        指定された`page_id`のページをNotionデータベースから削除（アーカイブ）する。
        """
        self.notion.pages.update(page_id=page_id, archived=True)

    def delete_all_pages(self):
        """
        データベース内の全ページを削除する。`ThreadPoolExecutor`を使用して並列削除を行う。
        """
        start_cursor = None
        has_more = True

        while has_more:
            response = self.notion.databases.query(database_id=self.database_id, start_cursor=start_cursor)
            with ThreadPoolExecutor() as executor:
                [executor.submit(self.delete_page, page['id']) for page in response['results']]

            has_more = response['has_more']
            start_cursor = response.get('next_cursor')

それぞれのデータを取得して、日付ソートして書き込む

はてなブログとZennの記事一覧を取得して、Notionに書き込む処理ができました。これから各記事をまとめて日付順にソートをして実行するコードを作ります.
(環境変数がハードコーディングになっている部分は、各自で直してください...)

import asyncio
from dotenv import load_dotenv
import os
import sys

from zenn_scraper import ZennScraper
from notion_manager import NotionManager
from haterna_scraper import HatenaScraper


async def main():
    load_dotenv()

    notion_api_key = os.environ["NOTION_API_KEY"]
    notion_database_id = os.environ["NOTION_DATABASE_ID"]

    publication_url = "url"

    zenn_scraper = ZennScraper(publication_url)
    zenn_scraper.get_articles("user name")
    
    hatena_scraper = HatenaScraper("hatena blog atom url", "user id", "api key")
    hatena_scraper.load_entry()
        
    # Zennとはてなの記事を結合
    articles = zenn_scraper.articles + hatena_scraper.entries

    # 記事を日付でソート
    articles.sort(key=lambda x: x['created_at'], reverse=True)
    
    if len(articles) == 0:
        print("記事がありません")
        sys.exit(0)

    notion_manager = NotionManager(notion_api_key, notion_database_id)
    notion_manager.delete_all_pages()

    for article in articles:
        notion_manager.add_article(article['title'], article['url'], article['created_at'])


# 非同期イベントループを使用してmain関数を実行
asyncio.run(main())

GitHub Actionsを使って定期的に実行する

上記で処理はできましたが、毎回自分で実行するのは大変なので GitHub Actionsを使って自動化します

name: Aggregator Zenn Article

on:
  schedule:
    # 一日ごとに実行
    - cron: "0 0 * * *"
  workflow_dispatch: {}
  push:
    branches:
      - main

jobs:
  run-docker-compose:
    timeout-minutes: 10
    runs-on: ubuntu-latest
    env:
      NOTION_API_KEY: ${{ secrets.NOTION_API_KEY }}
      NOTION_DATABASE_ID: ${{ secrets.NOTION_DATABASE_ID }}
    steps:
      - name: Checkout code
        uses: actions/checkout@v4.1.1
      - name: create env file
        run: |
          touch .env
          echo "NOTION_API_KEY=${NOTION_API_KEY}" >> .env
          echo "NOTION_DATABASE_ID=${NOTION_DATABASE_ID}" >> .env
      - name: Build and run Docker Compose
        run: |
          docker-compose up -d --build
      - name: Run Aggregator
        run: |
          docker-compose exec -T article-aggregator python opt/main.py