我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用email.message.seek()。
def _generate_toc(self): """Generate key-to-(start, stop) table of contents.""" starts, stops = [], [] self._file.seek(0) while True: line_pos = self._file.tell() line = self._file.readline() if line.startswith('From '): if len(stops) < len(starts): stops.append(line_pos - len(os.linesep)) starts.append(line_pos) elif line == '': stops.append(line_pos) break self._toc = dict(enumerate(zip(starts, stops))) self._next_key = len(self._toc) self._file_length = self._file.tell()
def get_message(self, key): """Return a Message representation or raise a KeyError.""" start, stop = self._lookup(key) self._file.seek(start) self._file.readline() # Skip '1,' line specifying labels. original_headers = StringIO.StringIO() while True: line = self._file.readline() if line == '*** EOOH ***' + os.linesep or line == '': break original_headers.write(line.replace(os.linesep, '\n')) visible_headers = StringIO.StringIO() while True: line = self._file.readline() if line == os.linesep or line == '': break visible_headers.write(line.replace(os.linesep, '\n')) body = self._file.read(stop - self._file.tell()).replace(os.linesep, '\n') msg = BabylMessage(original_headers.getvalue() + body) msg.set_visible(visible_headers.getvalue()) if key in self._labels: msg.set_labels(self._labels[key]) return msg
def get_string(self, key): """Return a string representation or raise a KeyError.""" start, stop = self._lookup(key) self._file.seek(start) self._file.readline() # Skip '1,' line specifying labels. original_headers = StringIO.StringIO() while True: line = self._file.readline() if line == '*** EOOH ***' + os.linesep or line == '': break original_headers.write(line.replace(os.linesep, '\n')) while True: line = self._file.readline() if line == os.linesep or line == '': break return original_headers.getvalue() + \ self._file.read(stop - self._file.tell()).replace(os.linesep, '\n')
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 self.factory(_PartialFile(self.fp, start, stop)) # Recommended to use PortableUnixMailbox instead!
def _append_message(self, message): """Append message to mailbox and return (start, stop) offsets.""" self._file.seek(0, 2) before = self._file.tell() if len(self._toc) == 0 and not self._pending: # This is the first message, and the _pre_mailbox_hook # hasn't yet been called. If self._pending is True, # messages have been removed, so _pre_mailbox_hook must # have been called already. self._pre_mailbox_hook(self._file) try: self._pre_message_hook(self._file) offsets = self._install_message(message) self._post_message_hook(self._file) except BaseException: self._file.truncate(before) raise self._file.flush() self._file_length = self._file.tell() # Record current length of mailbox return offsets
def _generate_toc(self): """Generate key-to-(start, stop) table of contents.""" starts, stops = [], [] self._file.seek(0) while True: line_pos = self._file.tell() line = self._file.readline() if line.startswith(b'From '): if len(stops) < len(starts): stops.append(line_pos - len(linesep)) starts.append(line_pos) elif not line: stops.append(line_pos) break self._toc = dict(enumerate(zip(starts, stops))) self._next_key = len(self._toc) self._file_length = self._file.tell()
def get_bytes(self, key): """Return a string representation or raise a KeyError.""" start, stop = self._lookup(key) self._file.seek(start) self._file.readline() # Skip b'1,' line specifying labels. original_headers = io.BytesIO() while True: line = self._file.readline() if line == b'*** EOOH ***' + linesep or not line: break original_headers.write(line.replace(linesep, b'\n')) while True: line = self._file.readline() if line == linesep or not line: break headers = original_headers.getvalue() n = stop - self._file.tell() assert n >= 0 data = self._file.read(n) data = data.replace(linesep, b'\n') return headers + data
def _dump_message(self, message, target, mangle_from_=False): # Most files are opened in binary mode to allow predictable seeking. # To get native line endings on disk, the user-friendly \n line endings # used in strings and by email.Message are translated here. """Dump message contents to target file.""" if isinstance(message, email.message.Message): buffer = StringIO.StringIO() gen = email.generator.Generator(buffer, mangle_from_, 0) gen.flatten(message) buffer.seek(0) target.write(buffer.read().replace('\n', os.linesep)) elif isinstance(message, str): if mangle_from_: message = message.replace('\nFrom ', '\n>From ') message = message.replace('\n', os.linesep) target.write(message) elif hasattr(message, 'read'): while True: line = message.readline() if line == '': break if mangle_from_ and line.startswith('From '): line = '>From ' + line[5:] line = line.replace('\n', os.linesep) target.write(line) else: raise TypeError('Invalid message type: %s' % type(message))
def _append_message(self, message): """Append message to mailbox and return (start, stop) offsets.""" self._file.seek(0, 2) before = self._file.tell() try: self._pre_message_hook(self._file) offsets = self._install_message(message) self._post_message_hook(self._file) except BaseException: self._file.truncate(before) raise self._file.flush() self._file_length = self._file.tell() # Record current length of mailbox return offsets
def get_message(self, key): """Return a Message representation or raise a KeyError.""" start, stop = self._lookup(key) self._file.seek(start) from_line = self._file.readline().replace(os.linesep, '') string = self._file.read(stop - self._file.tell()) msg = self._message_factory(string.replace(os.linesep, '\n')) msg.set_from(from_line[5:]) return msg
def get_string(self, key, from_=False): """Return a string representation or raise a KeyError.""" start, stop = self._lookup(key) self._file.seek(start) if not from_: self._file.readline() string = self._file.read(stop - self._file.tell()) return string.replace(os.linesep, '\n')
def _generate_toc(self): """Generate key-to-(start, stop) table of contents.""" starts, stops = [], [] self._file.seek(0) next_pos = 0 while True: line_pos = next_pos line = self._file.readline() next_pos = self._file.tell() if line.startswith('\001\001\001\001' + os.linesep): starts.append(next_pos) while True: line_pos = next_pos line = self._file.readline() next_pos = self._file.tell() if line == '\001\001\001\001' + os.linesep: stops.append(line_pos - len(os.linesep)) break elif line == '': stops.append(line_pos) break elif line == '': break self._toc = dict(enumerate(zip(starts, stops))) self._next_key = len(self._toc) self._file.seek(0, 2) self._file_length = self._file.tell()
def _generate_toc(self): """Generate key-to-(start, stop) table of contents.""" starts, stops = [], [] self._file.seek(0) next_pos = 0 label_lists = [] while True: line_pos = next_pos line = self._file.readline() next_pos = self._file.tell() if line == '\037\014' + os.linesep: if len(stops) < len(starts): stops.append(line_pos - len(os.linesep)) starts.append(next_pos) labels = [label.strip() for label in self._file.readline()[1:].split(',') if label.strip() != ''] label_lists.append(labels) elif line == '\037' or line == '\037' + os.linesep: if len(stops) < len(starts): stops.append(line_pos - len(os.linesep)) elif line == '': stops.append(line_pos - len(os.linesep)) break self._toc = dict(enumerate(zip(starts, stops))) self._labels = dict(enumerate(label_lists)) self._next_key = len(self._toc) self._file.seek(0, 2) self._file_length = self._file.tell()
def _read(self, size, read_method): """Read size bytes using read_method.""" if size is None: size = -1 self._file.seek(self._pos) result = read_method(size) self._pos = self._file.tell() return result
def seek(self, offset, whence=0): """Change position, possibly with respect to start or stop.""" if whence == 0: self._pos = self._start whence = 1 elif whence == 2: self._pos = self._stop whence = 1 _ProxyFile.seek(self, offset, whence)
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 ". # # BAW: According to #http://home.netscape.com/eng/mozilla/2.0/relnotes/demo/content-length.html # the only portable, reliable way to find message delimiters in a BSD (i.e # Unix mailbox) style folder is to search for "\n\nFrom .*\n", or at the # beginning of the file, "^From .*\n". While _fromlinepattern below seems # like a good idea, in practice, there are too many variations for more # strict parsing of the line to be completely accurate. # # _strict_isrealfromline() is the old version which tries to do stricter # parsing of the From_ line. _portable_isrealfromline() simply returns # true, since it's never called if the line doesn't already start with # "From ". # # This algorithm, and the way it interacts with _search_start() and # _search_end() may not be completely correct, because it doesn't check # that the two characters preceding "From " are \n\n or the beginning of # the file. Fixing this would require a more extensive rewrite than is # necessary. For convenience, we've added a PortableUnixMailbox class # which does no checking of the format of the 'From' line.
def _search_end(self): while 1: pos = self.fp.tell() line = self.fp.readline() if not line: return if line == '\037\014\n' or line == '\037': self.fp.seek(pos) return ## End: classes from the original module (for backward compatibility).
def _dump_message(self, message, target, mangle_from_=False): # Most files are opened in binary mode to allow predictable seeking. # To get native line endings on disk, the user-friendly \n line endings # used in strings and by email.Message are translated here. """Dump message contents to target file.""" if isinstance(message, email.message.Message): buffer = StringIO.StringIO() gen = email.generator.Generator(buffer, mangle_from_, 0) gen.flatten(message) buffer.seek(0) data = buffer.read().replace('\n', os.linesep) target.write(data) if self._append_newline and not data.endswith(os.linesep): # Make sure the message ends with a newline target.write(os.linesep) elif isinstance(message, str): if mangle_from_: message = message.replace('\nFrom ', '\n>From ') message = message.replace('\n', os.linesep) target.write(message) if self._append_newline and not message.endswith(os.linesep): # Make sure the message ends with a newline target.write(os.linesep) elif hasattr(message, 'read'): lastline = None while True: line = message.readline() if line == '': break if mangle_from_ and line.startswith('From '): line = '>From ' + line[5:] line = line.replace('\n', os.linesep) target.write(line) lastline = line if self._append_newline and lastline and not lastline.endswith(os.linesep): # Make sure the message ends with a newline target.write(os.linesep) else: raise TypeError('Invalid message type: %s' % type(message))
