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     def read(self, length = None):
  47         if self.pos >= self.stop:
  48             return ''
  49         remaining = self.stop - self.pos
  50         if length is None or length < 0:
  51             length = remaining
  52         elif length > remaining:
  53             length = remaining
  54         self.fp.seek(self.pos)
  55         data = self.fp.read(length)
  56         self.pos = self.fp.tell()
  57         return data
  58
  59     def readline(self, length = None):
  60         if self.pos >= self.stop:
  61             return ''
  62         if length is None:
  63             length = self.stop - self.pos
  64         self.fp.seek(self.pos)
  65         data = self.fp.readline(length)
  66         self.pos = self.fp.tell()
  67         return data
  68
  69     def readlines(self, sizehint = -1):
  70         lines = []
  71         while 1:
  72             line = self.readline()
  73             if not line:
  74                 break
  75             lines.append(line)
  76             if sizehint >= 0:
  77                 sizehint = sizehint - len(line)
  78                 if sizehint <= 0:
  79                     break
  80         return lines
  81
  82     def tell(self):
  83         return self.pos - self.start
  84
  85     def seek(self, pos, whence=0):
  86         if whence == 0:
  87             self.pos = self.start + pos
  88         elif whence == 1:
  89             self.pos = self.pos + pos
  90         elif whence == 2:
  91             self.pos = self.stop + pos
  92
  93     def close(self):
  94         del self.fp
  95
  96
  97 # Recommended to use PortableUnixMailbox instead!
  98 class UnixMailbox(_Mailbox):
  99
 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 PortableUnixMailbox class
 144     # which uses the more lenient _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 True
 158
 159     _isrealfromline = _strict_isrealfromline
 160
 161
 162 class PortableUnixMailbox(UnixMailbox):
 163     _isrealfromline = UnixMailbox._portable_isrealfromline
 164
 165
 166 class MmdfMailbox(_Mailbox):
 167
 168     def _search_start(self):
 169         while 1:
 170             line = self.fp.readline()
 171             if not line:
 172                 raise EOFError
 173             if line[:5] == '\001\001\001\001\n':
 174                 return
 175
 176     def _search_end(self):
 177         while 1:
 178             pos = self.fp.tell()
 179             line = self.fp.readline()
 180             if not line:
 181                 return
 182             if line == '\001\001\001\001\n':
 183                 self.fp.seek(pos)
 184                 return
 185
 186
 187 class MHMailbox:
 188
 189     def __init__(self, dirname, factory=rfc822.Message):
 190         import re
 191         pat = re.compile('^[1-9][0-9]*$')
 192         self.dirname = dirname
 193         # the three following lines could be combined into:
 194         # list = map(long, filter(pat.match, os.listdir(self.dirname)))
 195         list = os.listdir(self.dirname)
 196         list = filter(pat.match, list)
 197         list = map(long, list)
 198         list.sort()
 199         # This only works in Python 1.6 or later;
 200         # before that str() added 'L':
 201         self.boxes = map(str, list)
 202         self.factory = factory
 203
 204     def __iter__(self):
 205         return iter(self.next, None)
 206
 207     def next(self):
 208         if not self.boxes:
 209             return None
 210         fn = self.boxes.pop(0)
 211         fp = open(os.path.join(self.dirname, fn))
 212         msg = self.factory(fp)
 213         try:
 214             msg._mh_msgno = fn
 215         except (AttributeError, TypeError):
 216             pass
 217         return msg
 218
 219
 220 class Maildir:
 221     # Qmail directory mailbox
 222
 223     def __init__(self, dirname, factory=rfc822.Message):
 224         self.dirname = dirname
 225         self.factory = factory
 226
 227         # check for new mail
 228         newdir = os.path.join(self.dirname, 'new')
 229         boxes = [os.path.join(newdir, f)
 230                  for f in os.listdir(newdir) if f[0] != '.']
 231
 232         # Now check for current mail in this maildir
 233         curdir = os.path.join(self.dirname, 'cur')
 234         boxes += [os.path.join(curdir, f)
 235                   for f in os.listdir(curdir) if f[0] != '.']
 236
 237         self.boxes = boxes
 238
 239     def __iter__(self):
 240         return iter(self.next, None)
 241
 242     def next(self):
 243         if not self.boxes:
 244             return None
 245         fn = self.boxes.pop(0)
 246         fp = open(fn)
 247         return self.factory(fp)
 248
 249
 250 class BabylMailbox(_Mailbox):
 251
 252     def _search_start(self):
 253         while 1:
 254             line = self.fp.readline()
 255             if not line:
 256                 raise EOFError
 257             if line == '*** EOOH ***\n':
 258                 return
 259
 260     def _search_end(self):
 261         while 1:
 262             pos = self.fp.tell()
 263             line = self.fp.readline()
 264             if not line:
 265                 return
 266             if line == '\037\014\n':
 267                 self.fp.seek(pos)
 268                 return
 269
 270
 271 def _test():
 272     import sys
 273
 274     args = sys.argv[1:]
 275     if not args:
 276         for key in 'MAILDIR', 'MAIL', 'LOGNAME', 'USER':
 277             if key in os.environ:
 278                 mbox = os.environ[key]
 279                 break
 280         else:
 281             print "$MAIL, $LOGNAME nor $USER set -- who are you?"
 282             return
 283     else:
 284         mbox = args[0]
 285     if mbox[:1] == '+':
 286         mbox = os.environ['HOME'] + '/Mail/' + mbox[1:]
 287     elif not '/' in mbox:
 288         if os.path.isfile('/var/mail/' + mbox):
 289             mbox = '/var/mail/' + mbox
 290         else:
 291             mbox = '/usr/mail/' + mbox
 292     if os.path.isdir(mbox):
 293         if os.path.isdir(os.path.join(mbox, 'cur')):
 294             mb = Maildir(mbox)
 295         else:
 296             mb = MHMailbox(mbox)
 297     else:
 298         fp = open(mbox, 'r')
 299         mb = PortableUnixMailbox(fp)
 300
 301     msgs = []
 302     while 1:
 303         msg = mb.next()
 304         if msg is None:
 305             break
 306         msgs.append(msg)
 307         if len(args) <= 1:
 308             msg.fp = None
 309     if len(args) > 1:
 310         num = int(args[1])
 311         print 'Message %d body:'%num
 312         msg = msgs[num-1]
 313         msg.rewindbody()
 314         sys.stdout.write(msg.fp.read())
 315     else:
 316         print 'Mailbox',mbox,'has',len(msgs),'messages:'
 317         for msg in msgs:
 318             f = msg.getheader('from') or ""
 319             s = msg.getheader('subject') or ""
 320             d = msg.getheader('date') or ""
 321             print '-%20.20s   %20.20s   %-30.30s'%(f, d[5:], s)
 322
 323
 324 if __name__ == '__main__':
 325     _test()