Created
September 19, 2015 11:36
-
-
Save apconole/1325438590d5f8a7074b to your computer and use it in GitHub Desktop.
Simple python mail notification utility
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
# Copyright (C) 2015 Aaron Conole <aconole@bytheb.org> | |
# I cobbled this together looking for a really lightweight mail | |
# notification utility to couple with emacs. I never found | |
# a solution - this fits the bill for my needs | |
# | |
# LICENSE: GPLv2 with the caveat that it cannot be upgraded | |
# to GPLv3 | |
import datetime | |
import imaplib2 | |
import socket | |
import os | |
import re | |
import notify2 | |
from threading import Thread | |
from threading import Event | |
import ConfigParser | |
t2 = datetime.datetime.now() | |
class _Configurations(object): | |
def __init__(self): | |
self.config = ConfigParser.ConfigParser() | |
self.config.read(os.getenv("HOME")+"/.mail_notify") | |
def __repr__(self): | |
srepr = "\"mail_notify\": {\n" | |
for ItemName,ItemVal in self.config.items('default'): | |
srepr += "\"%s\": %s\n" % (ItemName, ItemVal) | |
srepr += "}\n" | |
return srepr | |
def FindOptionVal(self, OptionName): | |
for option in self.config.options('default'): | |
if option.lower() == OptionName.lower(): | |
return self.config.get('default', OptionName) | |
return None | |
# This is the threading object that does all the waiting on | |
# the event | |
class _Idler(object): | |
def __init__(self, conn): | |
self.thread = Thread(target=self.idle) | |
self.M = conn | |
self.event = Event() | |
def start(self): | |
self.thread.start() | |
def stop(self): | |
self.event.set() | |
def join(self): | |
self.thread.join() | |
def idle(self): | |
while True: | |
if self.event.isSet(): | |
return | |
self.needsync = False | |
def callback(args): | |
if not self.event.isSet(): | |
self.needsync = True | |
self.event.set() | |
self.M.idle(callback=callback) | |
self.event.wait() | |
if self.needsync: | |
self.event.clear() | |
self.dosync() | |
# The method that gets called when a new email arrives. | |
def dosync(self): | |
n = notify2.Notification("New Mail Arrived", | |
"You've received new mail", | |
"notification-message-im") | |
n.show() | |
def _GetGPGAuthInfoData(filename): | |
binary = "/usr/bin/gpg" | |
if _Configurations().FindOptionVal('gpg-binary') != None: | |
binary = _Configurations().FindOptionVal('gpg-binary') | |
if os.environ.get('GPG_AGENT_INFO') != None: | |
binary += " --use-agent" | |
cmd = "%s -q --no-tty -d %s" % (binary, filename) | |
authinfo = os.popen(cmd).read() | |
return authinfo | |
def _GetAuthInfoData(): | |
filename = os.environ['HOME'] | |
if os.path.isfile(filename + "/.authinfo.gpg"): | |
authinfo = _GetGPGAuthInfoData(filename + "/.authinfo.gpg") | |
elif os.path.isfile(filename + "/.authinfo"): | |
authinfo = open(filename + "/.authinfo").read() | |
elif os.path.isfile(filename + "/.netrc"): | |
authinfo = open(filename + "/.netrc") | |
else: | |
print "No authinfo or netrc file found..." | |
return None | |
return authinfo | |
def _GetImapObject(): | |
cfg = _Configurations() | |
machine = cfg.FindOptionVal('machine') | |
user = cfg.FindOptionVal('login') | |
data = _GetAuthInfoData() | |
if user is None: | |
user = "([^\W]*)" | |
p = re.search("machine %s login %s password ([^\W]*) (port )?([^\W]*)?" % (machine, user), | |
data) | |
if p is None: | |
return None | |
pgroup = 1 | |
if user == "([^\W]*)": | |
user = p.group(pgroup) | |
pgroup += 1 | |
password = p.group(pgroup) | |
port = "143" | |
if p.group(pgroup+1) is not None: | |
port = p.group(pgroup+2) | |
if port.isdigit() == False: | |
port = socket.getservbyname(port, 'tcp') | |
if cfg.FindOptionVal('ssl') is None or cfg.FindOptionVal('ssl').lower() == "true": | |
M = imaplib2.IMAP4_SSL(machine, port) | |
else: | |
M = imaplib2.IMAP4(machine, port) | |
M.login(user, password) | |
M.select("INBOX") | |
return M | |
def _Daemonize(): | |
def dofork(): | |
try: | |
pid = os.fork() | |
except OSError, e: | |
raise Exception, "%s [%d]" % (e.strerror, e.errno) | |
return pid | |
pid = dofork() | |
if pid == 0: | |
os.setsid() | |
pid = dofork() | |
if pid == 0: | |
os.chdir(os.environ['HOME']) | |
os.umask(0200) | |
else: | |
os._exit(0) | |
else: | |
os._exit(0) | |
os.open("/dev/null", os.O_RDWR) | |
os.dup2(0, 1) | |
os.dup2(0, 2) | |
return 0 | |
def main(): | |
_Daemonize() | |
notify2.init('Mail Notifier') | |
M = _GetImapObject() | |
idl = _Idler(M) | |
idl.start() | |
idl.join() | |
if __name__ == '__main__': | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment