python类MIMEMultipart()的实例源码

def as_mailobj(self):
        # Create the root message and fill in the from, to, and subject headers
        msgRoot = MIMEMultipart('related')
        msgRoot['Subject'] = self.subject
        msgRoot['From'] = self.get_sender()
        msgRoot['To'] = ", ".join(self.recipients)
        msgRoot.preamble = 'Generated by mailprod python tool'
        for i in range(len(self.headers)):
            name, value, params = self.headers[i]
            msgRoot.add_header(name, value, **params)
        for i in range(len(self.body)):
            part = self.body[i]
            msgRoot.attach(part)
        for i in range(len(self.attachments)):
            attachment = self.attachments[i]
            msgRoot.attach(attachment)
        return msgRoot

def mime_multipart(_, context, arg):
    """mime_multipart composes a MIME multipart string.

    Example:
        "UserData": {
          "Fn::Base64": {
            "CFPP::MimeMultipart": [
              ["text/x-shellscript", {"CFPP::FileToString": "userdata.sh"}],
              ["text/cloud-config", {"CFPP::FileToString": "cloud-init.yaml"}]
            ]
          }
        }
    """
    _raise_unless_mime_params(context, arg)
    mime_doc = MIMEMultipart()
    # set boundary explicitly so that they are stable based on path in the template.
    mime_doc.set_boundary("=" * 10 + hashlib.sha1(".".join(context)).hexdigest() + "=" * 3)
    for mime_type, contents in arg:
        sub_message = MIMEText(contents, contents, sys.getdefaultencoding())
        mime_doc.attach(sub_message)
    return mime_doc.as_string()

def test_mime_attachments_in_constructor(self):
        eq = self.assertEqual
        text1 = MIMEText('')
        text2 = MIMEText('')
        msg = MIMEMultipart(_subparts=(text1, text2))
        eq(len(msg.get_payload()), 2)
        eq(msg.get_payload(0), text1)
        eq(msg.get_payload(1), text2)



# A general test of parser->model->generator idempotency.  IOW, read a message
# in, parse it into a message object tree, then without touching the tree,
# regenerate the plain text.  The original text and the transformed text
# should be identical.  Note: that we ignore the Unix-From since that may
# contain a changed date.

def test__all__(self):
        module = __import__('email')
        # Can't use sorted() here due to Python 2.3 compatibility
        all = module.__all__[:]
        all.sort()
        self.assertEqual(all, [
            # Old names
            'Charset', 'Encoders', 'Errors', 'Generator',
            'Header', 'Iterators', 'MIMEAudio', 'MIMEBase',
            'MIMEImage', 'MIMEMessage', 'MIMEMultipart',
            'MIMENonMultipart', 'MIMEText', 'Message',
            'Parser', 'Utils', 'base64MIME',
            # new names
            'base64mime', 'charset', 'encoders', 'errors', 'generator',
            'header', 'iterators', 'message', 'message_from_file',
            'message_from_string', 'mime', 'parser',
            'quopriMIME', 'quoprimime', 'utils',
            ])

def get_mime_message(text, html_text=None, **kwargs):
    if not html_text:
        instance = MIMEText(text)
    else:
        instance = MIMEMultipart('alternative')
        instance.attach(MIMEText(text, 'plain'))
        instance.attach(MIMEText(html_text, 'html'))
        extra = MIMEBase('application', 'octet-stream')
        extra.set_payload(b'test content')
        encoders.encode_base64(extra)
        extra.add_header('Content-Disposition', 'attachment', filename='report.pdf')
        instance.attach(extra)
        instance['X-Accept-Language'] = 'en-us, en'
    for key, value in kwargs.items():
        instance[key] = value
    return instance

def from_mime(cls, message, manager):
        """
        Instantiates ``Email`` instance from ``MIMEText`` instance.

        :param message: ``email.mime.text.MIMEText`` instance.
        :param manager: :py:class:`EmailManager` instance.
        :return: :py:class:`Email`
        """
        if isinstance(message, MIMEMultipart):
            text, html, attachments = deconstruct_multipart(message)
        else:
            text, html, attachments = message.get_payload(decode=True).decode('utf8'), None, []
        subject = prepare_header(message['Subject'])
        sender = prepare_header(message['From'])
        to = prepare_header(message['To'])
        cc = prepare_header(message['Cc'])
        bcc = prepare_header(message['Bcc'])
        reply_to = prepare_header(message['Reply-To'])
        tag = getattr(message, 'tag', None)
        return cls(manager=manager, From=sender, To=to, TextBody=text, HtmlBody=html, Subject=subject, Cc=cc, Bcc=bcc,
                   ReplyTo=reply_to, Attachments=attachments, Tag=tag)

def test_mime_attachments_in_constructor(self):
        eq = self.assertEqual
        text1 = MIMEText('')
        text2 = MIMEText('')
        msg = MIMEMultipart(_subparts=(text1, text2))
        eq(len(msg.get_payload()), 2)
        eq(msg.get_payload(0), text1)
        eq(msg.get_payload(1), text2)



# A general test of parser->model->generator idempotency.  IOW, read a message
# in, parse it into a message object tree, then without touching the tree,
# regenerate the plain text.  The original text and the transformed text
# should be identical.  Note: that we ignore the Unix-From since that may
# contain a changed date.

def test__all__(self):
        module = __import__('email')
        # Can't use sorted() here due to Python 2.3 compatibility
        all = module.__all__[:]
        all.sort()
        self.assertEqual(all, [
            # Old names
            'Charset', 'Encoders', 'Errors', 'Generator',
            'Header', 'Iterators', 'MIMEAudio', 'MIMEBase',
            'MIMEImage', 'MIMEMessage', 'MIMEMultipart',
            'MIMENonMultipart', 'MIMEText', 'Message',
            'Parser', 'Utils', 'base64MIME',
            # new names
            'base64mime', 'charset', 'encoders', 'errors', 'generator',
            'header', 'iterators', 'message', 'message_from_file',
            'message_from_string', 'mime', 'parser',
            'quopriMIME', 'quoprimime', 'utils',
            ])

def __init__(self, server, sender, password, receiver, title, message=None, path=None):
        """???Email

        :param title: ????????
        :param message: ?????????
        :param path: ????????list??????str???????????
        :param server: smtp???????
        :param sender: ???????
        :param password: ?????????
        :param receiver: ?????????“?”??????
        """
        self.title = title
        self.message = message
        self.files = path

        self.msg = MIMEMultipart('related')

        self.server = server
        self.sender = sender
        self.receiver = receiver
        self.password = password

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_email(self):
        self.make_messages()
        if len(self.email_messages) > 0:
            for detail in self.email_messages:
                user_email = detail['email']
                user_message = detail['message']
                try:
                    email_conn = smtplib.SMTP(host, port)
                    email_conn.ehlo()
                    email_conn.starttls()
                    email_conn.login(username, password)
                    the_msg = MIMEMultipart("alternative")
                    the_msg['Subject'] = "Billing Update!"
                    the_msg["From"] = from_email
                    the_msg["To"]  = user_email
                    part_1 = MIMEText(user_message, 'plain')
                    the_msg.attach(part_1)
                    email_conn.sendmail(from_email, [user_email], the_msg.as_string())
                    email_conn.quit()
                except smtplib.SMTPException:
                    print("error sending message")
            return True
        return False

def send_exception_mail(exception):
    # noinspection PyBroadException
    try:
        msg = MIMEMultipart('alternative')
        msg['Subject'] = "Exception Occurred on betaPika"
        msg['From'] = FROM_MAIL
        msg['To'] = TO_MAIL
        plain = MIMEText(exception, "plain", "utf-8")
        msg.attach(plain)

        s = smtplib.SMTP()
        s.connect(SERVER, PORT)
        s.ehlo()
        s.starttls()
        s.ehlo()
        s.login(FROM_MAIL, FROM_PASSWORD)
        s.sendmail(FROM_MAIL, TO_MAIL, msg.as_string())
        s.quit()
    except:
        print "Mail Failed"

def send_txt(self, filename):
        # ????????????????????????????????????
        self.smtp = smtplib.SMTP_SSL(port=465)
        self.smtp.connect(self.server)
        self.smtp.login(self.username, self.password)
        #self.msg = MIMEMultipart()
        self.msg = MIMEText("content", 'plain', 'utf-8')
        self.msg['to'] = self.to_mail
        self.msg['from'] = self.from_mail
        self.msg['Subject'] = "459"
        self.filename = filename + ".txt"
        self.msg['Date'] = Utils.formatdate(localtime=1)
        content = open(self.filename.decode('utf-8'), 'rb').read()
        # print content
        #self.att = MIMEText(content, 'base64', 'utf-8')
        #self.att['Content-Type'] = 'application/octet-stream'
        # self.att["Content-Disposition"] = "attachment;filename=\"%s\"" %(self.filename.encode('gb2312'))
        #self.att["Content-Disposition"] = "attachment;filename=\"%s\"" % Header(self.filename, 'gb2312')
        # print self.att["Content-Disposition"]
        #self.msg.attach(self.att)

        #self.smtp.sendmail(self.msg['from'], self.msg['to'], self.msg.as_string())
        self.smtp.sendmail(self.msg['from'], self.msg['to'], self.msg.as_string())
        self.smtp.quit()

def send_mail(self, sender, receiver, subject="", body=""):
        """Send email from sender to receiver
        """
        mime_msg = MIMEMultipart('related')
        mime_msg['Subject'] = subject
        mime_msg['From'] = sender
        mime_msg['To'] = receiver
        msg_txt = MIMEText(body, 'plain')
        mime_msg.attach(msg_txt)
        try:
            host = getToolByName(self, 'MailHost')
            host.send(mime_msg.as_string(), immediate=True)
        except SMTPServerDisconnected as msg:
            logger.warn("SMTPServerDisconnected: %s." % msg)
        except SMTPRecipientsRefused as msg:
            raise WorkflowException(str(msg))

def test_mime_attachments_in_constructor(self):
        eq = self.assertEqual
        text1 = MIMEText('')
        text2 = MIMEText('')
        msg = MIMEMultipart(_subparts=(text1, text2))
        eq(len(msg.get_payload()), 2)
        eq(msg.get_payload(0), text1)
        eq(msg.get_payload(1), text2)



# A general test of parser->model->generator idempotency.  IOW, read a message
# in, parse it into a message object tree, then without touching the tree,
# regenerate the plain text.  The original text and the transformed text
# should be identical.  Note: that we ignore the Unix-From since that may
# contain a changed date.

def test__all__(self):
        module = __import__('email')
        # Can't use sorted() here due to Python 2.3 compatibility
        all = module.__all__[:]
        all.sort()
        self.assertEqual(all, [
            # Old names
            'Charset', 'Encoders', 'Errors', 'Generator',
            'Header', 'Iterators', 'MIMEAudio', 'MIMEBase',
            'MIMEImage', 'MIMEMessage', 'MIMEMultipart',
            'MIMENonMultipart', 'MIMEText', 'Message',
            'Parser', 'Utils', 'base64MIME',
            # new names
            'base64mime', 'charset', 'encoders', 'errors', 'generator',
            'header', 'iterators', 'message', 'message_from_file',
            'message_from_string', 'mime', 'parser',
            'quopriMIME', 'quoprimime', 'utils',
            ])

def testExtractAttachedKey(self):
        KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."

        message = MIMEMultipart()
        message.add_header("from", ADDRESS_2)
        key = MIMEApplication("", "pgp-keys")
        key.set_payload(KEY)
        message.attach(key)
        self.fetcher._keymanager.put_raw_key = Mock(
            return_value=defer.succeed(None))

        def put_raw_key_called(_):
            self.fetcher._keymanager.put_raw_key.assert_called_once_with(
                KEY, address=ADDRESS_2)

        d = self._do_fetch(message.as_string())
        d.addCallback(put_raw_key_called)
        return d

def testExtractInvalidAttachedKey(self):
        KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."

        message = MIMEMultipart()
        message.add_header("from", ADDRESS_2)
        key = MIMEApplication("", "pgp-keys")
        key.set_payload(KEY)
        message.attach(key)
        self.fetcher._keymanager.put_raw_key = Mock(
            return_value=defer.fail(KeyAddressMismatch()))

        def put_raw_key_called(_):
            self.fetcher._keymanager.put_raw_key.assert_called_once_with(
                KEY, address=ADDRESS_2)

        d = self._do_fetch(message.as_string())
        d.addCallback(put_raw_key_called)
        d.addErrback(log.err)
        return d

def testExtractAttachedKeyAndNotOpenPGPHeader(self):
        KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."
        KEYURL = "https://leap.se/key.txt"
        OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,)

        message = MIMEMultipart()
        message.add_header("from", ADDRESS_2)
        message.add_header("OpenPGP", OpenPGP)
        key = MIMEApplication("", "pgp-keys")
        key.set_payload(KEY)
        message.attach(key)

        self.fetcher._keymanager.put_raw_key = Mock(
            return_value=defer.succeed(None))
        self.fetcher._keymanager.fetch_key = Mock()

        def put_raw_key_called(_):
            self.fetcher._keymanager.put_raw_key.assert_called_once_with(
                KEY, address=ADDRESS_2)
            self.assertFalse(self.fetcher._keymanager.fetch_key.called)

        d = self._do_fetch(message.as_string())
        d.addCallback(put_raw_key_called)
        return d

def testExtractOpenPGPHeaderIfInvalidAttachedKey(self):
        KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."
        KEYURL = "https://leap.se/key.txt"
        OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,)

        message = MIMEMultipart()
        message.add_header("from", ADDRESS_2)
        message.add_header("OpenPGP", OpenPGP)
        key = MIMEApplication("", "pgp-keys")
        key.set_payload(KEY)
        message.attach(key)

        self.fetcher._keymanager.put_raw_key = Mock(
            return_value=defer.fail(KeyAddressMismatch()))
        self.fetcher._keymanager.fetch_key = Mock()

        def put_raw_key_called(_):
            self.fetcher._keymanager.put_raw_key.assert_called_once_with(
                KEY, address=ADDRESS_2)
            self.fetcher._keymanager.fetch_key.assert_called_once_with(
                ADDRESS_2, KEYURL)

        d = self._do_fetch(message.as_string())
        d.addCallback(put_raw_key_called)
        return d

def mail_client(host, port, fromAddress, password, toAddress, subject, body):
    msg = MIMEMultipart()

    msg['From'] = fromAddress
    msg['To'] = toAddress
    msg['Subject'] = subject
    message = body
    msg.attach(MIMEText(message))

    mailserver = smtplib.SMTP(host,port)

    # Identify to the SMTP Gmail Client
    mailserver.ehlo()

    # Secure with TLS Encryption
    mailserver.starttls()

    # Reidentifying as an encrypted connection
    mailserver.ehlo()
    mailserver.login(fromAddress, password)

    mailserver.sendmail(fromAddress,toAddress,msg.as_string())
    print ("Email sent from:", fromAddress)

    mailserver.quit()

def test_mime_attachments_in_constructor(self):
        eq = self.assertEqual
        text1 = MIMEText('')
        text2 = MIMEText('')
        msg = MIMEMultipart(_subparts=(text1, text2))
        eq(len(msg.get_payload()), 2)
        eq(msg.get_payload(0), text1)
        eq(msg.get_payload(1), text2)



# A general test of parser->model->generator idempotency.  IOW, read a message
# in, parse it into a message object tree, then without touching the tree,
# regenerate the plain text.  The original text and the transformed text
# should be identical.  Note: that we ignore the Unix-From since that may
# contain a changed date.

def test__all__(self):
        module = __import__('email')
        # Can't use sorted() here due to Python 2.3 compatibility
        all = module.__all__[:]
        all.sort()
        self.assertEqual(all, [
            # Old names
            'Charset', 'Encoders', 'Errors', 'Generator',
            'Header', 'Iterators', 'MIMEAudio', 'MIMEBase',
            'MIMEImage', 'MIMEMessage', 'MIMEMultipart',
            'MIMENonMultipart', 'MIMEText', 'Message',
            'Parser', 'Utils', 'base64MIME',
            # new names
            'base64mime', 'charset', 'encoders', 'errors', 'generator',
            'header', 'iterators', 'message', 'message_from_file',
            'message_from_string', 'mime', 'parser',
            'quopriMIME', 'quoprimime', 'utils',
            ])

def sendreport(self, body):
        """
        Send a notification by email

        :param body:    Body of the mail
        :return:
        """
        smtp = self.config['smtp']

        msg = multipart.MIMEMultipart()
        msg['From'] = smtp['from']
        msg['To'] = smtp['to']
        msg['Subject'] = smtp['subject']

        msg.attach(text.MIMEText(body, 'plain'))

        server = smtplib.SMTP(smtp['host'], smtp['port'])
        server.starttls()
        server.login(smtp['user'], smtp['password'])

        server.sendmail(smtp['from'], smtp['to'], msg.as_string())
        server.quit()