:以qq邮箱为例
使用的场景:
(1) 目前用在了重置用户密码
(2) 项目需要发送测试报告到用户邮箱
(3) 之前见过一个用户使用邮件的形式维护linux服务器(服务器端需要接收邮件进行解读)
注意:python3.7及以上会报错,我用的3.6
写代码前要准备的:
(1) 需要准备一个QQ号,因为要以这个QQ号发送和接受邮件。
(2) 需要这个QQ号邮箱开启SMTP/POP3服务
# 点开会验证,验证完会给个授权码,之后就是通过这个授权码代替用户密码
提醒:直接修改if __main__里的内容即可
发送邮件:
import os import random import smtplib import time from email.header import Header from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart class MyEmail(object): def __init__(self, sender, receiver, username, password, smtp_server='smtp.qq.com'): self.sender = sender self.receiver = receiver self.username = username self.password = password # 邮箱开启POP3和SMTP服务的授权码 self.smtp_server = smtp_server # 连接服务器 self.connect_server() # 表示一封邮件,需要邮件主题 def create_email(self, mail_title): # 创建一个带附件的实例 message = MIMEMultipart() message['From'] = self.sender message['To'] = self.receiver message['Subject'] = Header(mail_title, 'utf-8') self.message = message # 附件内容,如文本文件,图片文件等 def email_appendix(self, file_path): att1 = MIMEText(open(file_path, 'rb').read(), 'base64', 'utf-8') # 指定头部信息 att1["Content-Type"] = 'application/octet-stream' # 内容为二进制流 att1["Content-Disposition"] = 'attachment; filename="%s"' % (os.path.basename(file_path)) self.message.attach(att1) def email_text(self, content, content_type='plain'): # 邮件正文内容 # plain正常文本内容,html可以发送html格式内容 self.message.attach(MIMEText(content, content_type, 'utf-8')) def connect_server(self): # 连接邮件服务器 smtpObj = smtplib.SMTP_SSL() # 注意:如果遇到发送失败的情况(提示远程主机拒接连接),这里要使用SMTP_SSL方法 smtpObj.connect(self.smtp_server) try: # 连接qq邮箱服务器 smtpObj.login(self.username, self.password) # 给qq邮箱发送用户名和授权码,进行验证,如果账号没有授权会返回smtplib.SMTPAuthenticationError except smtplib.SMTPAuthenticationError: print('请检查用户名和授权码是否添加正确!') return else: self.smtpObj = smtpObj def send_mail(self): self.smtpObj.sendmail(self.sender, self.receiver, self.message.as_string()) # 发送一封邮件 print("邮件发送成功!!!") def __del__(self): self.smtpObj.close() # self.smtpObj.quit() t = ['乌鸦坐飞机', '邪恶在山顶', '双龙探珠', '螳螂拳', '蛇足拳', '水莲飘', ' 无相招', '佛朗明哥招', '飞天拳', ' 猫脚落地', ' 熊掌出击', ' 猫甩水', ' 猫转身'] if __name__ == '__main__': obj = MyEmail(sender='123456@qq.com', receiver='456789@qq.com', username='123456', password='pop3生成得密码') for i in range(10): # 创建邮件,及设置标题 obj.create_email('来自zezhou的轰炸消息') # 添加邮件内容 obj.email_text(random.choice(t)) # 添加附件,如图片或者文件啥的,需要文件的路径 # obj.email_appendix('suolong.jpg') # 发送邮件 obj.send_mail() time.sleep(0.5) else: obj.create_email('boom~') obj.email_text('boom boom boom') obj.send_mail()
效果:
发送邮件需要注意的:
# 如果想要发送html格式的内容
# 发送的html格式的内容,只会留body标签中的内容(且不包含script标签)
接受邮件:
import poplib import base64 import time from email.parser import Parser # 用来解析邮件主题 from email.header import decode_header # 用来解析邮件来源 from email.utils import parseaddr class AcceptEmail(object): def __init__(self, user_email, password, pop3_server='pop.qq.com'): self.user_email = user_email self.password = password self.pop3_server = pop3_server self.connect_email_server() def connect_email_server(self): self.server = poplib.POP3(self.pop3_server) # 打印POP3服务器的欢迎文字,验证是否正确连接到了邮件服务器 # print('连接成功 -- ', self.server.getwelcome().decode('utf8')) # +OK QQMail POP3 Server v1.0 Service Ready(QQMail v2.0) # 开始进行身份验证 self.server.user(self.user_email) self.server.pass_(self.password) def __del__(self): # 关闭与服务器的连接,释放资源 self.server.close() def get_email_count(self): # 返回邮件总数目和占用服务器的空间大小(字节数), 通过stat()方法即可 email_num, email_size = self.server.stat() # print("消息的数量: {0}, 消息的总大小: {1}".format(email_num, email_size)) return email_num def receive_email_info(self, now_count=None): # 返回邮件总数目和占用服务器的空间大小(字节数), 通过stat()方法即可 email_num, email_size = self.server.stat() # print("消息的数量: {0}, 消息的总大小: {1}".format(email_num, email_size)) self.email_count = email_num self.email_sumsize = email_size # 使用list()返回所有邮件的编号,默认为字节类型的串 rsp, msg_list, rsp_siz = self.server.list() # print(msg_list, '邮件数量',len(msg_list)) # print("服务器的响应: {0}, 消息列表: {1}, 返回消息的大小: {2}".format(rsp, msg_list, rsp_siz)) # print('邮件总数: {}'.format(len(msg_list))) self.response_status = rsp self.response_size = rsp_siz # 下面获取最新的一封邮件,某个邮件下标(1开始算) # total_mail_numbers = len(msg_list) # 动态取消息 total_mail_numbers = now_count rsp, msglines, msgsiz = self.server.retr(total_mail_numbers) # rsp, msglines, msgsiz = self.server.retr(1) # print("服务器的响应: {0}, 原始邮件内容: {1}, 该封邮件所占字节大小: {2}".format(rsp, msglines, msgsiz)) # 从邮件原内容中解析 msg_content = b' '.join(msglines).decode('gbk') msg = Parser().parsestr(text=msg_content) self.msg = msg # print('解码后的邮件信息: {}'.format(msg)) def recv(self, now_count=None): self.receive_email_info(now_count) self.parser() def get_email_title(self): subject = self.msg['Subject'] value, charset = decode_header(subject)[0] if charset: value = value.decode(charset) # print('邮件主题: {0}'.format(value)) self.email_title = value def get_sender_info(self): hdr, addr = parseaddr(self.msg['From']) # name 发送人邮箱名称, addr 发送人邮箱地址 name, charset = decode_header(hdr)[0] if charset: name = name.decode(charset) self.sender_qq_name = name self.sender_qq_email = addr # print('发送人邮箱名称: {0},发送人邮箱地址: {1}'.format(name, addr)) def get_email_content(self): content = self.msg.get_payload() # 文本信息 content_charset = content[0].get_content_charset() # 获取编码格式 text = content[0].as_string().split('base64')[-1] text_content = base64.b64decode(text).decode(content_charset) # base64解码 self.email_content = text_content # print('邮件内容: {0}'.format(text_content)) # 添加了HTML代码的信息 content_charset = content[1].get_content_charset() text = content[1].as_string().split('base64')[-1] # html_content = base64.b64decode(text).decode(content_charset) # print('文本信息: {0} 添加了HTML代码的信息: {1}'.format(text_content, html_content)) def parser(self): self.get_email_title() self.get_sender_info() self.get_email_content() def get_new_mail(dic, second=5): t = AcceptEmail(**dic) now_count = t.get_email_count() print('开启的时候的邮件数量为:%s' % now_count) # 每次需要重新连接邮箱服务器,才能获取到最新的消息 # 默认每隔5秒看一次是否有新内容 while True: obj = AcceptEmail(**dic) count = obj.get_email_count() if count > now_count: new_mail_count = count - now_count print('有新的邮件数量:%s' % new_mail_count) for i in range(1, new_mail_count + 1): obj = AcceptEmail(**dic) now_count += 1 obj.recv(now_count) yield {"title": obj.email_title, "sender": obj.sender_qq_name, "sender_email": obj.sender_qq_email, "email_content": obj.email_content} # print('-' * 30) # print("邮件主题:%s 发件人:%s 发件人邮箱:%s 邮件内容:%s" % ( # obj.email_title, obj.sender_qq_name, obj.sender_qq_email, obj.email_content)) # print('-' * 30) # else: # print('没有任何新消息.') time.sleep(second) if __name__ == '__main__': dic = { 'user_email': 'xxx@qq.com', 'password': 'xxx', } print('正在监听邮件服务器端是否有新消息---') try: iterator = get_new_mail(dic) except TypeError: print('监听的内容有误,有图片数据等,无法解析而报错,不是纯文本内容') else: for dic in iterator: # 如果需要过滤某个用户的邮件内容,加个if判断是否是该邮箱即可 # if dic.get("sender_email") == 'xxx': print('-' * 30) print("邮件主题:%s 发件人:%s 发件人邮箱:%s 邮件内容:%s" % ( dic["title"], dic["sender"], dic["sender_email"], dic["email_content"])) print('-' * 30)
过程中遇到的问题:
1.认证问题
解决方法:
在自己的QQ邮箱开启POP3和SMTP服务,得到授权码,填写正确的信息,重新进行认证。