Pushbullet + websocketsでリアルタイムに通知を受け取る

前提知識

Pushbulletとは

複数デバイス間（PC・スマートフォン・タブレットetc）で情報を共有するためのアプリ。
Androidを使っている場合、端末間での通知のミラーリングに対応しています。

websocketとは

ざっくり言うとリアルタイム通信のためのプロトコルです。
こちらに詳しいです。

やったこと

Android端末上の通知を、Pushbullet APIを介してPythonスクリプトで取得します。
リアルタイムの通信手段としてwebsocketを使用するので、Pythonのライブラリ「websockets」を利用していきます。

サンプルコード

import asyncio
import websockets
import json
import logging

# デバッグレベルのロガーを追加しておくと何かと助かります。必須ではないです。
logging.basicConfig(
    format="%(asctime)s %(message)s",
    level=logging.DEBUG,
)


async def process(msg):
    event = json.loads(msg)
    # Pushbulletから30秒に1回送信される {"type": "nop"} を無視
    if event.get("type", "nop") != "nop":
        p = event.get("push", {})
        title = p.get("title")
        body = p.get("body")
        app_name = p.get("application_name")
        print(title, body, app_name)


async def __main():
    async for websocket in websockets.connect("wss://stream.pushbullet.com/websocket/<KEY>",
                                              logger=logging.getLogger("websockets.server")):
        try:
            async for message in websocket:
                await process(message)
        except websockets.ConnectionClosed:
            continue


if __name__ == '__main__':
    asyncio.run(__main())

このスクリプトを実行しつつ、Pushbulletアプリで 「Mirroring」→「Send a test notification」 を押下すると、

Test notification If you see this on your computer, notification mirroring is working! Pushbullet

プッシュが受信され、このメッセージが吐き出されるはずです。