245 lines
7 KiB
Python
Executable file
245 lines
7 KiB
Python
Executable file
#!/bin/env python3
|
||
|
||
|
||
import glob
|
||
import json
|
||
import logging
|
||
import os
|
||
import tempfile
|
||
|
||
available_methods = []
|
||
try:
|
||
import gi.repository
|
||
gi.require_version('Notify', '0.7')
|
||
from gi.repository import Notify
|
||
available_methods.append('gnome')
|
||
except:
|
||
pass
|
||
|
||
|
||
|
||
class MailNotify:
|
||
NAME = "Mail Notify"
|
||
METHODS = ['gnome']
|
||
TMPFILE = 'mail-notify.json'
|
||
STATUS_MOD_TIME = 'mod'
|
||
STATUS_GNOME_ID = 'gnome-id'
|
||
MESSAGES = {
|
||
'header': "New e‑mails",
|
||
'body-sgl': "You have {} new e‑mail!",
|
||
'body-plr': "You have {} new e‑mails!"
|
||
}
|
||
|
||
|
||
def __init__(self, maildir, method):
|
||
# Set settings
|
||
self._maildir = os.path.expanduser(maildir)
|
||
if method in MailNotify.METHODS:
|
||
self._method = method
|
||
logging.info("Selected notification method: %s", method)
|
||
else:
|
||
self._method = MailNotify.METHODS[0]
|
||
logging.warning(
|
||
"Notification method “%s” is invalid, falling back to “%s”",
|
||
method, self._method
|
||
)
|
||
self._tmpfile = os.path.join(tempfile.gettempdir(), MailNotify.TMPFILE)
|
||
self._load_status()
|
||
self._is_available = self._init()
|
||
|
||
|
||
def get_maildir(self):
|
||
return self._maildir
|
||
|
||
|
||
def get_method(self):
|
||
return self._method
|
||
|
||
|
||
def get_tmpfile(self):
|
||
return self._tmpfile
|
||
|
||
|
||
def notify(self):
|
||
if not self._is_available:
|
||
logging.warn("Notification method %s is not initialized", self._method)
|
||
return
|
||
mod_time_old = self._load_mod_time()
|
||
mod_time_new = self._read_mod_time()
|
||
self._save_mod_time(mod_time_new)
|
||
if mod_time_new > mod_time_old:
|
||
logging.debug("Mail folder has been modified, proceeding")
|
||
mails_count = self._count_mails()
|
||
if mails_count > 0:
|
||
self._notify(mails_count)
|
||
self._save_status()
|
||
else:
|
||
logging.info("Mail folder is unchanged")
|
||
|
||
|
||
def uninit(self):
|
||
self._uninit()
|
||
|
||
|
||
def _load_status(self):
|
||
# Load application status
|
||
logging.debug("Read status from file %s", self._tmpfile)
|
||
self._status = {}
|
||
if os.path.isfile(self._tmpfile):
|
||
with open(self._tmpfile) as tmpfile:
|
||
self._status = json.load(tmpfile)
|
||
logging.debug("Status loaded: %s", self._status)
|
||
|
||
|
||
def _save_status(self):
|
||
# Save application status
|
||
logging.debug("Save status %s to file %s", self._status, self._tmpfile)
|
||
with open(self._tmpfile, 'w') as tmpfile:
|
||
json.dump(self._status, tmpfile)
|
||
|
||
|
||
def _init(self):
|
||
inited = False
|
||
if self._method == 'gnome':
|
||
inited = self._init_gnome()
|
||
logging.info("Notification method “%s” initialized: %r", self._method, inited)
|
||
return inited
|
||
|
||
|
||
def _init_gnome(self):
|
||
logging.debug("Initializing gnome Notify 0.7")
|
||
if 'gnome' in available_methods:
|
||
return Notify.init(MailNotify.NAME)
|
||
else:
|
||
logging.error("Missing dependency “gi.repository”")
|
||
return False
|
||
|
||
|
||
def _uninit(self):
|
||
if self._method == 'gnome':
|
||
self._uninit_gnome()
|
||
|
||
|
||
def _uninit_gnome(self):
|
||
Notify.uninit()
|
||
|
||
|
||
def _load_mod_time(self):
|
||
if MailNotify.STATUS_MOD_TIME in self._status:
|
||
return self._status[MailNotify.STATUS_MOD_TIME]
|
||
return 0
|
||
|
||
|
||
def _save_mod_time(self, time):
|
||
self._status[MailNotify.STATUS_MOD_TIME] = time
|
||
|
||
|
||
def _read_mod_time(self):
|
||
# read mod time of maildir folder
|
||
self._maildir = os.path.expanduser(self._maildir)
|
||
logging.debug("Read mod time of path %s", self._maildir)
|
||
mod_time = 0
|
||
for path in glob.glob(self._maildir, recursive=True):
|
||
time = os.path.getmtime(path)
|
||
logging.debug("Mod time of path %s: %d", path, mod_time)
|
||
if time > mod_time:
|
||
mod_time = time
|
||
logging.debug("Largest mod time: %s", mod_time)
|
||
return mod_time
|
||
|
||
|
||
def _count_mails(self):
|
||
# Count number of new mails
|
||
logging.debug("Count mails")
|
||
count = 0
|
||
maildir = os.path.expanduser(self._maildir)
|
||
for path in glob.glob(maildir, recursive=True):
|
||
logging.debug("Path: %s", path)
|
||
for file in os.listdir(path):
|
||
logging.debug("File: %s", os.path.join(path, file))
|
||
if os.path.isfile(os.path.join(path, file)):
|
||
count = count + 1
|
||
logging.info("Mails count: %d", count)
|
||
return count
|
||
|
||
|
||
def _notify(self, mails_count):
|
||
# Display notification with selected method
|
||
logging.debug("Notify of %d new mail(s) using method %s", mails_count, self._method)
|
||
if self._method == 'gnome':
|
||
self._notify_gnome(mails_count)
|
||
|
||
|
||
def _notify_gnome(self, mails_count):
|
||
note = Notify.Notification.new(
|
||
MailNotify.MESSAGES['header'],
|
||
self._get_message_body(mails_count).format(mails_count),
|
||
'mail-message-new'
|
||
)
|
||
note.set_category('email.arrived')
|
||
if MailNotify.STATUS_GNOME_ID in self._status:
|
||
note.set_property('id',self._status[MailNotify.STATUS_GNOME_ID])
|
||
note.update(
|
||
MailNotify.MESSAGES['header'],
|
||
self._get_message_body(mails_count).format(mails_count),
|
||
'mail-message-new'
|
||
)
|
||
try:
|
||
note.show()
|
||
self._status[MailNotify.STATUS_GNOME_ID] = note.get_property('id')
|
||
except Exception as e:
|
||
logging.error("Failed to connect to notification daemon: %s", e)
|
||
|
||
|
||
def _get_message_body(self, mails_count):
|
||
if mails_count == 1:
|
||
return MailNotify.MESSAGES['body-sgl']
|
||
return MailNotify.MESSAGES['body-plr']
|
||
|
||
|
||
|
||
|
||
def main():
|
||
# Load modules
|
||
import argparse
|
||
|
||
# Create argument parser
|
||
parser = argparse.ArgumentParser(description="Notify of new e‑mails.")
|
||
parser.add_argument(
|
||
'-v', '--verbose',
|
||
action='count', dest='verbosity',
|
||
default=0,
|
||
help='Verbosity (specify repeatedly to increase verbosity level)'
|
||
)
|
||
parser.add_argument(
|
||
'-n', '--method',
|
||
action='store', dest='method',
|
||
default=MailNotify.METHODS[0],
|
||
help="Notification method, default is “{}”".format(MailNotify.METHODS[0])
|
||
)
|
||
parser.add_argument(
|
||
'-m', '--maildir',
|
||
action='store', dest='maildir',
|
||
default='~/Dokumente/eMails/*/INBOX/new/',
|
||
help="Glob-style pattern of directory to check for new mails, default is ~/Dokumente/eMails/*/INBOX/new/"
|
||
)
|
||
|
||
# Parse arguments
|
||
args = parser.parse_args();
|
||
logging.basicConfig(level=logging.WARNING-(args.verbosity*10))
|
||
|
||
# Create new MailNotify object
|
||
notify = MailNotify(args.maildir, args.method)
|
||
notify.notify()
|
||
notify.uninit()
|
||
|
||
|
||
|
||
|
||
if __name__ == '__main__':
|
||
try:
|
||
main()
|
||
except ModuleNotFoundError as e:
|
||
logging.error("Missing dependency “%s”", e.name)
|
||
exit(1)
|