mirror of
https://github.com/vincentbernat/i3wm-configuration.git
synced 2025-08-04 10:14:34 +02:00
thunderbird: remove thunderbird-notify
Replaced by a custom extension.
This commit is contained in:
parent
e9bcb9ba61
commit
9ed4a7f87c
2 changed files with 0 additions and 150 deletions
|
@ -1,142 +0,0 @@
|
|||
#!/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()
|
Loading…
Add table
Add a link
Reference in a new issue