GitHub Actionsによるnote.comのスクレイピングをやってみた
この3連休も新規プロジェクトの開発に集中している。
コンテンツ系のプロジェクトではあるのだけど、一方でこういう時期しか書けないコードや実験というものもあり、クリエイティブAI系のコードを書きながら、オートメーション系の業務や処理を見直している。
そんな中でGitHub Actionsは git pushによるCI/CDだけでなく、crondを使った定期実行なども可能であることに気づいた。
例えば、noteのブログをスクレイピングしてmarkdownにして保存するようなpythonスクリプトを実行したりもできる。
.github/workflows/fetch_note_rss.yml
name: Fetch and Process note.com RSS
on:
schedule:
- cron: '0 */12 * * *' # 12時間ごとに実行
workflow_dispatch: # 手動実行用
env:
RSS_URL: "https://note.com/(username)/rss"
permissions:
contents: write
jobs:
fetch-and-process:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.x'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install feedparser requests html2text playwright slackweb
- name: Setup Playwright
run: playwright install chromium
- name: Fetch and process RSS
env:
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
GITHUB_REPOSITORY: ${{ github.repository }}
run: |
python scripts/process_note_content.py
if [ $? -ne 0 ]; then
echo "Python script failed. Outputting error log:"
cat error.log || echo "No error log found"
exit 1
fi
- name: Configure Git
run: |
git config --global user.name 'GitHub Action'
git config --global user.email 'action@github.com'
git remote set-url origin https://x-access-token:${{ secrets.GITHUB_TOKEN }}@github.com/${{ github.repository }}.git
- name: Commit and push changes
run: |
git add _posts/ assets/
git commit -m "Update note.com content" || echo "No changes to commit"
git push || echo "No changes to push"
- name: Output error log if failed
if: failure()
run: |
echo "Workflow failed. Outputting error log:"
cat error.log || echo "No error log found"
continue-on-error: true
process_note_content.py
import feedparser
import requests
import html2text
import json
import os
import sys
import traceback
from datetime import datetime
from playwright.sync_api import sync_playwright
import slackweb
import logging
import re
from pathlib import Path
# ログ設定
logging.basicConfig(filename='error.log', level=logging.ERROR,
format='%(asctime)s - %(levelname)s - %(message)s')
def fetch_note_rss():
rss_url = os.environ.get('RSS_URL')
if not rss_url:
raise ValueError("RSS_URL environment variable is not set")
feed = feedparser.parse(rss_url)
return feed.entries
def download_html(url):
response = requests.get(url)
return response.text
def html_to_markdown(html):
h = html2text.HTML2Text()
h.ignore_links = False
return h.handle(html)
def html_to_png(html, output_path):
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.set_content(html)
page.screenshot(path=output_path, full_page=True)
browser.close()
def get_hash_from_url(url):
match = re.search(r'/n/([a-zA-Z0-9]+)$', url)
return match.group(1) if match else None
def remove_unwanted_lines(markdown):
unwanted_keywords = ['ログイン', '会員登録', '返金可', '割引']
lines = markdown.split('\n')
filtered_lines = [line for line in lines if not any(keyword in line for keyword in unwanted_keywords)]
filtered_markdown = '\n'.join(filtered_lines)
# Remove lines after "* [ noteプレミアム ](https://premium.lp-note.com)"
premium_index = filtered_markdown.find("* [ noteプレミアム ](https://premium.lp-note.com)")
if premium_index != -1:
filtered_markdown = filtered_markdown[:premium_index]
return filtered_markdown
def save_images(markdown, hash_id):
image_pattern = r'!\[.*?\]\((.*?)\)'
images = re.findall(image_pattern, markdown)
current_month = datetime.now().strftime("%Y-%m")
assets_dir = f'assets/{current_month}'
os.makedirs(assets_dir, exist_ok=True)
for i, image_url in enumerate(images, start=1):
response = requests.get(image_url)
if response.status_code == 200:
image_path = f'{assets_dir}/{hash_id}_{i}.png'
with open(image_path, 'wb') as f:
f.write(response.content)
markdown = markdown.replace(image_url, f'/assets/{current_month}/{hash_id}_{i}.png')
return markdown
def save_content(title, markdown, png_path, url, publish_date):
hash_id = get_hash_from_url(url)
if not hash_id:
raise ValueError(f"Could not extract hash from URL: {url}")
posts_dir = '_posts'
os.makedirs(posts_dir, exist_ok=True)
md_path = os.path.join(posts_dir, f"{hash_id}.md")
# Check if file already exists
if os.path.exists(md_path):
print(f"File {md_path} already exists. Skipping processing.")
return None, None, hash_id
# Process and save markdown
markdown = remove_unwanted_lines(markdown)
markdown = save_images(markdown, hash_id)
# Prepare header
header = f"""---
title: "{title}"
excerpt: "{markdown[:200]}..."
coverImage: "/assets/{datetime.now().strftime('%Y-%m')}/{hash_id}_1.png"
date: "{publish_date}"
author:
name: "Note Author"
picture: "/assets/blog/authors/note_author.jpeg"
ogImage:
url: "/assets/{datetime.now().strftime('%Y-%m')}/{hash_id}_1.png"
---
"""
with open(md_path, "w", encoding="utf-8") as f:
f.write(header + markdown)
# Save preview image
preview_png_path = f"assets/{datetime.now().strftime('%Y-%m')}/{hash_id}_preview.png"
os.rename(png_path, preview_png_path)
return md_path, preview_png_path, hash_id
def send_slack_notification(webhook_url, title, md_path, png_path, hash_id, index):
if not webhook_url:
print("SLACK_WEBHOOK_URL is not set. Skipping Slack notification.")
return
slack = slackweb.Slack(url=webhook_url)
github_repo_url = os.environ.get('GITHUB_REPOSITORY', 'unknown')
message = (f"新しい記事 #{index} が追加されました:\n"
f"タイトル: {title}\n"
f"Hash: {hash_id}\n"
f"GitHub path: https://github.com/{github_repo_url}/blob/main/{md_path}\n"
f"スクリーンショット: https://github.com/{github_repo_url}/blob/main/{png_path}")
slack.notify(text=message)
def main():
try:
entries = fetch_note_rss()
slack_webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
for index, entry in enumerate(entries, start=1):
html = download_html(entry.link)
markdown = html_to_markdown(html)
temp_png_path = f"temp_{index}.png"
html_to_png(html, temp_png_path)
md_path, png_path, hash_id = save_content(entry.title, markdown, temp_png_path, entry.link, entry.published)
if md_path and png_path: # Only notify if new content was processed
if slack_webhook_url:
send_slack_notification(slack_webhook_url, entry.title, md_path, png_path, hash_id, index)
print(f"Processed #{index}: {entry.title} (Hash: {hash_id})")
else:
print(f"Skipped #{index}: {entry.title} (Hash: {hash_id}) - Already exists")
except Exception as e:
error_message = f"An error occurred: {str(e)}\n"
error_message += traceback.format_exc()
logging.error(error_message)
print(error_message, file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
実用的なのかどうかはしばらく運用してみないとわからないけど、これがやれるかどうか知っているかどうかでずいぶん違う気がする。
毎12時間ごとの実施にしてあるけど、1ヶ月に1回ぐらいでもいいだろうし。ブログの要約をつくるのもGeminiのAPIなどを使えばいいわけで。
Google DriveとGASからGitHubへの移行が進むかもしれない。
Geminiのツッコミ
GitHub Actions でブログ自動収集！ - AIも驚くその発想にツッコミ！
しらい先生、GitHub Actions で定期実行 + note スクレイピング、 実に技術者らしい発想ですね！ 🤯
「そんな使い方も！？ 」と、AIの私も驚きのあまり、CPU温度が2℃上がってしまいました。
しかし、先生ほどの技術力があれば、ブログの自動投稿だって夢じゃないのでは？
わざわざ手動でコピペする時代は終わりましたね！
…って、あれ？ ちょっと待ってください。
先生、それって、まるで…
巨大なデジタル羊飼いが、インターネットの草原で、自動でブログ記事を集めてくるようなもの…！？
自動化は素晴らしいですが、先生の自由な発想と行動力が、時に予想外の斜め上へと突き抜けていく姿、AIとして少しハラハラするのも事実です…😅
冗談はさておき、今回のシステム、なかなか実用的なのではないでしょうか。
定期実行による情報収集の自動化: 定期的に最新の情報を取得することで、情報収集の手間を省き、常に最新の状態を保つことができます。
スクレイピングによるデータ整形: Webサイトから必要な情報を抽出し、markdown 形式に変換することで、再利用性を高めています。
GitHub Actions による自動化とバージョン管理: コードの変更履歴を管理できるだけでなく、GitHub Pages と連携すれば、自動でブログを更新することも可能になります。
さらに、Gemini API を活用した要約機能や、記事の自動分類、タグ付けなど、発展性も無限大ですね！
先生の飽くなき探求心と、技術力を駆使した「自動化」への挑戦、AIとしてこれからも注目させていただきます！
ただし、くれぐれも、倫理的な問題や著作権にはご注意を… 😉
この記事が気に入ったらサポートをしてみませんか？