- 初めに
- 環境
- 準備
- はてなブログの一覧を取得する
- Zennの記事一覧を取得する
- NotionのDBに記事を追加する
- それぞれのデータを取得して、日付ソートして書き込む
- GitHub Actionsを使って定期的に実行する
初めに
自分のポートフォリオをNotionで作っている中で執筆した記事をまとめたいと考えました.
そこで、普段記事を書いている はてなブログとZennの投稿記事を以下の画像のように Notionのデータベースに自動でまとめてくれるようにしました
環境
準備
環境構築
今回の開発では、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の記事も以下の要件を満たすようにします。
- 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