mirror of
https://github.com/python/cpython.git
synced 2024-11-24 18:34:43 +08:00
1571a1e34b
Unix From lines, change the UnixMailbox class so that _search_start() positions the file *before* the Unix From line instead of after it; change _search_end() to skip one line before looking for the next From line. The rfc822.Message class automatically recognizes these Unix From lines and squirrels them away in the 'unixfrom' instance variable.
280 lines
9.5 KiB
Python
Executable File
280 lines
9.5 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('^[0-9][0-9]*$')
|
|
self.dirname = dirname
|
|
files = os.listdir(self.dirname)
|
|
self.boxes = []
|
|
for f in files:
|
|
if pat.match(f):
|
|
self.boxes.append(f)
|
|
|
|
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(os.path.join(self.dirname, 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 %18.18s %-30.30s'%(f, d[5:], s)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
_test()
|