message.py (13913B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\email\message.py 2 __all__ = ['Message'] 3 import re 4 import uu 5 import binascii 6 import warnings 7 from cStringIO import StringIO 8 import email.charset 9 from email import utils 10 from email import errors 11 SEMISPACE = '; ' 12 tspecials = re.compile('[ \\(\\)<>@,;:\\\\"/\\[\\]\\?=]') 13 14 def _splitparam(param): 15 a, sep, b = param.partition(';') 16 if not sep: 17 return (a.strip(), None) 18 return (a.strip(), b.strip()) 19 20 21 def _formatparam(param, value = None, quote = True): 22 if value is not None and len(value) > 0: 23 if isinstance(value, tuple): 24 param += '*' 25 value = utils.encode_rfc2231(value[2], value[0], value[1]) 26 if quote or tspecials.search(value): 27 return '%s="%s"' % (param, utils.quote(value)) 28 else: 29 return '%s=%s' % (param, value) 30 else: 31 return param 32 33 34 def _parseparam(s): 35 plist = [] 36 while s[:1] == ';': 37 s = s[1:] 38 end = s.find(';') 39 while end > 0 and (s.count('"', 0, end) - s.count('\\"', 0, end)) % 2: 40 end = s.find(';', end + 1) 41 42 if end < 0: 43 end = len(s) 44 f = s[:end] 45 if '=' in f: 46 i = f.index('=') 47 f = f[:i].strip().lower() + '=' + f[i + 1:].strip() 48 plist.append(f.strip()) 49 s = s[end:] 50 51 return plist 52 53 54 def _unquotevalue(value): 55 if isinstance(value, tuple): 56 return (value[0], value[1], utils.unquote(value[2])) 57 else: 58 return utils.unquote(value) 59 60 61 class Message: 62 63 def __init__(self): 64 self._headers = [] 65 self._unixfrom = None 66 self._payload = None 67 self._charset = None 68 self.preamble = self.epilogue = None 69 self.defects = [] 70 self._default_type = 'text/plain' 71 72 def __str__(self): 73 return self.as_string(unixfrom=True) 74 75 def as_string(self, unixfrom = False): 76 from email.generator import Generator 77 fp = StringIO() 78 g = Generator(fp) 79 g.flatten(self, unixfrom=unixfrom) 80 return fp.getvalue() 81 82 def is_multipart(self): 83 return isinstance(self._payload, list) 84 85 def set_unixfrom(self, unixfrom): 86 self._unixfrom = unixfrom 87 88 def get_unixfrom(self): 89 return self._unixfrom 90 91 def attach(self, payload): 92 if self._payload is None: 93 self._payload = [payload] 94 else: 95 self._payload.append(payload) 96 97 def get_payload(self, i = None, decode = False): 98 if i is None: 99 payload = self._payload 100 elif not isinstance(self._payload, list): 101 raise TypeError('Expected list, got %s' % type(self._payload)) 102 else: 103 payload = self._payload[i] 104 if decode: 105 if self.is_multipart(): 106 return 107 cte = self.get('content-transfer-encoding', '').lower() 108 if cte == 'quoted-printable': 109 return utils._qdecode(payload) 110 if cte == 'base64': 111 try: 112 return utils._bdecode(payload) 113 except binascii.Error: 114 return payload 115 116 elif cte in ('x-uuencode', 'uuencode', 'uue', 'x-uue'): 117 sfp = StringIO() 118 try: 119 uu.decode(StringIO(payload + '\n'), sfp, quiet=True) 120 payload = sfp.getvalue() 121 except uu.Error: 122 return payload 123 124 return payload 125 126 def set_payload(self, payload, charset = None): 127 self._payload = payload 128 if charset is not None: 129 self.set_charset(charset) 130 131 def set_charset(self, charset): 132 if charset is None: 133 self.del_param('charset') 134 self._charset = None 135 return 136 if isinstance(charset, basestring): 137 charset = email.charset.Charset(charset) 138 if not isinstance(charset, email.charset.Charset): 139 raise TypeError(charset) 140 self._charset = charset 141 if 'MIME-Version' not in self: 142 self.add_header('MIME-Version', '1.0') 143 if 'Content-Type' not in self: 144 self.add_header('Content-Type', 'text/plain', charset=charset.get_output_charset()) 145 else: 146 self.set_param('charset', charset.get_output_charset()) 147 if isinstance(self._payload, unicode): 148 self._payload = self._payload.encode(charset.output_charset) 149 if str(charset) != charset.get_output_charset(): 150 self._payload = charset.body_encode(self._payload) 151 if 'Content-Transfer-Encoding' not in self: 152 cte = charset.get_body_encoding() 153 try: 154 cte(self) 155 except TypeError: 156 self._payload = charset.body_encode(self._payload) 157 self.add_header('Content-Transfer-Encoding', cte) 158 159 def get_charset(self): 160 return self._charset 161 162 def __len__(self): 163 return len(self._headers) 164 165 def __getitem__(self, name): 166 return self.get(name) 167 168 def __setitem__(self, name, val): 169 self._headers.append((name, val)) 170 171 def __delitem__(self, name): 172 name = name.lower() 173 newheaders = [] 174 for k, v in self._headers: 175 if k.lower() != name: 176 newheaders.append((k, v)) 177 178 self._headers = newheaders 179 180 def __contains__(self, name): 181 return name.lower() in [ k.lower() for k, v in self._headers ] 182 183 def has_key(self, name): 184 missing = object() 185 return self.get(name, missing) is not missing 186 187 def keys(self): 188 return [ k for k, v in self._headers ] 189 190 def values(self): 191 return [ v for k, v in self._headers ] 192 193 def items(self): 194 return self._headers[:] 195 196 def get(self, name, failobj = None): 197 name = name.lower() 198 for k, v in self._headers: 199 if k.lower() == name: 200 return v 201 202 return failobj 203 204 def get_all(self, name, failobj = None): 205 values = [] 206 name = name.lower() 207 for k, v in self._headers: 208 if k.lower() == name: 209 values.append(v) 210 211 if not values: 212 return failobj 213 return values 214 215 def add_header(self, _name, _value, **_params): 216 parts = [] 217 for k, v in _params.items(): 218 if v is None: 219 parts.append(k.replace('_', '-')) 220 else: 221 parts.append(_formatparam(k.replace('_', '-'), v)) 222 223 if _value is not None: 224 parts.insert(0, _value) 225 self._headers.append((_name, SEMISPACE.join(parts))) 226 227 def replace_header(self, _name, _value): 228 _name = _name.lower() 229 for i, (k, v) in zip(range(len(self._headers)), self._headers): 230 if k.lower() == _name: 231 self._headers[i] = (k, _value) 232 break 233 else: 234 raise KeyError(_name) 235 236 def get_content_type(self): 237 missing = object() 238 value = self.get('content-type', missing) 239 if value is missing: 240 return self.get_default_type() 241 ctype = _splitparam(value)[0].lower() 242 if ctype.count('/') != 1: 243 return 'text/plain' 244 return ctype 245 246 def get_content_maintype(self): 247 ctype = self.get_content_type() 248 return ctype.split('/')[0] 249 250 def get_content_subtype(self): 251 ctype = self.get_content_type() 252 return ctype.split('/')[1] 253 254 def get_default_type(self): 255 return self._default_type 256 257 def set_default_type(self, ctype): 258 self._default_type = ctype 259 260 def _get_params_preserve(self, failobj, header): 261 missing = object() 262 value = self.get(header, missing) 263 if value is missing: 264 return failobj 265 params = [] 266 for p in _parseparam(';' + value): 267 try: 268 name, val = p.split('=', 1) 269 name = name.strip() 270 val = val.strip() 271 except ValueError: 272 name = p.strip() 273 val = '' 274 275 params.append((name, val)) 276 277 params = utils.decode_params(params) 278 return params 279 280 def get_params(self, failobj = None, header = 'content-type', unquote = True): 281 missing = object() 282 params = self._get_params_preserve(missing, header) 283 if params is missing: 284 return failobj 285 elif unquote: 286 return [ (k, _unquotevalue(v)) for k, v in params ] 287 else: 288 return params 289 290 def get_param(self, param, failobj = None, header = 'content-type', unquote = True): 291 if header not in self: 292 return failobj 293 for k, v in self._get_params_preserve(failobj, header): 294 if k.lower() == param.lower(): 295 if unquote: 296 return _unquotevalue(v) 297 else: 298 return v 299 300 return failobj 301 302 def set_param(self, param, value, header = 'Content-Type', requote = True, charset = None, language = ''): 303 if not isinstance(value, tuple) and charset: 304 value = (charset, language, value) 305 if header not in self and header.lower() == 'content-type': 306 ctype = 'text/plain' 307 else: 308 ctype = self.get(header) 309 if not self.get_param(param, header=header): 310 if not ctype: 311 ctype = _formatparam(param, value, requote) 312 else: 313 ctype = SEMISPACE.join([ctype, _formatparam(param, value, requote)]) 314 else: 315 ctype = '' 316 for old_param, old_value in self.get_params(header=header, unquote=requote): 317 append_param = '' 318 if old_param.lower() == param.lower(): 319 append_param = _formatparam(param, value, requote) 320 else: 321 append_param = _formatparam(old_param, old_value, requote) 322 if not ctype: 323 ctype = append_param 324 else: 325 ctype = SEMISPACE.join([ctype, append_param]) 326 327 if ctype != self.get(header): 328 del self[header] 329 self[header] = ctype 330 331 def del_param(self, param, header = 'content-type', requote = True): 332 if header not in self: 333 return 334 new_ctype = '' 335 for p, v in self.get_params(header=header, unquote=requote): 336 if p.lower() != param.lower(): 337 if not new_ctype: 338 new_ctype = _formatparam(p, v, requote) 339 else: 340 new_ctype = SEMISPACE.join([new_ctype, _formatparam(p, v, requote)]) 341 342 if new_ctype != self.get(header): 343 del self[header] 344 self[header] = new_ctype 345 346 def set_type(self, type, header = 'Content-Type', requote = True): 347 if not type.count('/') == 1: 348 raise ValueError 349 if header.lower() == 'content-type': 350 del self['mime-version'] 351 self['MIME-Version'] = '1.0' 352 if header not in self: 353 self[header] = type 354 return 355 params = self.get_params(header=header, unquote=requote) 356 del self[header] 357 self[header] = type 358 for p, v in params[1:]: 359 self.set_param(p, v, header, requote) 360 361 def get_filename(self, failobj = None): 362 missing = object() 363 filename = self.get_param('filename', missing, 'content-disposition') 364 if filename is missing: 365 filename = self.get_param('name', missing, 'content-type') 366 if filename is missing: 367 return failobj 368 return utils.collapse_rfc2231_value(filename).strip() 369 370 def get_boundary(self, failobj = None): 371 missing = object() 372 boundary = self.get_param('boundary', missing) 373 if boundary is missing: 374 return failobj 375 return utils.collapse_rfc2231_value(boundary).rstrip() 376 377 def set_boundary(self, boundary): 378 missing = object() 379 params = self._get_params_preserve(missing, 'content-type') 380 if params is missing: 381 raise errors.HeaderParseError('No Content-Type header found') 382 newparams = [] 383 foundp = False 384 for pk, pv in params: 385 if pk.lower() == 'boundary': 386 newparams.append(('boundary', '"%s"' % boundary)) 387 foundp = True 388 else: 389 newparams.append((pk, pv)) 390 391 if not foundp: 392 newparams.append(('boundary', '"%s"' % boundary)) 393 newheaders = [] 394 for h, v in self._headers: 395 if h.lower() == 'content-type': 396 parts = [] 397 for k, v in newparams: 398 if v == '': 399 parts.append(k) 400 else: 401 parts.append('%s=%s' % (k, v)) 402 403 newheaders.append((h, SEMISPACE.join(parts))) 404 else: 405 newheaders.append((h, v)) 406 407 self._headers = newheaders 408 409 def get_content_charset(self, failobj = None): 410 missing = object() 411 charset = self.get_param('charset', missing) 412 if charset is missing: 413 return failobj 414 if isinstance(charset, tuple): 415 pcharset = charset[0] or 'us-ascii' 416 try: 417 charset = unicode(charset[2], pcharset).encode('us-ascii') 418 except (LookupError, UnicodeError): 419 charset = charset[2] 420 421 try: 422 if isinstance(charset, str): 423 charset = unicode(charset, 'us-ascii') 424 charset = charset.encode('us-ascii') 425 except UnicodeError: 426 return failobj 427 428 return charset.lower() 429 430 def get_charsets(self, failobj = None): 431 return [ part.get_content_charset(failobj) for part in self.walk() ] 432 433 from email.iterators import walk