Mirror PushBullet notifications

pbnotify.py

import json
import time
from threading import Thread
import websocket
from base64 import b64decode
import subprocess
import os
from tempfile import TemporaryDirectory


WEBSOCKET_URL = "wss://stream.pushbullet.com/websocket/"

def on_push(push):
    if push["type"] == "push" and push["push"]["type"] == "mirror":
        title = push["push"]["title"]
        body = push["push"]["body"]
        icon_data = push["push"]["icon"]
        with TemporaryDirectory() as tmpdir:
            icon_path = os.path.join(tmpdir, "img")
            with open(icon_path, "wb") as image:
                image.write(b64decode(icon_data))
            subprocess.Popen(["dunstify", title, body, "-i", icon_path, "-r", push["push"]["notification_id"]])
            time.sleep(0.5)
    elif push["type"] == "push" and push["push"]["type"] == "dismissal":
        subprocess.Popen(["dunstify", "-C", push["push"]["notification_id"]])

class Listener(Thread, websocket.WebSocketApp):
    def __init__(self, key):

        Thread.__init__(self)
        websocket.WebSocketApp.__init__(
                self,
                WEBSOCKET_URL + key,
                on_message=self._on_message(),
        )

    def _on_message(self):
        def callback(*args):
            message = args[1] if len(args) > 1 else args[0]
            try:
                json_message = json.loads(message)
                if json_message["type"] != "nop":
                    on_push(json_message)
            except Exception:
                pass
        return callback

    def run_forever(self):
        websocket.WebSocketApp.run_forever(self)

if __name__ == "__main__":
    token = os.getenv('PB_TOKEN')
    listener = Listener(token)
    try:
        listener.run_forever()
    except KeyboardInterrupt:
        listener.close()