标签：outlook 协程 self emails connection key time email Python3

最近我们的服务需要批量发送邮件，先后试过了163邮箱和outlook企业邮箱，还是后者比较稳定。写完以后把代码整理成了一个脚本，如下所示，喜欢的客官可以拿去用了，有问题欢迎流言交流。

import ssl
import uuid
import time
import json
import redis
import django
import smtplib
import logging
import traceback
from random import choice
from threading import Thread
from django.conf import settings
from django.template.base import Template, Context
from email.utils import formatdate, formataddr, make_msgid
from django.core.mail.message import SafeMIMEText, sanitize_address
from asyncio import run_coroutine_threadsafe, ensure_future, gather, get_event_loop

TEMPLATES = [
    {
        'BACKEND': 'django.template.backends.django.DjangoTemplates',
        'DIRS': ['.'],
    }
]
settings.configure(TEMPLATES=TEMPLATES)
django.setup()


class Email:
    def __init__(self, title, message, sender_name, receives, charset, sender_host=''):
        '''
            sender_host是发件人邮箱地址
            填空值时收件人看到的是发件用户的邮箱账号
            填非空值时收件人看到的是填写的地址，以及由发件用户的邮箱账号代发的提示
        '''
        self.title = title
        self.message = message
        self.sender_name = sender_name
        self.receives = receives
        self.charset = charset
        self.sender_host = sender_host


class UnknownError(smtplib.SMTPException):
    def __init__(self, recipients):
        self.recipients = recipients
        self.args = (recipients,)


class ConnectionPool:
    def __init__(self, host='', port='', send_email_user='', send_email_password='', max_connections=0, use_ssl=False, connection_lifetime=60, re_helo_time=10):
        self.host = host  # 邮箱服务器的地址
        self.port = port  # 邮箱服务器的端口号
        self.send_email_user = send_email_user  # 发件用户的SMPT服务账号(收件人看到的发件地址)
        self.send_email_password = send_email_password  # 发件用户的SMPT服务账号的密码，注意是发件邮箱配置的SMPT服务的密码，不是发件邮箱登陆密码
        self.max_connections = max_connections  # 一个IP地址能够同时建立的连接数(连接池的大小)，163邮箱为10个，outlook邮箱为20个
        self.use_ssl = use_ssl  # smtp服务是否开启了ssl验证
        self.connection_lifetime = connection_lifetime  # 连接的存活时间，到达这个时间后就替换掉该连接，一般不用配置
        self.re_helo_time = re_helo_time  # 连接的心跳时间间隔，每隔一定时间和邮箱服务器helo一下保证服务器不断开连接，一般不用配置
        self.connections = {}
        self.context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
        self.running = True

    def __create(self):
        while self.running:
            try:
                connection = smtplib.SMTP(timeout=30, host=self.host, port=self.port)
                if self.use_ssl:
                    connection.starttls(context=self.context)
                connection.login(self.send_email_user, self.send_email_password)
                key = uuid.uuid1().hex
                self.connections[key] = [connection, time.time() + len(self.connections.keys())]
                if not self.running:
                    self.replace(key, connection)
                break
            except Exception as e:
                try:
                    connection.quit()
                except:
                    pass
                if e.args[0] == 421:  # (421, b'Too many connections')
                    sleep_time = choice(range(5, 11))
                elif e.args[0] == 554:  # (554, b'IP<*****> in blacklist')
                    sleep_time = 1800
                else:
                    sleep_time = choice(range(5, 11))
                time.sleep(sleep_time)

    def __add(self):
        thread = Thread(target=self.__create)
        thread.setDaemon(True)
        thread.start()

    def __keep(self):
        last_check_helo = time.time()
        while self.running:
            if time.time() - last_check_helo >= self.re_helo_time:
                re_helo = True
                last_check_helo = time.time()
            else:
                re_helo = False
            connections = dict(self.connections.items())
            for key, connection_info in connections.items():
                connection, connection_time = connection_info
                if time.time() - connection_time >= self.connection_lifetime:
                    self.replace(key, connection)
                elif re_helo:
                    try:
                        connection.helo()
                    except:
                        self.replace(key, connection)
            time.sleep(1)

    def start(self):
        threads = [Thread(target=self.__create) for index in range(self.max_connections)]
        for thread in threads:
            thread.setDaemon(True)
            thread.start()
        thread = Thread(target=self.__keep)
        thread.setDaemon(True)
        thread.start()

    def close(self):
        self.running = False
        connections = dict(self.connections.items())
        while connections:
            for key, connection_info in connections.items():
                connection, connection_time = connection_info
                self.replace(key, connection)
            connections = dict(self.connections.items())

    def replace(self, key, connection):
        try:
            connection.quit()
        except:
            pass
        try:
            self.connections.pop(key, None)
        except:
            pass
        if self.running:
            self.__add()

    def get(self):
        time_now = time.time()
        while self.running:
            try:
                connections = dict(self.connections.items())
                key = choice(list(connections.keys()))
                connection, connection_time = connections[key]
            except (IndexError, KeyError):
                if time.time() - time_now > 3:
                    return
                else:
                    time.sleep(0.1)
                    continue
            if time.time() - connection_time >= self.connection_lifetime:
                if time.time() - time_now > 3:
                    return
                else:
                    self.replace(key, connection)
                    continue
            try:
                connection.helo()
                return key, connection
            except:
                self.replace(key, connection)


class EmailServer:
    def __init__(self, send_step=1, emails_list_key='', redis=None, connection_pool_kwargs={}):
        self.emails_list_key = emails_list_key  # 邮件队列的key
        self.send_step = send_step  # 发送并发量大小，163邮箱每批次只能发送11封邮件，outlook邮箱为20封
        self.connection_pool = ConnectionPool(**connection_pool_kwargs)
        self.io_loop = get_event_loop()
        self.logging = logging.getLogger()
        self.redis = redis
        self.running = True

    def format_email(self, host_user, email, charset='utf-8', use_localtime=True):
        # use_localtime 是否使用本地时间，True使用本地时间（东8区），False使用标准世界时间
        from_email = sanitize_address(host_user, charset)
        recipients = [sanitize_address(receive, charset) for receive in email.receives]
        if not from_email or not recipients:
            return ('', [], '')
        subtype = 'html' if email.message.strip().endswith('</html>') else 'plain'
        msg = SafeMIMEText(email.message, subtype, email.charset)
        msg['Subject'] = email.title
        email_sender_host = email.sender_name or host_user
        msg['From'] = formataddr([email.sender_name, email_sender_host])
        msg['To'] = ', '.join(map(str, email.receives))
        msg['Date'] = formatdate(localtime=use_localtime)
        msg['Message-ID'] = make_msgid()
        return from_email, recipients, msg.as_bytes(linesep='\r\n')

    async def send_one_email(self, email):
        # 发送一封邮件
        error_str, no_connection_error, retry_num = '', False, self.connection_pool.max_connections * 2
        for index in range(retry_num):
            try:
                try:
                    key, connection = self.connection_pool.get()
                    if connection:
                        from_email, recipients, message = self.format_email(connection.user, Email(*email))
                        if not recipients:
                            return '邮件发送失败，请检查收件人信息是否正确'
                        if not from_email:
                            return '邮件发送失败，请检查发件人信息是否正确'
                        senderrs = connection.sendmail(from_email, recipients, message)
                        if senderrs:
                            raise UnknownError(senderrs)
                        return True
                    elif index == retry_num - 1:
                        no_connection_error = True
                except Exception as e:
                    error_str = traceback.format_exc()
                    raise e
            # 异常处理根据需要自定义
            except smtplib.SMTPRecipientsRefused as e:
                self.logging.error('%s\nEmail info: %s' % (error_str, email))
                return '邮件发送失败，请检查收件人邮箱是否正确'
            except (smtplib.SMTPSenderRefused, smtplib.SMTPDataError, AttributeError, ValueError) as e:
                self.logging.error('%s\nEmail info: %s' % (error_str, email))
                self.connection_pool.replace(key, connection)
            except (ssl.SSLError, smtplib.SMTPServerDisconnected) as e:
                self.logging.error('%s\nEmail info: %s' % (error_str, email))
                self.connection_pool.replace(key, connection)
            except Exception:
                self.logging.error('%s\nEmail info: %s' % (error_str, email))
                return '邮件发送失败，请稍后再试'
        if no_connection_error:
            if error_str:
                error_str = '邮件连接全部失效，请检查是否被邮箱服务器加入黑名单，最后的异常：\n' + error_str
            else:
                error_str = '邮件连接全部失效，请检查是否被邮箱服务器加入黑名单'
        self.logging.error('%s\nEmail info: %s' % (error_str, email))
        return '邮件发送失败，请稍后再试'

    async def send_some_emails(self, emails):
        ''' 发送一个批次的邮件 '''
        tasks = [ensure_future(self.send_one_email(email), loop=self.io_loop) for email in emails]
        results = await gather(*tasks, loop=self.io_loop, return_exceptions=True)
        return results

    async def send_all_emails(self, emails):
        ''' 按照步长分批次发送所有邮件 '''
        # 把邮件按照步长分成多个批次
        tasks = [ensure_future(self.send_some_emails(emails[index: index + self.send_step]), loop=self.io_loop) for index in range(0, len(emails), self.send_step)]
        the_results = await gather(*tasks, loop=self.io_loop, return_exceptions=True)
        results = []
        for result in the_results:
            results.extend(result)
        return results

    def run_send_email_server(self):
        while self.running:
            email_info = self.redis.lpop(self.emails_list_key)
            if email_info:
                email_info = json.loads(email_info)
                result = run_coroutine_threadsafe(self.send_all_emails(email_info['emails']), self.io_loop).result()
                self.redis.set(email_info['send_task_id'], json.dumps(result), 60)
            else:
                time.sleep(0.1)

    def start(self):
        # 初始化与邮箱服务器的连接和连接保活服务
        self.connection_pool.start()
        print('Connection pool started.')

        # 启动一个协程事件循环
        thread = Thread(target=self.io_loop.run_forever)
        thread.setDaemon(True)
        thread.start()

        # 启动发送短信的服务
        thread = Thread(target=self.run_send_email_server)
        thread.setDaemon(True)
        thread.start()

    def stop(self):
        self.running = False
        self.connection_pool.close()
        print('Connection pool closed.')


class EmailSender:
    def __init__(self, emails_list_key='', redis=None):
        self.emails_list_key = emails_list_key
        self.redis = redis

    def send_emails(self, emails):
        # 发送邮件并等待结果
        ok_redis_key = 'emails_ok:%s' % uuid.uuid1().hex
        self.redis.rpush(self.emails_list_key, json.dumps({'send_task_id': ok_redis_key, 'emails': emails}))
        while True:
            result = self.redis.get(ok_redis_key)
            if result:
                self.redis.delete(ok_redis_key)
                return json.loads(result)
            else:
                time.sleep(0.1)


def get_html_content(email_title='', email_charset='utf-8'):
    # 格式化邮件内容，以发送html格式的邮件为例
    content = '''
        <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
        <html xmlns="http://www.w3.org/1999/xhtml">
        　<head>
        　　<meta http-equiv="Content-Type" content="text/html; charset={{ render_data.charset }}" />
        　　<title>{{ render_data.title }}</title>
        　　<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
        <style>
         table th, table td{
              line-height: 1.4em;
              font-size: 14px;
          }
        </style>
        　</head>
          <body style="margin: 0; padding: 0;">
            {{ render_data.data }}
          </body>
        </html>
    '''
    template = Template(content)
    render_data = {'title': email_title, 'data': {'示例': '内容'}, 'charset': email_charset}
    return template.render(Context({'render_data': render_data}))


def get_emails(email_num=2, receivers=[]):
    email_charset = 'utf-8'
    emails = []
    for email_index in range(email_num):
        email_title = '这是一封测试邮件[%s]-[%.2f]' % (email_index, time.time())
        # 格式化邮件内容，以发送html格式的邮件为例
        email_content = get_html_content(email_title, email_charset)
        sender_host = ''  # 填写则为代发模式
        emails.append([email_title, email_content, '旷古的寂寞', receivers, email_charset, sender_host])
        # emails.append([email_title, '测试邮件内容', '旷古的寂寞', receivers, email_charset])
    return emails


emails_list_key = 'emails_list'
redis_session = redis.Redis()
receivers = ['*******@163.com', '********@qq.com']
email_server = EmailServer(send_step=20, emails_list_key=emails_list_key, redis=redis_session, connection_pool_kwargs={
    'host': '******',  # 邮箱服务器的地址
    'port': 587,  # 邮箱服务器的端口号
    'send_email_user': '*****',  # 发件用户的SMPT服务账号(收件人看到的发件地址)
    'send_email_password': '****',  # 发件用户的SMPT服务账号的密码，注意是发件邮箱配置的SMPT服务的密码，不是发件邮箱登陆密码
    'max_connections': 20,  # 一个IP地址能够同时建立的连接数(连接池的大小)，163邮箱为10个，outlook邮箱为20个
    'use_ssl': True,  # smtp服务是否开启了ssl验证
    # 'connection_lifetime': 60,  # 连接的存活时间，到达这个时间后就替换掉该连接，一般不用配置
    # 're_helo_time': 10  # 连接的心跳时间间隔，每隔一定时间和邮箱服务器helo一下保证服务器不断开连接，一般不用配置
})
email_sender = EmailSender(emails_list_key, redis_session)

if __name__ == '__main__':
    email_server.start()
    result = email_sender.send_emails(get_emails(email_num=2, receivers=receivers))
    print(result)
    email_server.stop()

标签：outlook,协程,self,emails,connection,key,time,email,Python3
来源： https://blog.csdn.net/kuanggudejimo/article/details/94588184

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。