Python教程

Python 异步发送邮件脚本

本文主要是介绍Python 异步发送邮件脚本，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！ 
import asyncio
import os
import time
from email.header import Header
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import List
from typing import Union

import aiosmtplib
from tenacity import RetryCallState
from tenacity import retry
from tenacity import stop_after_attempt
from tenacity import wait_fixed


def retry_error_callback(state: RetryCallState):
    e = state.outcome.exception()
    print(f"发送邮件失败, e:{e}")
    return None


class EmailTool:

    def __init__(self, host, port, send_user, pwd, received_users, cc: Union[str, List[str]], timeout):
        """
        :param host: SMTP服务器
        :param port: SMTP服务器端口
        :param send_user: 发送者邮箱
        :param pwd: 发送者邮箱秘钥
        :param received_users: 接收者
        :param cc: 抄送者
        :param timeout: 超时时间,单位秒
        """
        self.received_users = self.cover(received_users)
        self.timeout = timeout
        self.pwd = pwd
        self.user = send_user
        self.port = port
        self.host = host
        self.cc = self.cover(cc)

    @staticmethod
    def cover(v: Union[str, List[str]]) -> List[str]:
        if type(v) == str:
            return [v]
        else:
            return v

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(2), reraise=True, retry_error_callback=retry_error_callback)
    async def start_smtp(self, msg):
        async with aiosmtplib.SMTP(hostname=self.host,
                                   port=self.port,
                                   use_tls=False,
                                   validate_certs=False,
                                   timeout=self.timeout
                                   ) as smtp:
            await smtp.starttls()
            await smtp.login(self.user, self.pwd)
            await smtp.send_message(msg)
            print("发送邮件成功!")
        return smtp

    @staticmethod
    def __email_format_type(text, subtype, charset, policy=None):
        return MIMEText(text, subtype, charset, policy=policy)

    def html_type(self, text, charset='utf-8', policy=None):
        return self.__email_format_type(text, 'html', charset, policy)

    def text_type(self, text, charset='utf-8', policy=None):
        return self.__email_format_type(text, 'plain', charset, policy)

    @staticmethod
    def file_type(filename, mode='rb', content_type='application/octet-stream'):
        filename = os.path.abspath(filename)
        if os.path.isfile(filename):
            basename = os.path.basename(filename)
            with open(filename, mode) as f:
                content = f.read()
            application = MIMEApplication(content)
            application["Content-Type"] = content_type
            application.add_header('Content-Disposition', 'attachment', filename=basename)
        else:
            application = None
            print(f'未找到文件, path:{filename}')
        return application

    def joint_send_message(self, header, attachments, charset='utf-8'):
        msg = MIMEMultipart()
        msg['From'] = self.user
        msg['To'] = ', '.join(self.received_users)
        msg['Subject'] = Header(header, charset)
        msg['Cc'] = ', '.join(self.cc)
        for attachment in attachments:
            msg.attach(attachment)
        return msg


async def debug():
    obj = EmailTool(host='smtp.qq.com',
                    port=587, send_user='1234765482@qq.com', pwd='jlkjlKKK#s,',
                    received_users=['39876409@qq.com', '1150646501@qq.com'], cc=[], timeout=30)
    a2 = obj.text_type("---***---\n")
    a3 = obj.html_type("""<html>
  <head></head>
  <body>
    <p>嘿嘿,我是html文本</p>
    <img src="https://www.google.com.hk/imgres?imgurl=https%3A%2F%2Fimg95.699pic.com%2Fphoto%2F40094%2F7630.jpg_wh300.jpg&imgrefurl=https%3A%2F%2F699pic.com%2Ftupian%2Fwennuan.html&tbnid=gMhgKqwRR3MKgM&vet=12ahUKEwjuxZ_ioc_2AhUI6pQKHT7dAA0QMygCegUIARDVAQ..i&docid=hkhgCagkDd5CuM&w=540&h=300&q=%E5%9B%BE%E7%89%87&ved=2ahUKEwjuxZ_ioc_2AhUI6pQKHT7dAA0QMygCegUIARDVAQ" />
  </body>
</html>""")
    a4 = obj.file_type("demo.py")
    msg = obj.joint_send_message(f"这是测试标题:{time.time()}", [a3, a2, a1, a4])
    await  obj.start_smtp(msg)


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(debug())
备注

①host是SMPT服务器的地址，常用的比如qq是smtp.qq.com，office365是smtp.office365.com

运行

image-20220318164559431

这篇关于Python 异步发送邮件脚本的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
栏目导航