[Stackless-checkins] CVS: slpdev/src/2.2/src/Lib/email FeedParser.py, NONE, 1.1 _parseaddr.py, NONE, 1.1
Christian Tismer
tismer at centera.de
Sat May 1 02:54:07 CEST 2004
- Previous message: [Stackless-checkins] CVS: slpdev/src/2.2/src/Lib/hotshot stones.py, NONE, 1.1
- Next message: [Stackless-checkins] CVS: slpdev/src/2.2/src/Lib/distutils/command checkdep.py, NONE, 1.1 register.py, NONE, 1.1 wininst-6.exe, NONE, 1.1 wininst-7.1.exe, NONE, 1.1
- Messages sorted by:
[ date ]
[ thread ]
[ subject ]
[ author ]
Update of /home/cvs/slpdev/src/2.2/src/Lib/email
In directory centera.de:/home/tismer/slpdev/src/2.2/src/Lib/email
Added Files:
FeedParser.py _parseaddr.py
Log Message:
added files
--- NEW FILE: FeedParser.py ---
# A new Feed-style Parser
from email import Errors, Message
import re
NLCRE = re.compile('\r\n|\r|\n')
EMPTYSTRING = ''
NL = '\n'
NeedMoreData = object()
class FeedableLumpOfText:
"A file-like object that can have new data loaded into it"
def __init__(self):
self._partial = ''
self._done = False
# _pending is a list of lines, in reverse order
self._pending = []
def readline(self):
""" Return a line of data.
If data has been pushed back with unreadline(), the most recently
returned unreadline()d data will be returned.
"""
if not self._pending:
if self._done:
return ''
return NeedMoreData
return self._pending.pop()
def unreadline(self, line):
""" Push a line back into the object.
"""
self._pending.append(line)
def peekline(self):
""" Non-destructively look at the next line """
if not self._pending:
if self._done:
return ''
return NeedMoreData
return self._pending[-1]
# for r in self._input.readuntil(regexp):
# if r is NeedMoreData:
# yield NeedMoreData
# preamble, matchobj = r
def readuntil(self, matchre, afterblank=False, includematch=False):
""" Read a line at a time until we get the specified RE.
Returns the text up to (and including, if includematch is true) the
matched text, and the RE match object. If afterblank is true,
there must be a blank line before the matched text. Moves current
filepointer to the line following the matched line. If we reach
end-of-file, return what we've got so far, and return None as the
RE match object.
"""
prematch = []
blankseen = 0
while 1:
if not self._pending:
if self._done:
# end of file
yield EMPTYSTRING.join(prematch), None
else:
yield NeedMoreData
continue
line = self._pending.pop()
if afterblank:
if NLCRE.match(line):
blankseen = 1
continue
else:
blankseen = 0
m = matchre.match(line)
if (m and not afterblank) or (m and afterblank and blankseen):
if includematch:
prematch.append(line)
yield EMPTYSTRING.join(prematch), m
prematch.append(line)
NLatend = re.compile('(\r\n|\r|\n)$').match
NLCRE_crack = re.compile('(\r\n|\r|\n)')
def push(self, data):
""" Push some new data into this object """
# Handle any previous leftovers
data, self._partial = self._partial+data, ''
# Crack into lines, but leave the newlines on the end of each
lines = self.NLCRE_crack.split(data)
# The *ahem* interesting behaviour of re.split when supplied
# groups means that the last element is the data after the
# final RE. In the case of a NL/CR terminated string, this is
# the empty string.
self._partial = lines.pop()
o = []
for i in range(len(lines) / 2):
o.append(EMPTYSTRING.join([lines[i*2], lines[i*2+1]]))
self.pushlines(o)
def pushlines(self, lines):
""" Push a list of new lines into the object """
# Reverse and insert at the front of _pending
self._pending[:0] = lines[::-1]
def end(self):
""" There is no more data """
self._done = True
def is_done(self):
return self._done
def __iter__(self):
return self
def next(self):
l = self.readline()
if l == '':
raise StopIteration
return l
class FeedParser:
"A feed-style parser of email. copy docstring here"
def __init__(self, _class=Message.Message):
"fnord fnord fnord"
self._class = _class
self._input = FeedableLumpOfText()
self._root = None
self._objectstack = []
self._parse = self._parsegen().next
def end(self):
self._input.end()
self._call_parse()
return self._root
def feed(self, data):
self._input.push(data)
self._call_parse()
def _call_parse(self):
try:
self._parse()
except StopIteration:
pass
headerRE = re.compile(r'^(From |[-\w]{2,}:|[\t ])')
def _parse_headers(self,headerlist):
# Passed a list of strings that are the headers for the
# current object
lastheader = ''
lastvalue = []
for lineno, line in enumerate(headerlist):
# Check for continuation
if line[0] in ' \t':
if not lastheader:
raise Errors.HeaderParseError('First line must not be a continuation')
lastvalue.append(line)
continue
if lastheader:
# XXX reconsider the joining of folded lines
self._cur[lastheader] = EMPTYSTRING.join(lastvalue).rstrip()
lastheader, lastvalue = '', []
# Check for Unix-From
if line.startswith('From '):
if lineno == 0:
self._cur.set_unixfrom(line)
continue
elif lineno == len(headerlist) - 1:
# Something looking like a unix-from at the end - it's
# probably the first line of the body
self._input.unreadline(line)
return
else:
# Weirdly placed unix-from line. Ignore it.
continue
i = line.find(':')
if i < 0:
# The older parser had various special-cases here. We've
# already handled them
raise Errors.HeaderParseError(
"Not a header, not a continuation: ``%s''" % line)
lastheader = line[:i]
lastvalue = [line[i+1:].lstrip()]
if lastheader:
# XXX reconsider the joining of folded lines
self._cur[lastheader] = EMPTYSTRING.join(lastvalue).rstrip()
def _parsegen(self):
# Parse any currently available text
self._new_sub_object()
self._root = self._cur
completing = False
last = None
for line in self._input:
if line is NeedMoreData:
yield None # Need More Data
continue
self._input.unreadline(line)
if not completing:
headers = []
# Now collect all headers.
for line in self._input:
if line is NeedMoreData:
yield None # Need More Data
continue
if not self.headerRE.match(line):
self._parse_headers(headers)
# A message/rfc822 has no body and no internal
# boundary.
if self._cur.get_content_maintype() == "message":
self._new_sub_object()
completing = False
headers = []
continue
if line.strip():
# No blank line between headers and body.
# Push this line back, it's the first line of
# the body.
self._input.unreadline(line)
break
else:
headers.append(line)
else:
# We're done with the data and are still inside the headers
self._parse_headers(headers)
# Now we're dealing with the body
boundary = self._cur.get_boundary()
isdigest = (self._cur.get_content_type() == 'multipart/digest')
if boundary and not self._cur._finishing:
separator = '--' + boundary
self._cur._boundaryRE = re.compile(
r'(?P<sep>' + re.escape(separator) +
r')(?P<end>--)?(?P<ws>[ \t]*)(?P<linesep>\r\n|\r|\n)$')
for r in self._input.readuntil(self._cur._boundaryRE):
if r is NeedMoreData:
yield NeedMoreData
else:
preamble, matchobj = r
break
if not matchobj:
# Broken - we hit the end of file. Just set the body
# to the text.
if completing:
self._attach_trailer(last, preamble)
else:
self._attach_preamble(self._cur, preamble)
# XXX move back to the parent container.
self._pop_container()
completing = True
continue
if preamble:
if completing:
preamble = preamble[:-len(matchobj.group('linesep'))]
self._attach_trailer(last, preamble)
else:
self._attach_preamble(self._cur, preamble)
elif not completing:
# The module docs specify an empty preamble is None, not ''
self._cur.preamble = None
# If we _are_ completing, the last object gets no payload
if matchobj.group('end'):
# That was the end boundary tag. Bounce back to the
# parent container
last = self._pop_container()
self._input.unreadline(matchobj.group('linesep'))
completing = True
continue
# A number of MTAs produced by a nameless large company
# we shall call "SicroMoft" produce repeated boundary
# lines.
while True:
line = self._input.peekline()
if line is NeedMoreData:
yield None
continue
if self._cur._boundaryRE.match(line):
self._input.readline()
else:
break
self._new_sub_object()
completing = False
if isdigest:
self._cur.set_default_type('message/rfc822')
continue
else:
# non-multipart or after end-boundary
if last is not self._root:
last = self._pop_container()
if self._cur.get_content_maintype() == "message":
# We double-pop to leave the RFC822 object
self._pop_container()
completing = True
elif self._cur._boundaryRE and last <> self._root:
completing = True
else:
# Non-multipart top level, or in the trailer of the
# top level multipart
while not self._input.is_done():
yield None
data = list(self._input)
body = EMPTYSTRING.join(data)
self._attach_trailer(last, body)
def _attach_trailer(self, obj, trailer):
#import pdb ; pdb.set_trace()
if obj.get_content_maintype() in ( "multipart", "message" ):
obj.epilogue = trailer
else:
obj.set_payload(trailer)
def _attach_preamble(self, obj, trailer):
if obj.get_content_maintype() in ( "multipart", "message" ):
obj.preamble = trailer
else:
obj.set_payload(trailer)
def _new_sub_object(self):
new = self._class()
#print "pushing", self._objectstack, repr(new)
if self._objectstack:
self._objectstack[-1].attach(new)
self._objectstack.append(new)
new._boundaryRE = None
new._finishing = False
self._cur = new
def _pop_container(self):
# Move the pointer to the container of the current object.
# Returns the (old) current object
#import pdb ; pdb.set_trace()
#print "popping", self._objectstack
last = self._objectstack.pop()
if self._objectstack:
self._cur = self._objectstack[-1]
else:
self._cur._finishing = True
return last
--- NEW FILE: _parseaddr.py ---
# Copyright (C) 2002 Python Software Foundation
"""Email address parsing code.
Lifted directly from rfc822.py. This should eventually be rewritten.
"""
import time
from types import TupleType
try:
True, False
except NameError:
True = 1
False = 0
SPACE = ' '
EMPTYSTRING = ''
COMMASPACE = ', '
# Parse a date field
_monthnames = ['jan', 'feb', 'mar', 'apr', 'may', 'jun', 'jul',
'aug', 'sep', 'oct', 'nov', 'dec',
'january', 'february', 'march', 'april', 'may', 'june', 'july',
'august', 'september', 'october', 'november', 'december']
_daynames = ['mon', 'tue', 'wed', 'thu', 'fri', 'sat', 'sun']
# The timezone table does not include the military time zones defined
# in RFC822, other than Z. According to RFC1123, the description in
# RFC822 gets the signs wrong, so we can't rely on any such time
# zones. RFC1123 recommends that numeric timezone indicators be used
# instead of timezone names.
_timezones = {'UT':0, 'UTC':0, 'GMT':0, 'Z':0,
'AST': -400, 'ADT': -300, # Atlantic (used in Canada)
'EST': -500, 'EDT': -400, # Eastern
'CST': -600, 'CDT': -500, # Central
'MST': -700, 'MDT': -600, # Mountain
'PST': -800, 'PDT': -700 # Pacific
}
def parsedate_tz(data):
"""Convert a date string to a time tuple.
Accounts for military timezones.
"""
data = data.split()
# The FWS after the comma after the day-of-week is optional, so search and
# adjust for this.
if data[0].endswith(',') or data[0].lower() in _daynames:
# There's a dayname here. Skip it
del data[0]
else:
i = data[0].rfind(',')
if i >= 0:
data[0] = data[0][i+1:]
if len(data) == 3: # RFC 850 date, deprecated
stuff = data[0].split('-')
if len(stuff) == 3:
data = stuff + data[1:]
if len(data) == 4:
s = data[3]
i = s.find('+')
if i > 0:
data[3:] = [s[:i], s[i+1:]]
else:
data.append('') # Dummy tz
if len(data) < 5:
return None
data = data[:5]
[dd, mm, yy, tm, tz] = data
mm = mm.lower()
if mm not in _monthnames:
dd, mm = mm, dd.lower()
if mm not in _monthnames:
return None
mm = _monthnames.index(mm) + 1
if mm > 12:
mm -= 12
if dd[-1] == ',':
dd = dd[:-1]
i = yy.find(':')
if i > 0:
yy, tm = tm, yy
if yy[-1] == ',':
yy = yy[:-1]
if not yy[0].isdigit():
yy, tz = tz, yy
if tm[-1] == ',':
tm = tm[:-1]
tm = tm.split(':')
if len(tm) == 2:
[thh, tmm] = tm
tss = '0'
elif len(tm) == 3:
[thh, tmm, tss] = tm
else:
return None
try:
yy = int(yy)
dd = int(dd)
thh = int(thh)
tmm = int(tmm)
tss = int(tss)
except ValueError:
return None
tzoffset = None
tz = tz.upper()
if _timezones.has_key(tz):
tzoffset = _timezones[tz]
else:
try:
tzoffset = int(tz)
except ValueError:
pass
# Convert a timezone offset into seconds ; -0500 -> -18000
if tzoffset:
if tzoffset < 0:
tzsign = -1
tzoffset = -tzoffset
else:
tzsign = 1
tzoffset = tzsign * ( (tzoffset/100)*3600 + (tzoffset % 100)*60)
tuple = (yy, mm, dd, thh, tmm, tss, 0, 0, 0, tzoffset)
return tuple
def parsedate(data):
"""Convert a time string to a time tuple."""
t = parsedate_tz(data)
if isinstance(t, TupleType):
return t[:9]
else:
return t
def mktime_tz(data):
"""Turn a 10-tuple as returned by parsedate_tz() into a UTC timestamp."""
if data[9] is None:
# No zone info, so localtime is better assumption than GMT
return time.mktime(data[:8] + (-1,))
else:
t = time.mktime(data[:8] + (0,))
return t - data[9] - time.timezone
def quote(str):
"""Add quotes around a string."""
return str.replace('\\', '\\\\').replace('"', '\\"')
class AddrlistClass:
"""Address parser class by Ben Escoto.
To understand what this class does, it helps to have a copy of RFC 2822 in
front of you.
Note: this class interface is deprecated and may be removed in the future.
Use rfc822.AddressList instead.
"""
def __init__(self, field):
"""Initialize a new instance.
`field' is an unparsed address header field, containing
one or more addresses.
"""
self.specials = '()<>@,:;.\"[]'
self.pos = 0
self.LWS = ' \t'
self.CR = '\r\n'
self.atomends = self.specials + self.LWS + self.CR
# Note that RFC 2822 now specifies `.' as obs-phrase, meaning that it
# is obsolete syntax. RFC 2822 requires that we recognize obsolete
# syntax, so allow dots in phrases.
self.phraseends = self.atomends.replace('.', '')
self.field = field
self.commentlist = []
def gotonext(self):
"""Parse up to the start of the next address."""
while self.pos < len(self.field):
if self.field[self.pos] in self.LWS + '\n\r':
self.pos += 1
elif self.field[self.pos] == '(':
self.commentlist.append(self.getcomment())
else:
break
def getaddrlist(self):
"""Parse all addresses.
Returns a list containing all of the addresses.
"""
result = []
while self.pos < len(self.field):
ad = self.getaddress()
if ad:
result += ad
else:
result.append(('', ''))
return result
def getaddress(self):
"""Parse the next address."""
self.commentlist = []
self.gotonext()
oldpos = self.pos
oldcl = self.commentlist
plist = self.getphraselist()
self.gotonext()
returnlist = []
if self.pos >= len(self.field):
# Bad email address technically, no domain.
if plist:
returnlist = [(SPACE.join(self.commentlist), plist[0])]
elif self.field[self.pos] in '.@':
# email address is just an addrspec
# this isn't very efficient since we start over
self.pos = oldpos
self.commentlist = oldcl
addrspec = self.getaddrspec()
returnlist = [(SPACE.join(self.commentlist), addrspec)]
elif self.field[self.pos] == ':':
# address is a group
returnlist = []
fieldlen = len(self.field)
self.pos += 1
while self.pos < len(self.field):
self.gotonext()
if self.pos < fieldlen and self.field[self.pos] == ';':
self.pos += 1
break
returnlist = returnlist + self.getaddress()
elif self.field[self.pos] == '<':
# Address is a phrase then a route addr
routeaddr = self.getrouteaddr()
if self.commentlist:
returnlist = [(SPACE.join(plist) + ' (' +
' '.join(self.commentlist) + ')', routeaddr)]
else:
returnlist = [(SPACE.join(plist), routeaddr)]
else:
if plist:
returnlist = [(SPACE.join(self.commentlist), plist[0])]
elif self.field[self.pos] in self.specials:
self.pos += 1
self.gotonext()
if self.pos < len(self.field) and self.field[self.pos] == ',':
self.pos += 1
return returnlist
def getrouteaddr(self):
"""Parse a route address (Return-path value).
This method just skips all the route stuff and returns the addrspec.
"""
if self.field[self.pos] != '<':
return
expectroute = False
self.pos += 1
self.gotonext()
adlist = ''
while self.pos < len(self.field):
if expectroute:
self.getdomain()
expectroute = False
elif self.field[self.pos] == '>':
self.pos += 1
break
elif self.field[self.pos] == '@':
self.pos += 1
expectroute = True
elif self.field[self.pos] == ':':
self.pos += 1
else:
adlist = self.getaddrspec()
self.pos += 1
break
self.gotonext()
return adlist
def getaddrspec(self):
"""Parse an RFC 2822 addr-spec."""
aslist = []
self.gotonext()
while self.pos < len(self.field):
if self.field[self.pos] == '.':
aslist.append('.')
self.pos += 1
elif self.field[self.pos] == '"':
aslist.append('"%s"' % self.getquote())
elif self.field[self.pos] in self.atomends:
break
else:
aslist.append(self.getatom())
self.gotonext()
if self.pos >= len(self.field) or self.field[self.pos] != '@':
return EMPTYSTRING.join(aslist)
aslist.append('@')
self.pos += 1
self.gotonext()
return EMPTYSTRING.join(aslist) + self.getdomain()
def getdomain(self):
"""Get the complete domain name from an address."""
sdlist = []
while self.pos < len(self.field):
if self.field[self.pos] in self.LWS:
self.pos += 1
elif self.field[self.pos] == '(':
self.commentlist.append(self.getcomment())
elif self.field[self.pos] == '[':
sdlist.append(self.getdomainliteral())
elif self.field[self.pos] == '.':
self.pos += 1
sdlist.append('.')
elif self.field[self.pos] in self.atomends:
break
else:
sdlist.append(self.getatom())
return EMPTYSTRING.join(sdlist)
def getdelimited(self, beginchar, endchars, allowcomments=True):
"""Parse a header fragment delimited by special characters.
`beginchar' is the start character for the fragment.
If self is not looking at an instance of `beginchar' then
getdelimited returns the empty string.
`endchars' is a sequence of allowable end-delimiting characters.
Parsing stops when one of these is encountered.
If `allowcomments' is non-zero, embedded RFC 2822 comments are allowed
within the parsed fragment.
"""
if self.field[self.pos] != beginchar:
return ''
slist = ['']
quote = False
self.pos += 1
while self.pos < len(self.field):
if quote:
slist.append(self.field[self.pos])
quote = False
elif self.field[self.pos] in endchars:
self.pos += 1
break
elif allowcomments and self.field[self.pos] == '(':
slist.append(self.getcomment())
elif self.field[self.pos] == '\\':
quote = True
else:
slist.append(self.field[self.pos])
self.pos += 1
return EMPTYSTRING.join(slist)
def getquote(self):
"""Get a quote-delimited fragment from self's field."""
return self.getdelimited('"', '"\r', False)
def getcomment(self):
"""Get a parenthesis-delimited fragment from self's field."""
return self.getdelimited('(', ')\r', True)
def getdomainliteral(self):
"""Parse an RFC 2822 domain-literal."""
return '[%s]' % self.getdelimited('[', ']\r', False)
def getatom(self, atomends=None):
"""Parse an RFC 2822 atom.
Optional atomends specifies a different set of end token delimiters
(the default is to use self.atomends). This is used e.g. in
getphraselist() since phrase endings must not include the `.' (which
is legal in phrases)."""
atomlist = ['']
if atomends is None:
atomends = self.atomends
while self.pos < len(self.field):
if self.field[self.pos] in atomends:
break
else:
atomlist.append(self.field[self.pos])
self.pos += 1
return EMPTYSTRING.join(atomlist)
def getphraselist(self):
"""Parse a sequence of RFC 2822 phrases.
A phrase is a sequence of words, which are in turn either RFC 2822
atoms or quoted-strings. Phrases are canonicalized by squeezing all
runs of continuous whitespace into one space.
"""
plist = []
while self.pos < len(self.field):
if self.field[self.pos] in self.LWS:
self.pos += 1
elif self.field[self.pos] == '"':
plist.append(self.getquote())
elif self.field[self.pos] == '(':
self.commentlist.append(self.getcomment())
elif self.field[self.pos] in self.phraseends:
break
else:
plist.append(self.getatom(self.phraseends))
return plist
class AddressList(AddrlistClass):
"""An AddressList encapsulates a list of parsed RFC 2822 addresses."""
def __init__(self, field):
AddrlistClass.__init__(self, field)
if field:
self.addresslist = self.getaddrlist()
else:
self.addresslist = []
def __len__(self):
return len(self.addresslist)
def __str__(self):
return COMMASPACE.join(map(dump_address_pair, self.addresslist))
def __add__(self, other):
# Set union
newaddr = AddressList(None)
newaddr.addresslist = self.addresslist[:]
for x in other.addresslist:
if not x in self.addresslist:
newaddr.addresslist.append(x)
return newaddr
def __iadd__(self, other):
# Set union, in-place
for x in other.addresslist:
if not x in self.addresslist:
self.addresslist.append(x)
return self
def __sub__(self, other):
# Set difference
newaddr = AddressList(None)
for x in self.addresslist:
if not x in other.addresslist:
newaddr.addresslist.append(x)
return newaddr
def __isub__(self, other):
# Set difference, in-place
for x in other.addresslist:
if x in self.addresslist:
self.addresslist.remove(x)
return self
def __getitem__(self, index):
# Make indexing, slices, and 'in' work
return self.addresslist[index]
_______________________________________________
Stackless-checkins mailing list
Stackless-checkins at stackless.com
http://www.stackless.com/mailman/listinfo/stackless-checkins
- Previous message: [Stackless-checkins] CVS: slpdev/src/2.2/src/Lib/hotshot stones.py, NONE, 1.1
- Next message: [Stackless-checkins] CVS: slpdev/src/2.2/src/Lib/distutils/command checkdep.py, NONE, 1.1 register.py, NONE, 1.1 wininst-6.exe, NONE, 1.1 wininst-7.1.exe, NONE, 1.1
- Messages sorted by:
[ date ]
[ thread ]
[ subject ]
[ author ]
More information about the Stackless-checkins
mailing list