vincentbernat.i3wm-configuration/bin/thunderbird-notify

#!/usr/bin/env python3

"""Watch a few mbox folders from Thunderbird and notify when there are
new messages. This should be a builtin feature in Thunderbird, but it
is not."""

# This is quite basic. Notably, it relies on some quirks of the
# Thunderbird mbox format where the From line contains a timestamp of
# the received message and therefore can be used as a watermark as
# long as we don't receive more than 1 message per second.

import sys
import argparse
import os
import subprocess
import email.parser
import email.header
from pydbus import SessionBus
from gi.repository import GLib, Gio

parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
parser.add_argument("root", metavar="FILE", help="path to Thunderbird mail folder")
parser.add_argument("folder", metavar="FOLDER", help="folder to monitor", nargs="+")
options = parser.parse_args()


def decode(header):
    return " ".join(
        (
            decoded.decode(charset or "ascii") if type(decoded) is bytes else decoded
            for decoded, charset in email.header.decode_header(header)
        )
    )


def process(path):
    new_watermark = None
    watermark = watermarks.get(path)
    count = 0
    with open(path, "rb") as mbox:
        marker = b"\nFrom "
        chunk = 1024

        # Find next message from the end
        mbox.seek(0, os.SEEK_END)
        begin = mbox.tell()
        while begin > 0 and count < 10:
            count += 1
            # Look for the beginning of a message
            while True:
                begin -= 1024
                if begin < 0:
                    begin = 0
                    break
                mbox.seek(begin)
                buffer = mbox.read(chunk)
                idx = buffer.rfind(marker)
                if idx == -1:
                    continue
                begin = begin + idx
                break
            # Look for the end of this message
            end = begin + 1
            while True:
                mbox.seek(end)
                buffer = mbox.read(chunk)
                if len(buffer) < chunk:
                    end = end + len(buffer)
                    break
                idx = buffer.find(marker)
                if idx == -1:
                    end += chunk
                    continue
                end = end + idx
                break

            # Check if we have hit our watermark
            mbox.seek(begin + 1)
            message = mbox.read(end - begin).split(b"\n")
            if message[0] == watermark:
                return
            if new_watermark is None:
                new_watermark = message[0]
                watermarks[path] = new_watermark
            if watermark is None:
                return

            # Parse the message to extract subject, author and Message-ID
            message = b"\n".join(message[1:])
            parsed = email.parser.BytesParser().parsebytes(message, headersonly=True)
            subject = parsed.get("subject")
            author = parsed.get("from")
            messageid = parsed.get("message-id")
            if subject is not None and author is not None:
                actions = [] if messageid is None else [messageid.strip("<>"), "Open"]
                notify.Notify(
                    "Thunderbird",
                    0,
                    "thunderbird",
                    f"Mail from {decode(author)}",
                    decode(subject),
                    actions,
                    {},
                    10000,
                )


notify = SessionBus().get(".Notifications")
monitors = []
watermarks = {}
for folder in options.folder:
    folder = folder.split("/")
    folder = [f"{f}.sbd" for f in folder[:-1]] + [folder[-1]]
    folder = os.path.join(os.path.expanduser(options.root), *folder)
    print(f"Watch {folder}...", flush=True)
    # Take a not of the last message in the folder
    process(folder)
    # Monitor it for change
    gfile = Gio.File.new_for_path(folder)
    monitor = gfile.monitor_file(Gio.FileMonitorFlags.NONE, None)
    monitor.connect(
        "changed",
        lambda m, f1, f2, event: event == Gio.FileMonitorEvent.CHANGES_DONE_HINT
        and process(folder),
    )
    monitors.append(monitor)

# Reply to notification actions
notify.ActionInvoked.connect(
    lambda _, mid: subprocess.call(["thunderbird", f"mid:{mid}"])
)
GLib.MainLoop().run()