<s> """<STR_LIT>""" <EOL> __all__ = [ '<STR_LIT>' ] <EOL> import re <EOL> import uu <EOL> import binascii <EOL> import warnings <EOL> from cStringIO import StringIO <EOL> import email . charset <EOL> from email import utils <EOL> from email import errors <EOL> SEMISPACE = '<STR_LIT>' <EOL> tspecials = re . compile ( r'<STR_LIT>' ) <EOL> def _splitparam ( param ) : <EOL> a , sep , b = param . partition ( '<STR_LIT:;>' ) <EOL> if not sep : <EOL> return a . strip ( ) , None <EOL> return a . strip ( ) , b . strip ( ) <EOL> def _formatparam ( param , value = None , quote = True ) : <EOL> """<STR_LIT>""" <EOL> if value is not None and len ( value ) > <NUM_LIT:0> : <EOL> if isinstance ( value , tuple ) : <EOL> param += '<STR_LIT:*>' <EOL> value = utils . encode_rfc2231 ( value [ <NUM_LIT:2> ] , value [ <NUM_LIT:0> ] , value [ <NUM_LIT:1> ] ) <EOL> if quote or tspecials . search ( value ) : <EOL> return '<STR_LIT>' % ( param , utils . quote ( value ) ) <EOL> else : <EOL> return '<STR_LIT>' % ( param , value ) <EOL> else : <EOL> return param <EOL> def _parseparam ( s ) : <EOL> plist = [ ] <EOL> while s [ : <NUM_LIT:1> ] == '<STR_LIT:;>' : <EOL> s = s [ <NUM_LIT:1> : ] <EOL> end = s . find ( '<STR_LIT:;>' ) <EOL> while end > <NUM_LIT:0> and ( s . count ( '<STR_LIT:">' , <NUM_LIT:0> , end ) - s . count ( '<STR_LIT>' , <NUM_LIT:0> , end ) ) % <NUM_LIT:2> : <EOL> end = s . find ( '<STR_LIT:;>' , end + <NUM_LIT:1> ) <EOL> if end < <NUM_LIT:0> : <EOL> end = len ( s ) <EOL> f = s [ : end ] <EOL> if '<STR_LIT:=>' in f : <EOL> i = f . index ( '<STR_LIT:=>' ) <EOL> f = f [ : i ] . strip ( ) . lower ( ) + '<STR_LIT:=>' + f [ i + <NUM_LIT:1> : ] . strip ( ) <EOL> plist . append ( f . strip ( ) ) <EOL> s = s [ end : ] <EOL> return plist <EOL> def _unquotevalue ( value ) : <EOL> if isinstance ( value , tuple ) : <EOL> return value [ <NUM_LIT:0> ] , value [ <NUM_LIT:1> ] , utils . unquote ( value [ <NUM_LIT:2> ] ) <EOL> else : <EOL> return utils . unquote ( value ) <EOL> class Message : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _headers = [ ] <EOL> self . _unixfrom = None <EOL> self . _payload = None <EOL> self . _charset = None <EOL> self . preamble = self . epilogue = None <EOL> self . defects = [ ] <EOL> self . _default_type = '<STR_LIT>' <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . as_string ( unixfrom = True ) <EOL> def as_string ( self , unixfrom = False ) : <EOL> """<STR_LIT>""" <EOL> from email . generator import Generator <EOL> fp = StringIO ( ) <EOL> g = Generator ( fp ) <EOL> g . flatten ( self , unixfrom = unixfrom ) <EOL> return fp . getvalue ( ) <EOL> def is_multipart ( self ) : <EOL> """<STR_LIT>""" <EOL> return isinstance ( self . _payload , list ) <EOL> def set_unixfrom ( self , unixfrom ) : <EOL> self . _unixfrom = unixfrom <EOL> def get_unixfrom ( self ) : <EOL> return self . _unixfrom <EOL> def attach ( self , payload ) : <EOL> """<STR_LIT>""" <EOL> if self . _payload is None : <EOL> self . _payload = [ payload ] <EOL> else : <EOL> self . _payload . append ( payload ) <EOL> def get_payload ( self , i = None , decode = False ) : <EOL> """<STR_LIT>""" <EOL> if i is None : <EOL> payload = self . _payload <EOL> elif not isinstance ( self . _payload , list ) : <EOL> raise TypeError ( '<STR_LIT>' % type ( self . _payload ) ) <EOL> else : <EOL> payload = self . _payload [ i ] <EOL> if decode : <EOL> if self . is_multipart ( ) : <EOL> return None <EOL> cte = self . get ( '<STR_LIT>' , '<STR_LIT>' ) . lower ( ) <EOL> if cte == '<STR_LIT>' : <EOL> return utils . _qdecode ( payload ) <EOL> elif cte == '<STR_LIT>' : <EOL> try : <EOL> return utils . _bdecode ( payload ) <EOL> except binascii . Error : <EOL> return payload <EOL> elif cte in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> sfp = StringIO ( ) <EOL> try : <EOL> uu . decode ( StringIO ( payload + '<STR_LIT:\n>' ) , sfp , quiet = True ) <EOL> payload = sfp . getvalue ( ) <EOL> except uu . Error : <EOL> return payload <EOL> return payload <EOL> def set_payload ( self , payload , charset = None ) : <EOL> """<STR_LIT>""" <EOL> self . _payload = payload <EOL> if charset is not None : <EOL> self . set_charset ( charset ) <EOL> def set_charset ( self , charset ) : <EOL> """<STR_LIT>""" <EOL> if charset is None : <EOL> self . del_param ( '<STR_LIT>' ) <EOL> self . _charset = None <EOL> return <EOL> if isinstance ( charset , basestring ) : <EOL> charset = email . charset . Charset ( charset ) <EOL> if not isinstance ( charset , email . charset . Charset ) : <EOL> raise TypeError ( charset ) <EOL> self . _charset = charset <EOL> if '<STR_LIT>' not in self : <EOL> self . add_header ( '<STR_LIT>' , '<STR_LIT:1.0>' ) <EOL> if '<STR_LIT:Content-Type>' not in self : <EOL> self . add_header ( '<STR_LIT:Content-Type>' , '<STR_LIT>' , <EOL> charset = charset . get_output_charset ( ) ) <EOL> else : <EOL> self . set_param ( '<STR_LIT>' , charset . get_output_charset ( ) ) <EOL> if isinstance ( self . _payload , unicode ) : <EOL> self . _payload = self . _payload . encode ( charset . output_charset ) <EOL> if str ( charset ) != charset . get_output_charset ( ) : <EOL> self . _payload = charset . body_encode ( self . _payload ) <EOL> if '<STR_LIT>' not in self : <EOL> cte = charset . get_body_encoding ( ) <EOL> try : <EOL> cte ( self ) <EOL> except TypeError : <EOL> self . _payload = charset . body_encode ( self . _payload ) <EOL> self . add_header ( '<STR_LIT>' , cte ) <EOL> def get_charset ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _charset <EOL> def __len__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . _headers ) <EOL> def __getitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . get ( name ) <EOL> def __setitem__ ( self , name , val ) : <EOL> """<STR_LIT>""" <EOL> self . _headers . append ( ( name , val ) ) <EOL> def __delitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> name = name . lower ( ) <EOL> newheaders = [ ] <EOL> for k , v in self . _headers : <EOL> if k . lower ( ) != name : <EOL> newheaders . append ( ( k , v ) ) <EOL> self . _headers = newheaders <EOL> def __contains__ ( self , name ) : <EOL> return name . lower ( ) in [ k . lower ( ) for k , v in self . _headers ] <EOL> def has_key ( self , name ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> return self . get ( name , missing ) is not missing <EOL> def keys ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ k for k , v in self . _headers ] <EOL> def values ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ v for k , v in self . _headers ] <EOL> def items ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _headers [ : ] <EOL> def get ( self , name , failobj = None ) : <EOL> """<STR_LIT>""" <EOL> name = name . lower ( ) <EOL> for k , v in self . _headers : <EOL> if k . lower ( ) == name : <EOL> return v <EOL> return failobj <EOL> def get_all ( self , name , failobj = None ) : <EOL> """<STR_LIT>""" <EOL> values = [ ] <EOL> name = name . lower ( ) <EOL> for k , v in self . _headers : <EOL> if k . lower ( ) == name : <EOL> values . append ( v ) <EOL> if not values : <EOL> return failobj <EOL> return values <EOL> def add_header ( self , _name , _value , ** _params ) : <EOL> """<STR_LIT>""" <EOL> parts = [ ] <EOL> for k , v in _params . items ( ) : <EOL> if v is None : <EOL> parts . append ( k . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) ) <EOL> else : <EOL> parts . append ( _formatparam ( k . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) , v ) ) <EOL> if _value is not None : <EOL> parts . insert ( <NUM_LIT:0> , _value ) <EOL> self . _headers . append ( ( _name , SEMISPACE . join ( parts ) ) ) <EOL> def replace_header ( self , _name , _value ) : <EOL> """<STR_LIT>""" <EOL> _name = _name . lower ( ) <EOL> for i , ( k , v ) in zip ( range ( len ( self . _headers ) ) , self . _headers ) : <EOL> if k . lower ( ) == _name : <EOL> self . _headers [ i ] = ( k , _value ) <EOL> break <EOL> else : <EOL> raise KeyError ( _name ) <EOL> def get_content_type ( self ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> value = self . get ( '<STR_LIT>' , missing ) <EOL> if value is missing : <EOL> return self . get_default_type ( ) <EOL> ctype = _splitparam ( value ) [ <NUM_LIT:0> ] . lower ( ) <EOL> if ctype . count ( '<STR_LIT:/>' ) != <NUM_LIT:1> : <EOL> return '<STR_LIT>' <EOL> return ctype <EOL> def get_content_maintype ( self ) : <EOL> """<STR_LIT>""" <EOL> ctype = self . get_content_type ( ) <EOL> return ctype . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> def get_content_subtype ( self ) : <EOL> """<STR_LIT>""" <EOL> ctype = self . get_content_type ( ) <EOL> return ctype . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] <EOL> def get_default_type ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _default_type <EOL> def set_default_type ( self , ctype ) : <EOL> """<STR_LIT>""" <EOL> self . _default_type = ctype <EOL> def _get_params_preserve ( self , failobj , header ) : <EOL> missing = object ( ) <EOL> value = self . get ( header , missing ) <EOL> if value is missing : <EOL> return failobj <EOL> params = [ ] <EOL> for p in _parseparam ( '<STR_LIT:;>' + value ) : <EOL> try : <EOL> name , val = p . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) <EOL> name = name . strip ( ) <EOL> val = val . strip ( ) <EOL> except ValueError : <EOL> name = p . strip ( ) <EOL> val = '<STR_LIT>' <EOL> params . append ( ( name , val ) ) <EOL> params = utils . decode_params ( params ) <EOL> return params <EOL> def get_params ( self , failobj = None , header = '<STR_LIT>' , unquote = True ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> params = self . _get_params_preserve ( missing , header ) <EOL> if params is missing : <EOL> return failobj <EOL> if unquote : <EOL> return [ ( k , _unquotevalue ( v ) ) for k , v in params ] <EOL> else : <EOL> return params <EOL> def get_param ( self , param , failobj = None , header = '<STR_LIT>' , <EOL> unquote = True ) : <EOL> """<STR_LIT>""" <EOL> if header not in self : <EOL> return failobj <EOL> for k , v in self . _get_params_preserve ( failobj , header ) : <EOL> if k . lower ( ) == param . lower ( ) : <EOL> if unquote : <EOL> return _unquotevalue ( v ) <EOL> else : <EOL> return v <EOL> return failobj <EOL> def set_param ( self , param , value , header = '<STR_LIT:Content-Type>' , requote = True , <EOL> charset = None , language = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( value , tuple ) and charset : <EOL> value = ( charset , language , value ) <EOL> if header not in self and header . lower ( ) == '<STR_LIT>' : <EOL> ctype = '<STR_LIT>' <EOL> else : <EOL> ctype = self . get ( header ) <EOL> if not self . get_param ( param , header = header ) : <EOL> if not ctype : <EOL> ctype = _formatparam ( param , value , requote ) <EOL> else : <EOL> ctype = SEMISPACE . join ( <EOL> [ ctype , _formatparam ( param , value , requote ) ] ) <EOL> else : <EOL> ctype = '<STR_LIT>' <EOL> for old_param , old_value in self . get_params ( header = header , <EOL> unquote = requote ) : <EOL> append_param = '<STR_LIT>' <EOL> if old_param . lower ( ) == param . lower ( ) : <EOL> append_param = _formatparam ( param , value , requote ) <EOL> else : <EOL> append_param = _formatparam ( old_param , old_value , requote ) <EOL> if not ctype : <EOL> ctype = append_param <EOL> else : <EOL> ctype = SEMISPACE . join ( [ ctype , append_param ] ) <EOL> if ctype != self . get ( header ) : <EOL> del self [ header ] <EOL> self [ header ] = ctype <EOL> def del_param ( self , param , header = '<STR_LIT>' , requote = True ) : <EOL> """<STR_LIT>""" <EOL> if header not in self : <EOL> return <EOL> new_ctype = '<STR_LIT>' <EOL> for p , v in self . get_params ( header = header , unquote = requote ) : <EOL> if p . lower ( ) != param . lower ( ) : <EOL> if not new_ctype : <EOL> new_ctype = _formatparam ( p , v , requote ) <EOL> else : <EOL> new_ctype = SEMISPACE . join ( [ new_ctype , <EOL> _formatparam ( p , v , requote ) ] ) <EOL> if new_ctype != self . get ( header ) : <EOL> del self [ header ] <EOL> self [ header ] = new_ctype <EOL> def set_type ( self , type , header = '<STR_LIT:Content-Type>' , requote = True ) : <EOL> """<STR_LIT>""" <EOL> if not type . count ( '<STR_LIT:/>' ) == <NUM_LIT:1> : <EOL> raise ValueError <EOL> if header . lower ( ) == '<STR_LIT>' : <EOL> del self [ '<STR_LIT>' ] <EOL> self [ '<STR_LIT>' ] = '<STR_LIT:1.0>' <EOL> if header not in self : <EOL> self [ header ] = type <EOL> return <EOL> params = self . get_params ( header = header , unquote = requote ) <EOL> del self [ header ] <EOL> self [ header ] = type <EOL> for p , v in params [ <NUM_LIT:1> : ] : <EOL> self . set_param ( p , v , header , requote ) <EOL> def get_filename ( self , failobj = None ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> filename = self . get_param ( '<STR_LIT:filename>' , missing , '<STR_LIT>' ) <EOL> if filename is missing : <EOL> filename = self . get_param ( '<STR_LIT:name>' , missing , '<STR_LIT>' ) <EOL> if filename is missing : <EOL> return failobj <EOL> return utils . collapse_rfc2231_value ( filename ) . strip ( ) <EOL> def get_boundary ( self , failobj = None ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> boundary = self . get_param ( '<STR_LIT>' , missing ) <EOL> if boundary is missing : <EOL> return failobj <EOL> return utils . collapse_rfc2231_value ( boundary ) . rstrip ( ) <EOL> def set_boundary ( self , boundary ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> params = self . _get_params_preserve ( missing , '<STR_LIT>' ) <EOL> if params is missing : <EOL> raise errors . HeaderParseError ( '<STR_LIT>' ) <EOL> newparams = [ ] <EOL> foundp = False <EOL> for pk , pv in params : <EOL> if pk . lower ( ) == '<STR_LIT>' : <EOL> newparams . append ( ( '<STR_LIT>' , '<STR_LIT>' % boundary ) ) <EOL> foundp = True <EOL> else : <EOL> newparams . append ( ( pk , pv ) ) <EOL> if not foundp : <EOL> newparams . append ( ( '<STR_LIT>' , '<STR_LIT>' % boundary ) ) <EOL> newheaders = [ ] <EOL> for h , v in self . _headers : <EOL> if h . lower ( ) == '<STR_LIT>' : <EOL> parts = [ ] <EOL> for k , v in newparams : <EOL> if v == '<STR_LIT>' : <EOL> parts . append ( k ) <EOL> else : <EOL> parts . append ( '<STR_LIT>' % ( k , v ) ) <EOL> newheaders . append ( ( h , SEMISPACE . join ( parts ) ) ) <EOL> else : <EOL> newheaders . append ( ( h , v ) ) <EOL> self . _headers = newheaders <EOL> def get_content_charset ( self , failobj = None ) : <EOL> """<STR_LIT>""" <EOL> missing = object ( ) <EOL> charset = self . get_param ( '<STR_LIT>' , missing ) <EOL> if charset is missing : <EOL> return failobj <EOL> if isinstance ( charset , tuple ) : <EOL> pcharset = charset [ <NUM_LIT:0> ] or '<STR_LIT>' <EOL> try : <EOL> charset = unicode ( charset [ <NUM_LIT:2> ] , pcharset ) . encode ( '<STR_LIT>' ) <EOL> except ( LookupError , UnicodeError ) : <EOL> charset = charset [ <NUM_LIT:2> ] <EOL> try : <EOL> if isinstance ( charset , str ) : <EOL> charset = unicode ( charset , '<STR_LIT>' ) <EOL> charset = charset . encode ( '<STR_LIT>' ) <EOL> except UnicodeError : <EOL> return failobj <EOL> return charset . lower ( ) <EOL> def get_charsets ( self , failobj = None ) : <EOL> """<STR_LIT>""" <EOL> return [ part . get_content_charset ( failobj ) for part in self . walk ( ) ] <EOL> from email . iterators import walk </s>
