Mastodon.py でストリームAPIを使うサンプル

stream_tiny.py

from mastodon import Mastodon, StreamListener
# import logging
# logging.basicConfig(level=logging.DEBUG)

class MyStreamListener(StreamListener):
    def __init__(self):
        super(MyStreamListener, self).__init__()
        # self.logger = logging.getLogger(self.__class__.__name__)
    
    # ここはオーバーライドしなくてもよいが、例外を拾いたいならここで。
    def handle_stream(self, response):
        try:
            super().handle_stream(response)
        except:
            # do something
            raise
    
    # 投稿の度に呼ばれる
    def on_update(self, status):
        # self.logger.info(status_info_string(status))
        pass

    # 削除イベントは投稿のidだけが渡ってくる
    def on_delete(self, status_id):
        # self.logger.info(f"status delete_event: {status_id}")
        pass

    
if __name__ == "__main___":
    # ブラウザのMastodonの設定の開発タブからアプリ登録の作業をして、
    # クライアントキーとクライアントシークレットをそれぞれ１行で書いたファイルと
    # アクセストークンを１行だけ書いたファイルを用意するほうが後から管理しやすい

    # インスタンスへアプリ登録
    Mastodon.create_app("AppNameHere", api_base_url = "https://mstdn-workers.com", to_file = "my_clientcred_workers.txt")
    # そのアプリとしてインスタンスへ接続
    mastodon = Mastodon(client_id="my_clientcred_workers.txt",api_base_url = "https://mstdn-workers.com")
    # 自身のアカウントへアプリ接続
    mastodon.log_in("mail address", "passwd", to_file = "my_usercred_workers.txt")

    # 二回目以降はファイルに保存されたトークンを使う

    #mastodon = Mastodon(
    #    client_id="my_clientcred_workers.txt",
    #    access_token="my_usercred_workers.txt",
    #    api_base_url = "https://mstdn-workers.com"
    #)

    listener = MyStreamListener()
    mastodon.stream_local(listener)