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.

It takes a configuration file as first argument written in YAML. It
should look like this::

    root: ~/.thunderbird/something-default/ImapMail/imap.example.com
    folders:
      - INBOX
      - Notifications/GitHub

"""

# 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 time
import yaml
import email.parser
import email.policy
from pydbus import SessionBus
from gi.repository import GLib, Gio

parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
parser.add_argument("config", metavar="CONFIG", help="configuration file", type=open)
options = parser.parse_args()
config = yaml.safe_load(options.config)


def notify_new_messages(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")
            current_date = message[0][len("From - ") :].decode("ascii")
            current_date = time.strptime(current_date, "%a %b %d %H:%M:%S %Y")
            if watermark is not None and current_date <= watermark:
                return
            if new_watermark is None:
                new_watermark = current_date
                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(policy=email.policy.default).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 {author}",
                    subject,
                    actions,
                    {},
                    10000,
                )


notify = SessionBus().get(".Notifications")
monitors = []
watermarks = {}
for folder in config["folders"]:
    folder = folder.split("/")
    folder = [f"{f}.sbd" for f in folder[:-1]] + [folder[-1]]
    folder = os.path.join(os.path.expanduser(config["root"]), *folder)
    print(f"Watch {folder}...", flush=True)
    # Take a not of the last message in the folder
    notify_new_messages(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, folder: event == Gio.FileMonitorEvent.CHANGES_DONE_HINT
        and notify_new_messages(folder),
        folder,
    )
    monitors.append(monitor)

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