mirror of
https://github.com/vincentbernat/i3wm-configuration.git
synced 2025-07-31 16:24:25 +02:00
d2a308f472
This should be more robust in case something gets deleted. However, Thunderbird only mark messages for deletion, but this could still happen on compaction.
131 lines
4.4 KiB
Python
Executable file
131 lines
4.4 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."""
|
|
|
|
# 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 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("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 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 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
|
|
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()
|