Python 非同步傳送郵件指令碼
阿新 • • 發佈：2022-03-18
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='[email protected]', pwd='jlkjlKKK#s,', received_users=['[email protected]', '[email protected]'], 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