Lib/mailbox.py

   1 #! /usr/bin/env python
   2
   3 """Classes to handle Unix style, MMDF style, and MH style mailboxes."""
   4
   5
   6 import rfc822
   7 import os
   8
   9 __all__ = ["UnixMailbox","MmdfMailbox","MHMailbox","Maildir","BabylMailbox"]
  10
  11 class _Mailbox:
  12     def __init__(self, fp, factory=rfc822.Message):
  13         self.fp = fp
  14         self.seekp = 0
  15         self.factory = factory
  16
  17     def seek(self, pos, whence=0):
  18         if whence==1:           # Relative to current position
  19             self.pos = self.pos + pos
  20         if whence==2:           # Relative to file's end
  21             self.pos = self.stop + pos
  22         else:                   # Default - absolute position
  23             self.pos = self.start + pos
  24
  25     def next(self):
  26         while 1:
  27             self.fp.seek(self.seekp)
  28             try:
  29                 self._search_start()
  30             except EOFError:
  31                 self.seekp = self.fp.tell()
  32                 return None
  33             start = self.fp.tell()
  34             self._search_end()
  35             self.seekp = stop = self.fp.tell()
  36             if start != stop:
  37                 break
  38         return self.factory(_Subfile(self.fp, start, stop))
  39
  40
  41 class _Subfile:
  42     def __init__(self, fp, start, stop):
  43         self.fp = fp
  44         self.start = start
  45         self.stop = stop
  46         self.pos = self.start
  47
  48     def read(self, length = None):
  49         if self.pos >= self.stop:
  50             return ''
  51         remaining = self.stop - self.pos
  52         if length is None or length < 0:
  53             length = remaining
  54         elif length > remaining:
  55             length = remaining
  56         self.fp.seek(self.pos)
  57         data = self.fp.read(length)
  58         self.pos = self.fp.tell()
  59         return data
  60
  61     def readline(self, length = None):
  62         if self.pos >= self.stop:
  63             return ''
  64         if length is None:
  65             length = self.stop - self.pos
  66         self.fp.seek(self.pos)
  67         data = self.fp.readline(length)
  68         self.pos = self.fp.tell()
  69         return data
  70
  71     def readlines(self, sizehint = -1):
  72         lines = []
  73         while 1:
  74             line = self.readline()
  75             if not line:
  76                 break
  77             lines.append(line)
  78             if sizehint >= 0:
  79                 sizehint = sizehint - len(line)
  80                 if sizehint <= 0:
  81                     break
  82         return lines
  83
  84     def tell(self):
  85         return self.pos - self.start
  86
  87     def seek(self, pos, whence=0):
  88         if whence == 0:
  89             self.pos = self.start + pos
  90         elif whence == 1:
  91             self.pos = self.pos + pos
  92         elif whence == 2:
  93             self.pos = self.stop + pos
  94
  95     def close(self):
  96         del self.fp
  97
  98
  99 class UnixMailbox(_Mailbox):
 100     def _search_start(self):
 101         while 1:
 102             pos = self.fp.tell()
 103             line = self.fp.readline()
 104             if not line:
 105                 raise EOFError
 106             if line[:5] == 'From ' and self._isrealfromline(line):
 107                 self.fp.seek(pos)
 108                 return
 109
 110     def _search_end(self):
 111         self.fp.readline()      # Throw away header line
 112         while 1:
 113             pos = self.fp.tell()
 114             line = self.fp.readline()
 115             if not line:
 116                 return
 117             if line[:5] == 'From ' and self._isrealfromline(line):
 118                 self.fp.seek(pos)
 119                 return
 120
 121     # An overridable mechanism to test for From-line-ness.  You can either
 122     # specify a different regular expression or define a whole new
 123     # _isrealfromline() method.  Note that this only gets called for lines
 124     # starting with the 5 characters "From ".
 125     #
 126     # BAW: According to
 127     #http://home.netscape.com/eng/mozilla/2.0/relnotes/demo/content-length.html
 128     # the only portable, reliable way to find message delimiters in a BSD (i.e
 129     # Unix mailbox) style folder is to search for "\n\nFrom .*\n", or at the
 130     # beginning of the file, "^From .*\n".  While _fromlinepattern below seems
 131     # like a good idea, in practice, there are too many variations for more
 132     # strict parsing of the line to be completely accurate.
 133     #
 134     # _strict_isrealfromline() is the old version which tries to do stricter
 135     # parsing of the From_ line.  _portable_isrealfromline() simply returns
 136     # true, since it's never called if the line doesn't already start with
 137     # "From ".
 138     #
 139     # This algorithm, and the way it interacts with _search_start() and
 140     # _search_end() may not be completely correct, because it doesn't check
 141     # that the two characters preceding "From " are \n\n or the beginning of
 142     # the file.  Fixing this would require a more extensive rewrite than is
 143     # necessary.  For convenience, we've added a StrictUnixMailbox class which
 144     # uses the older, more strict _fromlinepattern regular expression.
 145
 146     _fromlinepattern = r"From \s*[^\s]+\s+\w\w\w\s+\w\w\w\s+\d?\d\s+" \
 147                        r"\d?\d:\d\d(:\d\d)?(\s+[^\s]+)?\s+\d\d\d\d\s*$"
 148     _regexp = None
 149
 150     def _strict_isrealfromline(self, line):
 151         if not self._regexp:
 152             import re
 153             self._regexp = re.compile(self._fromlinepattern)
 154         return self._regexp.match(line)
 155
 156     def _portable_isrealfromline(self, line):
 157         return 1
 158
 159     _isrealfromline = _strict_isrealfromline
 160
 161
 162 class PortableUnixMailbox(UnixMailbox):
 163     _isrealfromline = UnixMailbox._portable_isrealfromline
 164
 165
 166 class MmdfMailbox(_Mailbox):
 167     def _search_start(self):
 168         while 1:
 169             line = self.fp.readline()
 170             if not line:
 171                 raise EOFError
 172             if line[:5] == '\001\001\001\001\n':
 173                 return
 174
 175     def _search_end(self):
 176         while 1:
 177             pos = self.fp.tell()
 178             line = self.fp.readline()
 179             if not line:
 180                 return
 181             if line == '\001\001\001\001\n':
 182                 self.fp.seek(pos)
 183                 return
 184
 185
 186 class MHMailbox:
 187     def __init__(self, dirname, factory=rfc822.Message):
 188         import re
 189         pat = re.compile('^[1-9][0-9]*$')
 190         self.dirname = dirname
 191         # the three following lines could be combined into:
 192         # list = map(long, filter(pat.match, os.listdir(self.dirname)))
 193         list = os.listdir(self.dirname)
 194         list = filter(pat.match, list)
 195         list = map(long, list)
 196         list.sort()
 197         # This only works in Python 1.6 or later;
 198         # before that str() added 'L':
 199         self.boxes = map(str, list)
 200         self.factory = factory
 201
 202     def next(self):
 203         if not self.boxes:
 204             return None
 205         fn = self.boxes[0]
 206         del self.boxes[0]
 207         fp = open(os.path.join(self.dirname, fn))
 208         return self.factory(fp)
 209
 210
 211 class Maildir:
 212     # Qmail directory mailbox
 213
 214     def __init__(self, dirname, factory=rfc822.Message):
 215         self.dirname = dirname
 216         self.factory = factory
 217
 218         # check for new mail
 219         newdir = os.path.join(self.dirname, 'new')
 220         boxes = [os.path.join(newdir, f)
 221                  for f in os.listdir(newdir) if f[0] != '.']
 222
 223         # Now check for current mail in this maildir
 224         curdir = os.path.join(self.dirname, 'cur')
 225         boxes += [os.path.join(curdir, f)
 226                   for f in os.listdir(curdir) if f[0] != '.']
 227
 228         self.boxes = boxes
 229
 230     def next(self):
 231         if not self.boxes:
 232             return None
 233         fn = self.boxes[0]
 234         del self.boxes[0]
 235         fp = open(fn)
 236         return self.factory(fp)
 237
 238
 239 class BabylMailbox(_Mailbox):
 240     def _search_start(self):
 241         while 1:
 242             line = self.fp.readline()
 243             if not line:
 244                 raise EOFError
 245             if line == '*** EOOH ***\n':
 246                 return
 247
 248     def _search_end(self):
 249         while 1:
 250             pos = self.fp.tell()
 251             line = self.fp.readline()
 252             if not line:
 253                 return
 254             if line == '\037\014\n':
 255                 self.fp.seek(pos)
 256                 return
 257
 258
 259 def _test():
 260     import time
 261     import sys
 262     import os
 263
 264     args = sys.argv[1:]
 265     if not args:
 266         for key in 'MAILDIR', 'MAIL', 'LOGNAME', 'USER':
 267             if os.environ.has_key(key):
 268                 mbox = os.environ[key]
 269                 break
 270         else:
 271             print "$MAIL, $LOGNAME nor $USER set -- who are you?"
 272             return
 273     else:
 274         mbox = args[0]
 275     if mbox[:1] == '+':
 276         mbox = os.environ['HOME'] + '/Mail/' + mbox[1:]
 277     elif not '/' in mbox:
 278         mbox = '/usr/mail/' + mbox
 279     if os.path.isdir(mbox):
 280         if os.path.isdir(os.path.join(mbox, 'cur')):
 281             mb = Maildir(mbox)
 282         else:
 283             mb = MHMailbox(mbox)
 284     else:
 285         fp = open(mbox, 'r')
 286         mb = UnixMailbox(fp)
 287
 288     msgs = []
 289     while 1:
 290         msg = mb.next()
 291         if msg is None:
 292             break
 293         msgs.append(msg)
 294         if len(args) <= 1:
 295             msg.fp = None
 296     if len(args) > 1:
 297         num = int(args[1])
 298         print 'Message %d body:'%num
 299         msg = msgs[num-1]
 300         msg.rewindbody()
 301         sys.stdout.write(msg.fp.read())
 302     else:
 303         print 'Mailbox',mbox,'has',len(msgs),'messages:'
 304         for msg in msgs:
 305             f = msg.getheader('from') or ""
 306             s = msg.getheader('subject') or ""
 307             d = msg.getheader('date') or ""
 308             print '-%20.20s   %20.20s   %-30.30s'%(f, d[5:], s)
 309
 310
 311 if __name__ == '__main__':
 312     _test()