vincentbernat.i3wm-configur.../bin/thunderbird-notify

145 lines
4.8 KiB
Python
Executable file

#!/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")
# https://vincent.bernat.ch/en/x-mozilla-status
status = parsed.get("x-mozilla-status")
status = int(status) & 0x1 if status else 0
if subject is not None and author is not None and status == 0:
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()