twisted/mail/bounce.py

   1 # -*- test-case-name: twisted.mail.test.test_bounce -*-
   2 #
   3 # Copyright (c) Twisted Matrix Laboratories.
   4 # See LICENSE for details.
   5
   6
   7 import StringIO
   8 import rfc822
   9 import time
  10 import os
  11
  12
  13 from twisted.mail import smtp
  14
  15 BOUNCE_FORMAT = """\
  16 From: postmaster@%(failedDomain)s
  17 To: %(failedFrom)s
  18 Subject: Returned Mail: see transcript for details
  19 Message-ID: %(messageID)s
  20 Content-Type: multipart/report; report-type=delivery-status;
  21     boundary="%(boundary)s"
  22
  23 --%(boundary)s
  24
  25 %(transcript)s
  26
  27 --%(boundary)s
  28 Content-Type: message/delivery-status
  29 Arrival-Date: %(ctime)s
  30 Final-Recipient: RFC822; %(failedTo)s
  31 """
  32
  33 def generateBounce(message, failedFrom, failedTo, transcript=''):
  34     if not transcript:
  35         transcript = '''\
  36 I'm sorry, the following address has permanent errors: %(failedTo)s.
  37 I've given up, and I will not retry the message again.
  38 ''' % vars()
  39
  40     boundary = "%s_%s_%s" % (time.time(), os.getpid(), 'XXXXX')
  41     failedAddress = rfc822.AddressList(failedTo)[0][1]
  42     failedDomain = failedAddress.split('@', 1)[1]
  43     messageID = smtp.messageid(uniq='bounce')
  44     ctime = time.ctime(time.time())
  45
  46     fp = StringIO.StringIO()
  47     fp.write(BOUNCE_FORMAT % vars())
  48     orig = message.tell()
  49     message.seek(2, 0)
  50     sz = message.tell()
  51     message.seek(0, orig)
  52     if sz > 10000:
  53         while 1:
  54             line = message.readline()
  55             if len(line)<=1:
  56                 break
  57             fp.write(line)
  58     else:
  59         fp.write(message.read())
  60     return '', failedFrom, fp.getvalue()