imap_check.py

   1 """
   2         imap_check.py - An imap folder checker
   3
   4         Copyright 2006 Kenneth Hayber <ken@hayber.us>,
   5         All rights reserved.
   6 """
   7
   8 import imaplib, sys, socket
   9 from checker import Checker
  10
  11 class IMAPChecker(Checker):
  12         def __init__(self, config=None):
  13                 Checker.__init__(self, config)
  14                 self.protocol = 'IMAP'
  15                 try:
  16                         self.server = config['server']
  17                         self.port = int(config['port'])
  18                         self.ssl = (config['ssl'] == 'True')
  19                         self.username = config['username']
  20                         self.password = config['password']
  21                         self.folders = config['folders'].split(',')
  22                 except:
  23                         self.server = 'localhost'
  24                         self.port = 143
  25                         self.ssl = False
  26                         self.username = ''
  27                         self.password = ''
  28                         self.folders = []
  29
  30         def check(self):
  31                 """Check all folders"""
  32                 try:
  33                         if self.ssl:
  34                                 im = imaplib.IMAP4_SSL(self.server, self.port)
  35                         else:
  36                                 im = imaplib.IMAP4(self.server, self.port)
  37                         im.login(self.username, self.password)
  38                 except:
  39                         self.results = "%s (%s)\n" % (self.name, _('Login Error'))
  40                         self.blocker.trigger()
  41                         return
  42
  43                 yield None #let someone else run for a while
  44
  45                 self.results = ""
  46                 self.unseen = 0
  47
  48                 for folder in self.folders:
  49                         folder = folder.strip()
  50
  51                         try:
  52                                 result = im.select(folder, readonly=True)
  53                         except socket.timeout:
  54                                 self.results += "  %s (Timeout)\n" % (folder,)
  55                                 yield None #let someone else run for a while
  56                                 continue
  57
  58                         if result[0] == 'OK':
  59                                 if result[1][0] == '':
  60                                         count = 0
  61                                 else:
  62                                         count = int(result[1][0])
  63                         else:
  64                                 count = -1
  65                         if count == -1:
  66                                 self.results += "  %s (Error)\n" % (folder,)
  67                                 yield None #let someone else run for a while
  68                                 continue
  69
  70                         result = im.search(None, "UNSEEN")
  71                         if result[0] == 'OK':
  72                                 if result[1][0] == '':
  73                                         unseen = 0
  74                                 else:
  75                                         unseen = len(result[1][0].split())
  76                                         self.unseen += unseen
  77                         else:
  78                                 unseen = -1
  79                         if unseen > 0:
  80                                 self.results += "  %s (%d/%d)\n" % (folder, unseen, count)
  81                         yield None #let someone else run for a while
  82
  83                 try:
  84                         im.close()
  85                         im.logout()
  86                 except:
  87                         pass
  88
  89                 if len(self.results):
  90                         self.results = "%s:\n%s" % (self.name, self.results)
  91
  92                 # We're done.  Trigger the main task
  93                 self.blocker.trigger()
  94