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(float(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                 self.errors = 0
  33                 try:
  34                         if self.ssl:
  35                                 im = imaplib.IMAP4_SSL(self.server, self.port)
  36                         else:
  37                                 im = imaplib.IMAP4(self.server, self.port)
  38                         im.login(self.username, self.password)
  39                 except:
  40                         self.results = "%s (%s)\n" % (self.name, _('Login Error'))
  41                         self.blocker.trigger()
  42                         self.errors += 1
  43                         return
  44
  45                 yield None #let someone else run for a while
  46
  47                 self.results = ""
  48                 self.unseen = 0
  49
  50                 for folder in self.folders:
  51                         folder = folder.strip()
  52
  53                         try:
  54                                 result = im.select(folder, readonly=True)
  55                         except socket.timeout:
  56                                 self.results += "  %s (Timeout)\n" % (folder,)
  57                                 self.errors += 1
  58                                 yield None #let someone else run for a while
  59                                 continue
  60
  61                         if result[0] == 'OK':
  62                                 if result[1][0] == '':
  63                                         count = 0
  64                                 else:
  65                                         count = int(result[1][0])
  66                         else:
  67                                 count = -1
  68                         if count == -1:
  69                                 self.results += "  %s (Error)\n" % (folder,)
  70                                 yield None #let someone else run for a while
  71                                 continue
  72
  73                         result = im.search(None, "UNSEEN")
  74                         if result[0] == 'OK':
  75                                 if result[1][0] == '':
  76                                         unseen = 0
  77                                 else:
  78                                         unseen = len(result[1][0].split())
  79                                         self.unseen += unseen
  80                         else:
  81                                 unseen = -1
  82                                 self.errors += 1
  83                         if unseen > 0:
  84                                 self.results += "  %s (%d/%d)\n" % (folder, unseen, count)
  85                         yield None #let someone else run for a while
  86
  87                 try:
  88                         im.close()
  89                         im.logout()
  90                 except:
  91                         pass
  92
  93                 if len(self.results):
  94                         self.results = "%s:\n%s" % (self.name, self.results)
  95
  96                 # We're done.  Trigger the main task
  97                 self.blocker.trigger()
  98