python邮件发送小工具

  1 # -*- coding:utf-8 -*-
  2 """
  3 邮件发送小工具
  4 """
  5 __author__ = "shu2015626"
  6 
  7 import os
  8 import smtplib
  9 import logging
 10 import datetime
 11 import traceback
 12 from typing import Dict, List, Sequence
 13 from collections.abc import Iterable
 14 from email.header import Header
 15 from email.mime.text import MIMEText
 16 from email.utils import formataddr, parseaddr
 17 from email.mime.multipart import MIMEMultipart
 18 from email.mime.application import MIMEApplication
 19 
 20 import jinja2
 21 
 22 logger = logging.getLogger(‘proj‘)
 23 
 24 
 25 class EmailOper(object):
 26     def __init__(self, email_host: str, email_port: int, email_sender: str, sender_pass: str, sender_alias: str = None):
 27         """
 28         配置邮件发送服务
 29         :param email_host: 发送邮件的主机
 30         :param email_port: 邮件发送端口
 31         :param email_sender: 发件人
 32         :param sender_pass: 发件人密码
 33         :param sender_alias: 发件人别称
 34         """
 35         self.__email_host = email_host
 36         self.__email_port = email_port
 37         self.__email_sender = email_sender
 38         self.__sender_pass = sender_pass
 39         self.__sender_alias = sender_alias or self.__email_sender.split("@", 1)[0]
 40 
 41     @property
 42     def email_host(self):
 43         return self.__email_host
 44 
 45     @email_host.setter
 46     def email_host(self, value: str):
 47         self.__email_host = value
 48 
 49     @property
 50     def email_port(self):
 51         return self.__email_port
 52 
 53     @email_port.setter
 54     def email_port(self, value: int):
 55         self.__email_port = value
 56 
 57     @property
 58     def email_sender(self):
 59         return self.__email_sender
 60 
 61     @email_sender.setter
 62     def email_sender(self, value: str):
 63         self.__email_sender = value
 64 
 65     @property
 66     def sender_pass(self):
 67         return self.__sender_pass
 68 
 69     @sender_pass.setter
 70     def sender_pass(self, value: str):
 71         self.__sender_pass = value
 72 
 73     @property
 74     def sender_alias(self):
 75         return self.__sender_alias
 76 
 77     @sender_alias.setter
 78     def sender_alias(self, value: str):
 79         self.__sender_alias = value
 80 
 81     def build_html_email(self, package_name: str, template_name: str, **context) -> str:
 82         """
 83         用法同 flask.render_template:
 84 
 85         build_html_email(‘package‘, ‘template.html‘, var1=‘foo‘, var2=‘bar‘)
 86         :param package_name:
 87         :param template_name:
 88         :param context:
 89         :return:
 90         """
 91         env = jinja2.Environment(
 92             loader=jinja2.PackageLoader(package_name, ‘templates‘)
 93         )
 94         template = env.get_template(template_name)
 95 
 96         return template.render(**context)
 97 
 98     def __format_receivers(self, receivers: Sequence):
 99         """
100         格式化收件人的地址。
101         若传入的是字典。应该类似下面的格式：
102         {
103             “test@123.com”: "测试",
104             “admin@123.com”: "管理员"
105         }
106         若传入的是list/tuple。应该类似下面的格式：
107         [“test@123.com”, “admin@123.com”]
108         此函数会将输出转化为：
109         {
110             “test@123.com”: "test",
111             “admin@123.com”: "admin"
112         }
113         :param receivers: 收件人邮箱
114         :return:
115         """
116         assert isinstance(receivers, (list, tuple, dict)), ‘收件人地址必须以list/tuple或者dict形式传入，以支持多个收件人‘
117         if isinstance(receivers, dict):
118             return receivers
119         elif isinstance(receivers, (list, tuple)):
120             receivers = {receiver: receiver.split("@", 1)[0] for receiver in receivers}
121             return receivers
122 
123     def __format_addr(self, address: str, encoding: str = ‘utf-8‘):
124         """
125         格式化每个收件地址，以在邮箱中显示姓名
126         :param address:
127         :param encoding:
128         :return:
129         """
130         name, addr = parseaddr(addr=address)
131         return formataddr((Header(name, encoding).encode(), addr))
132 
133     def __build_email(self, receivers: Sequence, email_body: str, subject: str, email_type: str = ‘plain‘,
134                       encoding: str = ‘utf-8‘, attachments: Sequence = None):
135         if not attachments:
136             # MIMEText三个参数：第一个为文本内容，第二个 plain 设置文本格式，第三个 utf-8 设置编码
137             message = MIMEText(email_body, email_type, encoding)
138             message[‘From‘] = formataddr([self.__sender_alias, self.__email_sender])
139             receivers = self.__format_receivers(receivers)  # 全部变成字典格式，邮箱：名称
140             to_string = ‘;‘.join([self.__format_addr(‘{0}<{1}>‘.format(addr_alias, addr), encoding)
141                                   for addr, addr_alias in receivers.items()])
142             message[‘To‘] = Header(to_string)
143             message[‘Subject‘] = Header(subject, encoding)
144         else:
145             if isinstance(attachments, str):
146                 attachments = [attachments]
147             elif isinstance(attachments, Iterable):
148                 attachments = attachments
149             else:
150                 raise Exception("传入的附件，应该是一个文件的全路径，或者列表，包含多个附件的具体路径")
151             message = MIMEMultipart()
152             # 邮件正文内容
153             message.attach(MIMEText(email_body, email_type, encoding))
154             # 构造附件
155             for attach in attachments:
156                 with open(attach, ‘rb‘) as f:
157                     att = MIMEApplication(f.read())  # 如果只是文本文件可以用MIMEText代替，但像office之类的，需要MIMEApplication
158                     att.add_header(‘Content-Disposition‘, ‘attachment‘, filename=os.path.basename(attach))
159                     message.attach(att)
160 
161             message[‘From‘] = formataddr([self.__sender_alias, self.__email_sender])
162             receivers = self.__format_receivers(receivers)  # 全部变成字典格式，邮箱：名称
163             to_string = ‘;‘.join([self.__format_addr(‘{0}<{1}>‘.format(addr_alias, addr), encoding)
164                                   for addr, addr_alias in receivers.items()])
165             message[‘To‘] = Header(to_string)
166             message[‘Subject‘] = Header(subject, encoding)
167 
168         return message
169 
170     def send_email(self, receivers: Sequence, email_body: str, subject: str, email_type: str = ‘plain‘,
171                    encoding: str = ‘utf-8‘, attachments: Sequence = None):
172         """
173         发送邮件
174         :param receivers: 收件人列表
175         :param email_body: 邮件正文
176         :param subject: 邮件主题
177         :param email_type: 邮件类型：plain 或者 html
178         :param encoding: 编码
179         :param attachments: 附件全路径列表
180         :return:
181         """
182         obj_smtp = None
183         try:
184             if self.__email_port == 25:
185                 obj_smtp = smtplib.SMTP(timeout=10)
186             elif self.__email_port == 465:
187                 obj_smtp = smtplib.SMTP_SSL(timeout=10)
188             else:
189                 raise ValueError(f"不是标准的邮箱端口：{self.__email_port}, 请确认端口，应该为25或者465")
190 
191             obj_smtp.connect(self.__email_host, self.__email_port)
192             if self.__sender_pass:
193                 obj_smtp.login(self.__email_sender, self.__sender_pass)
194 
195             # 格式化收件人地址为{addr1:name1, ……}的形式
196             receivers = self.__format_receivers(receivers)
197             # 构造邮件
198             message = self.__build_email(receivers, email_body, subject, email_type, encoding, attachments)
199             # 发送邮件
200             obj_smtp.sendmail(self.__email_sender, receivers, message.as_string())
201         except ValueError as e:
202             info = ‘无法发送邮件。错误信息:{}‘.format(traceback.format_exc())
203             logger.error(info)
204             raise ValueError from e
205         except TimeoutError as e:
206             info = ‘无法发送邮件。错误信息:{}‘.format(traceback.format_exc())
207             logger.error(info)
208             raise TimeoutError from e
209         finally:
210             try:
211                 if obj_smtp:
212                     obj_smtp.quit()
213             except Exception as e:
214                 obj_smtp = None
215 
216 
217 if __name__ == "__main__":
218     EMAIL_HOST = "smtp.qq.com"
219     EMAIL_PORT = 465
220     EMAIL_SENDER = "12345678@qq.com"
221     SENDER_PASS = "dsfd"
222     SENDER_ALIAS = "admin"
223     RECEIVERS = ["12345678@qq.com", ]
224 
225     obj_email_oper = EmailOper(EMAIL_HOST, EMAIL_PORT, EMAIL_SENDER, SENDER_PASS, SENDER_ALIAS)
226     email_body = "测试邮件，勿回！"
227     SUBJECT = f"[%s - %s]测试邮件" % (‘项目名‘, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
228 
229     obj_email_oper.send_email(
230         RECEIVERS, email_body, SUBJECT, ‘plain‘
231     )