Lib/mimetools.py

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