masq/i3pystatus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mailchecker ; didn't really test it yet

Browse Source
This commit is contained in:
enkore 2013-02-15 21:14:39 +01:00
parent c4e80b3e40
commit 7e2c61c21a
1 changed files with 34 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
i3pystatus/mailchecker.py
View File

@ -6,9 +6,9 @@ import json
from datetime import datetime,timedelta
import imaplib
from i3pystatus import Module
from i3pystatus import IntervalModule
class MailChecker(Module):
class MailChecker(IntervalModule):
"""
This class handles mailservers and outputs i3status compatible
json data for the accumulated unread count. The mail server
@ -16,8 +16,8 @@ class MailChecker(Module):
"""
settings = {
'color': '#ff0000',
'servers': []
"color": "#ff0000",
"servers": []
}
servers = []
@ -25,29 +25,29 @@ class MailChecker(Module):
def __init__(self, settings = None):
self.settings.update(settings)
for server in settings['servers']:
for server in settings["servers"]:
srv = MailChecker.MailServer(server)
self.servers.append(srv)
def output(self):
unread = 0
for srv in self.servers:
unread += srv.get_unread_count()
def run(self):
unread = sum([server.get_unread_count() for server in self.servers])
if not unread:
return None
return {'full_text' : '%d new email%s' % (unread, ('s' if unread > 1 else '')),
'name' : 'newmail',
'urgent' : 'true',
'color' : self.settings['color']}
self.output = {
"full_text" : "%d new email%s" % (unread, ("s" if unread > 1 else "")),
"name" : "newmail",
"urgent" : "true",
"color" : self.settings["color"]
}
class MailServer:
"""
This class provides the functionality to connect
to a mail server and fetch the count of unread emails.
When the server connection is lost, it returns 0 and
tries to reconnect. It checks every 'pause' seconds.
tries to reconnect. It checks every "pause" seconds.
"""
host = ""
@ -56,52 +56,36 @@ class MailChecker(Module):
username = ""
password = ""
connection = None
pause = 30
unread_cache = 0
last_checked = datetime.now()
def __init__(self, settings_dict):
self.host = settings_dict['host']
self.port = settings_dict['port']
self.username = settings_dict['username']
self.password = settings_dict['password']
self.pause = settings_dict['pause']
self.host = settings_dict["host"]
self.port = settings_dict["port"]
self.username = settings_dict["username"]
self.password = settings_dict["password"]
if settings_dict['ssl']:
if settings_dict["ssl"]:
self.imap_class = imaplib.IMAP4_SSL
self.last_checked = \
datetime.now() - timedelta(seconds=self.pause)
def get_connection(self):
if not has_internet_connection():
self.connection = None
else:
if not self.connection:
try:
self.connection = self.imap_class(self.host, self.port)
self.connection.login(self.username, self.password)
self.connection.select()
except Exception:
self.connection = None
if not self.connection:
try:
self.connection = self.imap_class(self.host, self.port)
self.connection.login(self.username, self.password)
self.connection.select()
except Exception as e:
except Exception:
self.connection = None
try:
self.connection.select()
except Exception as e:
self.connection = None
return self.connection
def get_unread_count(self):
delta = datetime.now() - self.last_checked
if delta.total_seconds() > self.pause:
unread = 0
conn = self.get_connection()
if conn:
unread += len(conn.search(None,'UnSeen')[1][0].split())
self.unread_cache = unread
self.last_checked = datetime.now()
return self.unread_cache
unread = 0
conn = self.get_connection()
if conn:
unread += len(conn.search(None,"UnSeen")[1][0].split())
return unread