1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
import logging
from collections import deque
import six
import flanker.addresslib.address
from flanker import _email
from flanker.mime.message.headers import parametrized
from flanker.mime.message.utils import to_utf8
_log = logging.getLogger(__name__)
_ADDRESS_HEADERS = ('From', 'To', 'Delivered-To', 'Cc', 'Bcc', 'Reply-To', 'Sender')
def to_mime(key, value):
if not value:
return ''
if type(value) == list:
return '; '.join(encode(key, v) for v in value)
return encode(key, value)
def encode(name, value):
try:
if parametrized.is_parametrized(name, value):
value, params = value
return _encode_parametrized(name, value, params)
return _encode_unstructured(name, value)
except Exception:
_log.exception('Failed to encode %s %s' % (name, value))
raise
def _encode_unstructured(name, value):
try:
return _email.encode_header(name, value.encode('ascii'), 'ascii')
except (UnicodeEncodeError, UnicodeDecodeError):
if _is_address_header(name, value):
return _encode_address_header(name, value)
return _email.encode_header(name, to_utf8(value), 'utf-8')
def _encode_address_header(name, value):
out = deque()
for addr in flanker.addresslib.address.parse_list(value):
if addr.requires_non_ascii():
encoded_addr = addr.to_unicode()
if six.PY2:
encoded_addr = encoded_addr.encode('utf-8')
else:
encoded_addr = addr.full_spec()
out.append(encoded_addr)
return '; '.join(out)
def _encode_parametrized(key, value, params):
if params:
params = [_encode_param(key, n, v) for n, v in six.iteritems(params)]
return value + '; ' + ('; '.join(params))
return value
def _encode_param(key, name, value):
try:
if six.PY2:
value = value.encode('ascii')
return _email.format_param(name, value)
except Exception:
value = value.encode('utf-8')
encoded_param = _email.encode_header(key, value, 'utf-8')
return _email.format_param(name, encoded_param)
def _is_address_header(key, val):
return key in _ADDRESS_HEADERS and '@' in val
|
