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