标签：异步 Python self smtp def msg import type 邮件
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
运行
标签：异步,Python,self,smtp,def,msg,import,type,邮件 来源： https://www.cnblogs.com/rainbow-tan/p/16022570.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。