mirror of
https://github.com/python/cpython.git
synced 2025-07-23 11:15:24 +00:00
e108a02723
mailbox.py (from the CVS tree) doesn't work with qmail Maildirs: Filenames are completed when the directories are scanned, and the directory name is prepended again in the next() method. Another suggestion: Change the print statement in the _test() driver to show two more date characters (probably the length has increased due to the recent Y2K hype ;). Now it shows the complete date, including the seconds -- at least for me. (I've also made the sender field left justified, in case it is ever shorter than the field width).
282 lines
7.7 KiB
Python
Executable file
282 lines
7.7 KiB
Python
Executable file
#! /usr/bin/env python
|
|
|
|
"""Classes to handle Unix style, MMDF style, and MH style mailboxes."""
|
|
|
|
|
|
import rfc822
|
|
import os
|
|
|
|
class _Mailbox:
|
|
def __init__(self, fp):
|
|
self.fp = fp
|
|
self.seekp = 0
|
|
|
|
def seek(self, pos, whence=0):
|
|
if whence==1: # Relative to current position
|
|
self.pos = self.pos + pos
|
|
if whence==2: # Relative to file's end
|
|
self.pos = self.stop + pos
|
|
else: # Default - absolute position
|
|
self.pos = self.start + pos
|
|
|
|
def next(self):
|
|
while 1:
|
|
self.fp.seek(self.seekp)
|
|
try:
|
|
self._search_start()
|
|
except EOFError:
|
|
self.seekp = self.fp.tell()
|
|
return None
|
|
start = self.fp.tell()
|
|
self._search_end()
|
|
self.seekp = stop = self.fp.tell()
|
|
if start <> stop:
|
|
break
|
|
return rfc822.Message(_Subfile(self.fp, start, stop))
|
|
|
|
|
|
class _Subfile:
|
|
def __init__(self, fp, start, stop):
|
|
self.fp = fp
|
|
self.start = start
|
|
self.stop = stop
|
|
self.pos = self.start
|
|
|
|
def read(self, length = None):
|
|
if self.pos >= self.stop:
|
|
return ''
|
|
remaining = self.stop - self.pos
|
|
if length is None or length < 0:
|
|
length = remaining
|
|
elif length > remaining:
|
|
length = remaining
|
|
self.fp.seek(self.pos)
|
|
data = self.fp.read(length)
|
|
self.pos = self.fp.tell()
|
|
return data
|
|
|
|
def readline(self, length = None):
|
|
if self.pos >= self.stop:
|
|
return ''
|
|
if length is None:
|
|
length = self.stop - self.pos
|
|
self.fp.seek(self.pos)
|
|
data = self.fp.readline(length)
|
|
self.pos = self.fp.tell()
|
|
return data
|
|
|
|
def readlines(self, sizehint = -1):
|
|
lines = []
|
|
while 1:
|
|
line = self.readline()
|
|
if not line:
|
|
break
|
|
lines.append(line)
|
|
if sizehint >= 0:
|
|
sizehint = sizehint - len(line)
|
|
if sizehint <= 0:
|
|
break
|
|
return lines
|
|
|
|
def tell(self):
|
|
return self.pos - self.start
|
|
|
|
def seek(self, pos, whence=0):
|
|
if whence == 0:
|
|
self.pos = self.start + pos
|
|
elif whence == 1:
|
|
self.pos = self.pos + pos
|
|
elif whence == 2:
|
|
self.pos = self.stop + pos
|
|
|
|
def close(self):
|
|
del self.fp
|
|
|
|
|
|
class UnixMailbox(_Mailbox):
|
|
def _search_start(self):
|
|
while 1:
|
|
pos = self.fp.tell()
|
|
line = self.fp.readline()
|
|
if not line:
|
|
raise EOFError
|
|
if line[:5] == 'From ' and self._isrealfromline(line):
|
|
self.fp.seek(pos)
|
|
return
|
|
|
|
def _search_end(self):
|
|
self.fp.readline() # Throw away header line
|
|
while 1:
|
|
pos = self.fp.tell()
|
|
line = self.fp.readline()
|
|
if not line:
|
|
return
|
|
if line[:5] == 'From ' and self._isrealfromline(line):
|
|
self.fp.seek(pos)
|
|
return
|
|
|
|
# An overridable mechanism to test for From-line-ness.
|
|
# You can either specify a different regular expression
|
|
# or define a whole new _isrealfromline() method.
|
|
# Note that this only gets called for lines starting with
|
|
# the 5 characters "From ".
|
|
|
|
_fromlinepattern = r"From \s*[^\s]+\s+\w\w\w\s+\w\w\w\s+\d?\d\s+" \
|
|
r"\d?\d:\d\d(:\d\d)?(\s+[^\s]+)?\s+\d\d\d\d\s*$"
|
|
_regexp = None
|
|
|
|
def _isrealfromline(self, line):
|
|
if not self._regexp:
|
|
import re
|
|
self._regexp = re.compile(self._fromlinepattern)
|
|
return self._regexp.match(line)
|
|
|
|
|
|
class MmdfMailbox(_Mailbox):
|
|
def _search_start(self):
|
|
while 1:
|
|
line = self.fp.readline()
|
|
if not line:
|
|
raise EOFError
|
|
if line[:5] == '\001\001\001\001\n':
|
|
return
|
|
|
|
def _search_end(self):
|
|
while 1:
|
|
pos = self.fp.tell()
|
|
line = self.fp.readline()
|
|
if not line:
|
|
return
|
|
if line == '\001\001\001\001\n':
|
|
self.fp.seek(pos)
|
|
return
|
|
|
|
|
|
class MHMailbox:
|
|
def __init__(self, dirname):
|
|
import re
|
|
pat = re.compile('^[1-9][0-9]*$')
|
|
self.dirname = dirname
|
|
# the three following lines could be combined into:
|
|
# list = map(long, filter(pat.match, os.listdir(self.dirname)))
|
|
list = os.listdir(self.dirname)
|
|
list = filter(pat.match, list)
|
|
list = map(long, list)
|
|
list.sort()
|
|
# This only works in Python 1.6 or later;
|
|
# before that str() added 'L':
|
|
self.boxes = map(str, list)
|
|
|
|
def next(self):
|
|
if not self.boxes:
|
|
return None
|
|
fn = self.boxes[0]
|
|
del self.boxes[0]
|
|
fp = open(os.path.join(self.dirname, fn))
|
|
return rfc822.Message(fp)
|
|
|
|
|
|
class Maildir:
|
|
# Qmail directory mailbox
|
|
|
|
def __init__(self, dirname):
|
|
import string
|
|
self.dirname = dirname
|
|
self.boxes = []
|
|
|
|
# check for new mail
|
|
newdir = os.path.join(self.dirname, 'new')
|
|
for file in os.listdir(newdir):
|
|
if len(string.split(file, '.')) > 2:
|
|
self.boxes.append(os.path.join(newdir, file))
|
|
|
|
# Now check for current mail in this maildir
|
|
curdir = os.path.join(self.dirname, 'cur')
|
|
for file in os.listdir(curdir):
|
|
if len(string.split(file, '.')) > 2:
|
|
self.boxes.append(os.path.join(curdir, file))
|
|
|
|
def next(self):
|
|
if not self.boxes:
|
|
return None
|
|
fn = self.boxes[0]
|
|
del self.boxes[0]
|
|
fp = open(fn)
|
|
return rfc822.Message(fp)
|
|
|
|
|
|
class BabylMailbox(_Mailbox):
|
|
def _search_start(self):
|
|
while 1:
|
|
line = self.fp.readline()
|
|
if not line:
|
|
raise EOFError
|
|
if line == '*** EOOH ***\n':
|
|
return
|
|
|
|
def _search_end(self):
|
|
while 1:
|
|
pos = self.fp.tell()
|
|
line = self.fp.readline()
|
|
if not line:
|
|
return
|
|
if line == '\037\014\n':
|
|
self.fp.seek(pos)
|
|
return
|
|
|
|
|
|
def _test():
|
|
import time
|
|
import sys
|
|
import string
|
|
import os
|
|
|
|
args = sys.argv[1:]
|
|
if not args:
|
|
for key in 'MAILDIR', 'MAIL', 'LOGNAME', 'USER':
|
|
if os.environ.has_key(key):
|
|
mbox = os.environ[key]
|
|
break
|
|
else:
|
|
print "$MAIL, $LOGNAME nor $USER set -- who are you?"
|
|
return
|
|
else:
|
|
mbox = args[0]
|
|
if mbox[:1] == '+':
|
|
mbox = os.environ['HOME'] + '/Mail/' + mbox[1:]
|
|
elif not '/' in mbox:
|
|
mbox = '/usr/mail/' + mbox
|
|
if os.path.isdir(mbox):
|
|
if os.path.isdir(os.path.join(mbox, 'cur')):
|
|
mb = Maildir(mbox)
|
|
else:
|
|
mb = MHMailbox(mbox)
|
|
else:
|
|
fp = open(mbox, 'r')
|
|
mb = UnixMailbox(fp)
|
|
|
|
msgs = []
|
|
while 1:
|
|
msg = mb.next()
|
|
if msg is None:
|
|
break
|
|
msgs.append(msg)
|
|
if len(args) <= 1:
|
|
msg.fp = None
|
|
if len(args) > 1:
|
|
num = string.atoi(args[1])
|
|
print 'Message %d body:'%num
|
|
msg = msgs[num-1]
|
|
msg.rewindbody()
|
|
sys.stdout.write(msg.fp.read())
|
|
else:
|
|
print 'Mailbox',mbox,'has',len(msgs),'messages:'
|
|
for msg in msgs:
|
|
f = msg.getheader('from') or ""
|
|
s = msg.getheader('subject') or ""
|
|
d = msg.getheader('date') or ""
|
|
print '-%20.20s %20.20 %-30.30s'%(f, d[5:], s)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
_test()
|