Lib/mimetools.py

   1 """Various tools used by MIME-reading or MIME-writing programs."""
   2
   3
   4 import os
   5 import sys
   6 import tempfile
   7 from warnings import filterwarnings, catch_warnings
   8 with catch_warnings():
   9     if sys.py3kwarning:
  10         filterwarnings("ignore", ".*rfc822 has been removed", DeprecationWarning)
  11     import rfc822
  12
  13 from warnings import warnpy3k
  14 warnpy3k("in 3.x, mimetools has been removed in favor of the email package",
  15          stacklevel=2)
  16
  17 __all__ = ["Message","choose_boundary","encode","decode","copyliteral",
  18            "copybinary"]
  19
  20 class Message(rfc822.Message):
  21     """A derived class of rfc822.Message that knows about MIME headers and
  22     contains some hooks for decoding encoded and multipart messages."""
  23
  24     def __init__(self, fp, seekable = 1):
  25         rfc822.Message.__init__(self, fp, seekable)
  26         self.encodingheader = \
  27                 self.getheader('content-transfer-encoding')
  28         self.typeheader = \
  29                 self.getheader('content-type')
  30         self.parsetype()
  31         self.parseplist()
  32
  33     def parsetype(self):
  34         str = self.typeheader
  35         if str is None:
  36             str = 'text/plain'
  37         if ';' in str:
  38             i = str.index(';')
  39             self.plisttext = str[i:]
  40             str = str[:i]
  41         else:
  42             self.plisttext = ''
  43         fields = str.split('/')
  44         for i in range(len(fields)):
  45             fields[i] = fields[i].strip().lower()
  46         self.type = '/'.join(fields)
  47         self.maintype = fields[0]
  48         self.subtype = '/'.join(fields[1:])
  49
  50     def parseplist(self):
  51         str = self.plisttext
  52         self.plist = []
  53         while str[:1] == ';':
  54             str = str[1:]
  55             if ';' in str:
  56                 # XXX Should parse quotes!
  57                 end = str.index(';')
  58             else:
  59                 end = len(str)
  60             f = str[:end]
  61             if '=' in f:
  62                 i = f.index('=')
  63                 f = f[:i].strip().lower() + \
  64                         '=' + f[i+1:].strip()
  65             self.plist.append(f.strip())
  66             str = str[end:]
  67
  68     def getplist(self):
  69         return self.plist
  70
  71     def getparam(self, name):
  72         name = name.lower() + '='
  73         n = len(name)
  74         for p in self.plist:
  75             if p[:n] == name:
  76                 return rfc822.unquote(p[n:])
  77         return None
  78
  79     def getparamnames(self):
  80         result = []
  81         for p in self.plist:
  82             i = p.find('=')
  83             if i >= 0:
  84                 result.append(p[:i].lower())
  85         return result
  86
  87     def getencoding(self):
  88         if self.encodingheader is None:
  89             return '7bit'
  90         return self.encodingheader.lower()
  91
  92     def gettype(self):
  93         return self.type
  94
  95     def getmaintype(self):
  96         return self.maintype
  97
  98     def getsubtype(self):
  99         return self.subtype
 100
 101
 102
 103
 104 # Utility functions
 105 # -----------------
 106
 107 try:
 108     import thread
 109 except ImportError:
 110     import dummy_thread as thread
 111 _counter_lock = thread.allocate_lock()
 112 del thread
 113
 114 _counter = 0
 115 def _get_next_counter():
 116     global _counter
 117     _counter_lock.acquire()
 118     _counter += 1
 119     result = _counter
 120     _counter_lock.release()
 121     return result
 122
 123 _prefix = None
 124
 125 def choose_boundary():
 126     """Return a string usable as a multipart boundary.
 127
 128     The string chosen is unique within a single program run, and
 129     incorporates the user id (if available), process id (if available),
 130     and current time.  So it's very unlikely the returned string appears
 131     in message text, but there's no guarantee.
 132
 133     The boundary contains dots so you have to quote it in the header."""
 134
 135     global _prefix
 136     import time
 137     if _prefix is None:
 138         import socket
 139         try:
 140             hostid = socket.gethostbyname(socket.gethostname())
 141         except socket.gaierror:
 142             hostid = '127.0.0.1'
 143         try:
 144             uid = repr(os.getuid())
 145         except AttributeError:
 146             uid = '1'
 147         try:
 148             pid = repr(os.getpid())
 149         except AttributeError:
 150             pid = '1'
 151         _prefix = hostid + '.' + uid + '.' + pid
 152     return "%s.%.3f.%d" % (_prefix, time.time(), _get_next_counter())
 153
 154
 155 # Subroutines for decoding some common content-transfer-types
 156
 157 def decode(input, output, encoding):
 158     """Decode common content-transfer-encodings (base64, quopri, uuencode)."""
 159     if encoding == 'base64':
 160         import base64
 161         return base64.decode(input, output)
 162     if encoding == 'quoted-printable':
 163         import quopri
 164         return quopri.decode(input, output)
 165     if encoding in ('uuencode', 'x-uuencode', 'uue', 'x-uue'):
 166         import uu
 167         return uu.decode(input, output)
 168     if encoding in ('7bit', '8bit'):
 169         return output.write(input.read())
 170     if encoding in decodetab:
 171         pipethrough(input, decodetab[encoding], output)
 172     else:
 173         raise ValueError, \
 174               'unknown Content-Transfer-Encoding: %s' % encoding
 175
 176 def encode(input, output, encoding):
 177     """Encode common content-transfer-encodings (base64, quopri, uuencode)."""
 178     if encoding == 'base64':
 179         import base64
 180         return base64.encode(input, output)
 181     if encoding == 'quoted-printable':
 182         import quopri
 183         return quopri.encode(input, output, 0)
 184     if encoding in ('uuencode', 'x-uuencode', 'uue', 'x-uue'):
 185         import uu
 186         return uu.encode(input, output)
 187     if encoding in ('7bit', '8bit'):
 188         return output.write(input.read())
 189     if encoding in encodetab:
 190         pipethrough(input, encodetab[encoding], output)
 191     else:
 192         raise ValueError, \
 193               'unknown Content-Transfer-Encoding: %s' % encoding
 194
 195 # The following is no longer used for standard encodings
 196
 197 # XXX This requires that uudecode and mmencode are in $PATH
 198
 199 uudecode_pipe = '''(
 200 TEMP=/tmp/@uu.$$
 201 sed "s%^begin [0-7][0-7]* .*%begin 600 $TEMP%" | uudecode
 202 cat $TEMP
 203 rm $TEMP
 204 )'''
 205
 206 decodetab = {
 207         'uuencode':             uudecode_pipe,
 208         'x-uuencode':           uudecode_pipe,
 209         'uue':                  uudecode_pipe,
 210         'x-uue':                uudecode_pipe,
 211         'quoted-printable':     'mmencode -u -q',
 212         'base64':               'mmencode -u -b',
 213 }
 214
 215 encodetab = {
 216         'x-uuencode':           'uuencode tempfile',
 217         'uuencode':             'uuencode tempfile',
 218         'x-uue':                'uuencode tempfile',
 219         'uue':                  'uuencode tempfile',
 220         'quoted-printable':     'mmencode -q',
 221         'base64':               'mmencode -b',
 222 }
 223
 224 def pipeto(input, command):
 225     pipe = os.popen(command, 'w')
 226     copyliteral(input, pipe)
 227     pipe.close()
 228
 229 def pipethrough(input, command, output):
 230     (fd, tempname) = tempfile.mkstemp()
 231     temp = os.fdopen(fd, 'w')
 232     copyliteral(input, temp)
 233     temp.close()
 234     pipe = os.popen(command + ' <' + tempname, 'r')
 235     copybinary(pipe, output)
 236     pipe.close()
 237     os.unlink(tempname)
 238
 239 def copyliteral(input, output):
 240     while 1:
 241         line = input.readline()
 242         if not line: break
 243         output.write(line)
 244
 245 def copybinary(input, output):
 246     BUFSIZE = 8192
 247     while 1:
 248         line = input.read(BUFSIZE)
 249         if not line: break
 250         output.write(line)