見出し画像

AT Protocol を利用して Bluesky にポストしたお知らせを JSON 形式で取得する in Python

cultivationdata.net

1. 概要

このスクリプトは、AT Protocol を通して Bluesky のアカウントから「#お知らせ」タグを含む投稿を定期的に取得し、JSON 形式で保存する Python スクリプトです。

自サイトでも、この JSON ファイルを元に Web API 化してお知らせに使っています。

今回取得しているデータは RSS からも取得可能ですが、AT Protocol であれば必要に応じて他のデータも追加できます（画像URLやリンクなど）。

2. 主な機能

  • Bluesky アカウントへのログインと投稿（feed）取得

  • 指定した日数範囲内の「#お知らせ」投稿のフィルタリング

  • 取得した投稿の JSON 形式での保存

  • エラーハンドリングとログ記録

  • 定期的な実行

3. 設定パラメータ

  • USERNAME: Bluesky のユーザー名（例: cultivationdata.net）

  • PASSWORD: 環境変数から読み込むパスワード

  • CHECK_INTERVAL: チェック間隔（900秒 = 15分）

  • NOTICE_COUNT: 取得するお知らせの最大数（50件）

  • DAYS_RANGE: 取得する投稿の日付範囲（90日）

  • JSON_FILE_PATH: 保存先 JSON ファイルのパス（/var/data/notices/notice.json）

4. 主要な関数

is_within_date_range()

  • 投稿が指定された日数範囲内（現在から90日以内）かを判定

  • UTC 日時文字列を受け取り、boolean 値を返す

get_bluesky_client()

  • Bluesky API クライアントの初期化とログイン処理

  • ログイン失敗時は BlueskyNoticeError を発生

save_to_json()

  • 取得したデータを JSON 形式でファイルに保存

  • 保存先ディレクトリが存在しない場合は作成

bsky_get_notice()

  • メイン処理を行う関数

  • 以下の処理を15分間隔で繰り返し実行：

    1. Bluesky にログイン

    2. ユーザーの投稿を取得（最大50件）

    3. #お知らせ タグを含む投稿をフィルタリング

    4. 90日以内の投稿のみを抽出

    5. 投稿テキスト、URL、日時を JSON として保存

5. エラーハンドリング

  • カスタム例外クラス BlueskyNoticeError を定義

  • ログイン失敗、ファイル操作エラーなどを個別に処理

  • すべてのエラーをログに記録

  • エラー発生時も処理を継続（15分後に再試行）

6. ログ機能

  • INFO/ERROR レベルのログを記録

  • 標準出力とファイル（/var/log/bluesky/notice.log）の両方に出力

  • 主要な処理状況とエラーをログに記録

7. セキュリティ

  • パスワードは環境変数から読み込み

  • 環境変数が未設定の場合はエラーを発生

8. 使用方法

  1. 環境変数 BLUESKY_PASSWORD にパスワードを設定

  2. スクリプトを実行

  3. Ctrl+C で終了するまで継続実行

9. 依存ライブラリ

  • atproto: Bluesky API 操作

  • python-dotenv: 環境変数の読み込み

  • pytz: タイムゾーン処理

10. データ形式

保存される JSON の構造：

{
    "0": {
        "text": "投稿本文",
        "url": "https://bsky.app/profile/username/post/xxxx",
        "datetime": "2024-10-23T08:37:28.402Z"
    },
    ...
}

11. コード全文

from atproto import Client, IdResolver, models
from atproto.exceptions import AtProtocolError
import json
import time
import logging
from pathlib import Path
from typing import Dict, Optional
from datetime import datetime, timedelta
import pytz

from dotenv import load_dotenv
import os

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('/var/log/bluesky/notice.log', encoding='utf-8')
    ]
)

# 環境変数の読み込み
load_dotenv()

# 定数の定義
USERNAME = 'cultivationdata.net'
PASSWORD = os.getenv("BLUESKY_PASSWORD")
if not PASSWORD:
    raise ValueError("環境変数 BLUESKY_PASSWORD が設定されていません")

CHECK_INTERVAL = 900  # 15分
NOTICE_COUNT = 50  # 取得するお知らせの数
DAYS_RANGE = 90  # 今日から何日前までのお知らせを取得するか
JSON_FILE_PATH = Path('/var/data/notices/notice.json')

def is_within_date_range(utc_datetime_str: str) -> bool:
    """
    投稿日が指定された日数範囲内かどうかを判定します。

    Args:
        utc_datetime_str (str): "2024-10-23T08:37:28.402Z" 形式のUTC日時文字列

    Returns:
        bool: 日付範囲内の場合はTrue、それ以外はFalse
    """
    # 現在のUTC日時を取得
    current_date = datetime.now(pytz.UTC).date()
    
    # UTC日時文字列をdatetimeオブジェクトに変換
    post_date = datetime.fromisoformat(utc_datetime_str.replace('Z', '+00:00')).date()
    
    # 日付範囲を計算
    date_threshold = current_date - timedelta(days=DAYS_RANGE)
    
    return post_date >= date_threshold

class BlueskyNoticeError(Exception):
    """Blueskyお知らせ取得に関する基本例外クラス"""
    pass

def ensure_directory_exists(file_path: Path) -> None:
    """
    指定されたファイルパスのディレクトリが存在することを確認し、
    存在しない場合は作成します。

    Args:
        file_path (Path): 確認するファイルパス
    """
    file_path.parent.mkdir(parents=True, exist_ok=True)

def get_bluesky_client() -> Client:
    """
    Bluesky APIクライアントの初期化とログインを行います。

    Returns:
        Client: ログイン済みのBluesky APIクライアント

    Raises:
        BlueskyNoticeError: ログインに失敗した場合
    """
    try:
        client = Client()
        client.login(USERNAME, PASSWORD)
        return client
    except AtProtocolError as e:
        raise BlueskyNoticeError(f"Blueskyへのログインに失敗しました: {str(e)}")

def save_to_json(data: Dict, file_path: Path) -> None:
    """
    データをJSONファイルとして保存します。

    Args:
        data (Dict): 保存するデータ
        file_path (Path): 保存先のファイルパス

    Raises:
        BlueskyNoticeError: ファイルの書き込みに失敗した場合
    """
    try:
        ensure_directory_exists(file_path)
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=4)
        logging.info(f"JSONファイルを保存しました: {file_path}")
    except IOError as e:
        raise BlueskyNoticeError(f"JSONファイルの書き込みに失敗しました: {str(e)}")

def bsky_get_notice() -> None:
    """
    Blueskyから '#お知らせ' を含む投稿を取得し、JSONファイルとして保存します。
    エラーが発生した場合は、ログに記録して一定時間後に再試行します。
    """
    while True:
        try:
            client = get_bluesky_client()
            
            # フィードの取得
            feed_data = client.get_author_feed(
                actor=USERNAME,
                limit=50  # より多くの投稿を取得して #お知らせ をフィルタリング
            )

            posts = [post.post for post in feed_data.feed]
            notice_json: Dict[str, Dict[str, str]] = {}
            count = 0

            # お知らせ投稿の抽出
            for post in posts:
                if '#お知らせ' in post.record.text:                    
                    # 日付範囲内かチェック
                    if not is_within_date_range(post.record.created_at):
                        continue
                        
                    notice_json[str(count)] = {
                        "text": post.record.text,
                        "url": f"https://bsky.app/profile/{USERNAME}/post/{post.uri.split('app.bsky.feed.post/')[1]}",
                        "datetime": post.record.created_at
                    }
                    
                    count += 1
                    logging.info(f"お知らせを取得しました: {post.record.text[:50]}...")
                    
                    if count == NOTICE_COUNT:
                        break

            # JSONファイルの保存
            save_to_json(notice_json, JSON_FILE_PATH)
            
        except BlueskyNoticeError as e:
            logging.error(f"エラーが発生しました: {str(e)}")
        except Exception as e:
            logging.error(f"予期しないエラーが発生しました: {str(e)}")
        
        logging.info(f"{CHECK_INTERVAL}秒後に再度チェックを行います")
        time.sleep(CHECK_INTERVAL)

if __name__ == '__main__':
    try:
        logging.info("Blueskyお知らせ取得スクリプトを開始します")
        bsky_get_notice()
    except KeyboardInterrupt:
        logging.info("スクリプトを終了します")

いいなと思ったら応援しよう！