NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

header.py (9915B)


      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\email\header.py
      2 __all__ = ['Header', 'decode_header', 'make_header']
      3 import re
      4 import binascii
      5 import email.quoprimime
      6 import email.base64mime
      7 from email.errors import HeaderParseError
      8 from email.charset import Charset
      9 NL = '\n'
     10 SPACE = ' '
     11 USPACE = u' '
     12 SPACE8 = ' ' * 8
     13 UEMPTYSTRING = u''
     14 MAXLINELEN = 76
     15 USASCII = Charset('us-ascii')
     16 UTF8 = Charset('utf-8')
     17 ecre = re.compile('\n  =\\?                   # literal =?\n  (?P<charset>[^?]*?)   # non-greedy up to the next ? is the charset\n  \\?                    # literal ?\n  (?P<encoding>[qb])    # either a "q" or a "b", case insensitive\n  \\?                    # literal ?\n  (?P<encoded>.*?)      # non-greedy up to the next ?= is the encoded string\n  \\?=                   # literal ?=\n  (?=[ \\t]|$)           # whitespace or the end of the string\n  ', re.VERBOSE | re.IGNORECASE | re.MULTILINE)
     18 fcre = re.compile('[\\041-\\176]+:$')
     19 _max_append = email.quoprimime._max_append
     20 
     21 def decode_header(header):
     22     header = str(header)
     23     if not ecre.search(header):
     24         return [(header, None)]
     25     decoded = []
     26     dec = ''
     27     for line in header.splitlines():
     28         if not ecre.search(line):
     29             decoded.append((line, None))
     30             continue
     31         parts = ecre.split(line)
     32         while parts:
     33             unenc = parts.pop(0).strip()
     34             if unenc:
     35                 if decoded and decoded[-1][1] is None:
     36                     decoded[-1] = (decoded[-1][0] + SPACE + unenc, None)
     37                 else:
     38                     decoded.append((unenc, None))
     39             if parts:
     40                 charset, encoding = [ s.lower() for s in parts[0:2] ]
     41                 encoded = parts[2]
     42                 dec = None
     43                 if encoding == 'q':
     44                     dec = email.quoprimime.header_decode(encoded)
     45                 elif encoding == 'b':
     46                     paderr = len(encoded) % 4
     47                     if paderr:
     48                         encoded += '==='[:4 - paderr]
     49                     try:
     50                         dec = email.base64mime.decode(encoded)
     51                     except binascii.Error:
     52                         raise HeaderParseError
     53 
     54                 if dec is None:
     55                     dec = encoded
     56                 if decoded and decoded[-1][1] == charset:
     57                     decoded[-1] = (decoded[-1][0] + dec, decoded[-1][1])
     58                 else:
     59                     decoded.append((dec, charset))
     60             del parts[0:3]
     61 
     62     return decoded
     63 
     64 
     65 def make_header(decoded_seq, maxlinelen = None, header_name = None, continuation_ws = ' '):
     66     h = Header(maxlinelen=maxlinelen, header_name=header_name, continuation_ws=continuation_ws)
     67     for s, charset in decoded_seq:
     68         if charset is not None and not isinstance(charset, Charset):
     69             charset = Charset(charset)
     70         h.append(s, charset)
     71 
     72     return h
     73 
     74 
     75 class Header():
     76 
     77     def __init__(self, s = None, charset = None, maxlinelen = None, header_name = None, continuation_ws = ' ', errors = 'strict'):
     78         if charset is None:
     79             charset = USASCII
     80         if not isinstance(charset, Charset):
     81             charset = Charset(charset)
     82         self._charset = charset
     83         self._continuation_ws = continuation_ws
     84         cws_expanded_len = len(continuation_ws.replace('\t', SPACE8))
     85         self._chunks = []
     86         if s is not None:
     87             self.append(s, charset, errors)
     88         if maxlinelen is None:
     89             maxlinelen = MAXLINELEN
     90         if header_name is None:
     91             self._firstlinelen = maxlinelen
     92         else:
     93             self._firstlinelen = maxlinelen - len(header_name) - 2
     94         self._maxlinelen = maxlinelen - cws_expanded_len
     95 
     96     def __str__(self):
     97         return self.encode()
     98 
     99     def __unicode__(self):
    100         uchunks = []
    101         lastcs = None
    102         for s, charset in self._chunks:
    103             nextcs = charset
    104             if uchunks:
    105                 if lastcs not in (None, 'us-ascii'):
    106                     if nextcs in (None, 'us-ascii'):
    107                         uchunks.append(USPACE)
    108                         nextcs = None
    109                 elif nextcs not in (None, 'us-ascii'):
    110                     uchunks.append(USPACE)
    111             lastcs = nextcs
    112             uchunks.append(unicode(s, str(charset)))
    113 
    114         return UEMPTYSTRING.join(uchunks)
    115 
    116     def __eq__(self, other):
    117         return other == self.encode()
    118 
    119     def __ne__(self, other):
    120         return not self == other
    121 
    122     def append(self, s, charset = None, errors = 'strict'):
    123         if charset is None:
    124             charset = self._charset
    125         elif not isinstance(charset, Charset):
    126             charset = Charset(charset)
    127         if charset != '8bit':
    128             if isinstance(s, str):
    129                 incodec = charset.input_codec or 'us-ascii'
    130                 ustr = unicode(s, incodec, errors)
    131                 outcodec = charset.output_codec or 'us-ascii'
    132                 ustr.encode(outcodec, errors)
    133             elif isinstance(s, unicode):
    134                 for charset in (USASCII, charset, UTF8):
    135                     try:
    136                         outcodec = charset.output_codec or 'us-ascii'
    137                         s = s.encode(outcodec, errors)
    138                         break
    139                     except UnicodeError:
    140                         pass
    141 
    142         self._chunks.append((s, charset))
    143 
    144     def _split(self, s, charset, maxlinelen, splitchars):
    145         splittable = charset.to_splittable(s)
    146         encoded = charset.from_splittable(splittable, True)
    147         elen = charset.encoded_header_len(encoded)
    148         if elen <= maxlinelen:
    149             return [(encoded, charset)]
    150         if charset == '8bit':
    151             return [(s, charset)]
    152         if charset == 'us-ascii':
    153             return self._split_ascii(s, charset, maxlinelen, splitchars)
    154         if elen == len(s):
    155             splitpnt = maxlinelen
    156             first = charset.from_splittable(splittable[:splitpnt], False)
    157             last = charset.from_splittable(splittable[splitpnt:], False)
    158         else:
    159             first, last = _binsplit(splittable, charset, maxlinelen)
    160         fsplittable = charset.to_splittable(first)
    161         fencoded = charset.from_splittable(fsplittable, True)
    162         chunk = [(fencoded, charset)]
    163         return chunk + self._split(last, charset, self._maxlinelen, splitchars)
    164 
    165     def _split_ascii(self, s, charset, firstlen, splitchars):
    166         chunks = _split_ascii(s, firstlen, self._maxlinelen, self._continuation_ws, splitchars)
    167         return zip(chunks, [charset] * len(chunks))
    168 
    169     def _encode_chunks(self, newchunks, maxlinelen):
    170         chunks = []
    171         for header, charset in newchunks:
    172             if not header:
    173                 continue
    174             if charset is None or charset.header_encoding is None:
    175                 s = header
    176             else:
    177                 s = charset.header_encode(header)
    178             if chunks and chunks[-1].endswith(' '):
    179                 extra = ''
    180             else:
    181                 extra = ' '
    182             _max_append(chunks, s, maxlinelen, extra)
    183 
    184         joiner = NL + self._continuation_ws
    185         return joiner.join(chunks)
    186 
    187     def encode(self, splitchars = ';, '):
    188         newchunks = []
    189         maxlinelen = self._firstlinelen
    190         lastlen = 0
    191         for s, charset in self._chunks:
    192             targetlen = maxlinelen - lastlen - 1
    193             if targetlen < charset.encoded_header_len(''):
    194                 targetlen = maxlinelen
    195             newchunks += self._split(s, charset, targetlen, splitchars)
    196             lastchunk, lastcharset = newchunks[-1]
    197             lastlen = lastcharset.encoded_header_len(lastchunk)
    198 
    199         return self._encode_chunks(newchunks, maxlinelen)
    200 
    201 
    202 def _split_ascii(s, firstlen, restlen, continuation_ws, splitchars):
    203     lines = []
    204     maxlen = firstlen
    205     for line in s.splitlines():
    206         line = line.lstrip()
    207         if len(line) < maxlen:
    208             lines.append(line)
    209             maxlen = restlen
    210             continue
    211         for ch in splitchars:
    212             if ch in line:
    213                 break
    214         else:
    215             lines.append(line)
    216             maxlen = restlen
    217             continue
    218 
    219         cre = re.compile('%s\\s*' % ch)
    220         if ch in ';,':
    221             eol = ch
    222         else:
    223             eol = ''
    224         joiner = eol + ' '
    225         joinlen = len(joiner)
    226         wslen = len(continuation_ws.replace('\t', SPACE8))
    227         this = []
    228         linelen = 0
    229         for part in cre.split(line):
    230             curlen = linelen + max(0, len(this) - 1) * joinlen
    231             partlen = len(part)
    232             onfirstline = not lines
    233             if ch == ' ' and onfirstline and len(this) == 1 and fcre.match(this[0]):
    234                 this.append(part)
    235                 linelen += partlen
    236             elif curlen + partlen > maxlen:
    237                 if this:
    238                     lines.append(joiner.join(this) + eol)
    239                 if partlen > maxlen and ch != ' ':
    240                     subl = _split_ascii(part, maxlen, restlen, continuation_ws, ' ')
    241                     lines.extend(subl[:-1])
    242                     this = [subl[-1]]
    243                 else:
    244                     this = [part]
    245                 linelen = wslen + len(this[-1])
    246                 maxlen = restlen
    247             else:
    248                 this.append(part)
    249                 linelen += partlen
    250 
    251         if this:
    252             lines.append(joiner.join(this))
    253 
    254     return lines
    255 
    256 
    257 def _binsplit(splittable, charset, maxlinelen):
    258     i = 0
    259     j = len(splittable)
    260     while i < j:
    261         m = i + j + 1 >> 1
    262         chunk = charset.from_splittable(splittable[:m], True)
    263         chunklen = charset.encoded_header_len(chunk)
    264         if chunklen <= maxlinelen:
    265             i = m
    266         else:
    267             j = m - 1
    268 
    269     first = charset.from_splittable(splittable[:i], False)
    270     last = charset.from_splittable(splittable[i:], False)
    271     return (first, last)