這篇文章主要介紹了Python基於smtplib實現異步發送郵件服務,需要的朋友可以參考下
基於smtplib包制作而成，但在實踐中發現一個不知道算不算是smtplib留的一個坑，在網絡斷開的情況下發送郵件時會拋出一個socket.gaierror的異常，但是smtplib中並沒有捕獲這個異常，導致程序會因這個異常終止，因此代碼中針對這部分的異常進行處理，確保不會異常終止。
?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 #!/usr/bin/env python # -*- coding: utf-8 -*- __author__ = 'Zoa Chou' # see http://www.mudoom.com/Article/show/id/29.html for detail import logging import smtplib import mimetypes import socket from email import encoders from email.header import Header from email.mime.text import MIMEText, MIMENonMultipart from email.mime.base import MIMEBase from email.utils import parseaddr, formataddr class Mailer(object): def __init__(self): pass def send_mail(self, smtp_server, from_address, to_address, subject, body, files=None): """ 發送郵件主程序 :param smtp_server: dict 郵件服務器設置 :keyword host: string smtp服務器地址 :keyword port: int smtp服務器端口號 :keyword user: string 用戶名 :keyword passwd: string 密碼 :keyword ssl: bool 是否啟用ssl,默認False :keyword timeout: int 超時時間,默認10s :param from_address: 發件人郵箱 :param to_address: 收件人郵箱 :param subject: 郵件標題 :param body: 郵件內容 :param files: 附件 :raise: NetworkError/MailerException """ # 格式化郵件內容 body = self._encode_utf8(body) # 郵件類型 content_type = 'html' if body.startswith('<html>') else 'plain' msg = MIMENonMultipart() if files else MIMEText(body, content_type, 'utf-8') # 格式化郵件數據 msg['From'] = self._format_address(from_address) msg['To'] = ', '.join(self._format_list(to_address)) msg['subject'] = self._encode_utf8(subject) # 構造附件數據 if files: msg.attach(MIMEText(body, content_type, 'utf-8')) cid = 0 for file_name, payload in files: file_name = self._encode_utf8(file_name) main_type, sub_type = self._get_file_type(file_name) if hasattr(payload, 'read'): payload = payload.read() f_name = self._encode_header(file_name) mime = MIMEBase(main_type, sub_type, filename=f_name) mime.add_header('Content-Disposition', 'attachment', filename=f_name) mime.add_header('Content-ID', '<%s>' % cid) mime.add_header('X-Attachment-Id', '%s' % cid) mime.set_payload(payload) encoders.encode_base64(mime) msg.attach(mime) cid += 1 host = smtp_server.get('host') port = smtp_server.get('port') user = smtp_server.get('user') passwd = smtp_server.get('passwd') ssl = smtp_server.get('ssl', False) time_out = smtp_server.get('timeout', 10) # 沒有輸入端口則使用默認端口 if port is None or port == 0: if ssl: port = 465 else: port = 25 logging.debug('Send mail form %s to %s' % (msg['From'], msg['To'])) try: if ssl: # 開啟ssl連接模式 server = smtplib.SMTP_SSL('%s:%d' % (host, port), timeout=time_out) else: server = smtplib.SMTP('%s:%d' % (host, port), timeout=time_out) # 開啟調試模式 # server.set_debuglevel(1) # 如果存在用戶名密碼則嘗試登錄 if user and passwd: server.login(user, passwd) # 發送郵件 server.sendmail(from_address, to_address, msg.as_string()) logging.debug('Mail sent success.') # 關閉stmp連接 server.quit() except socket.gaierror, e: """ 網絡無法連接 """ logging.exception(e) raise NetworkError(e) except smtplib.SMTPServerDisconnected, e: """ 網絡連接異常 """ logging.exception(e) raise NetworkError(e) except smtplib.SMTPException, e: """ 郵件發送異常 """ logging.exception(e) raise MailerException(e) def _format_address(self, s): """ 格式化郵件地址 :param s:string 郵件地址 :return: string 格式化後的郵件地址 """ name, address = parseaddr(s) return formataddr((self._encode_header(name), self._encode_utf8(address))) def _encode_header(self, s): """ 格式化符合MIME的頭部數據 :param s: string 待格式化數據 :return: 格式化後的數據 """ return Header(s, 'utf-8').encode() def _encode_utf8(self, s): """ 格式化成utf-8編碼 :param s: string 待格式化數據 :return: string 格式化後的數據 """ if isinstance(s, unicode): return s.encode('utf-8') else: return s def _get_file_type(self, file_name): """ 獲取附件類型 :param file_name: 附件文件名 :return: dict 附件MIME """ s = file_name.lower() pos = s.rfind('.') if pos == -1: return 'application', 'octet-stream' ext = s[pos:] mime = mimetypes.types_map.get(ext, 'application/octet-stream') pos = mime.find('/') if pos == (-1): return mime, '' return mime[:pos], mime[pos+1:] def _format_list(self, address): """ 將收件人地址格式化成list :param address: string/list 收件人郵箱 :return: list 收件人郵箱list """ l = address if isinstance(l, basestring): l = [l] return [self._format_address(s) for s in l] class MailerException(Exception): """ 郵件發送異常類 """ pass class NetworkError(MailerException): """ 網絡異常類 """ pass # test for @qq.com if __name__ == '__main__': import sys def prompt(prompt): """ 接收終端輸入的數據 """ sys.stdout.write(prompt + ": ") return sys.stdin.readline().strip() from_address = prompt("From(Only @qq.com)") passwd = prompt("Password") to_address = prompt("To").split(',') subject = prompt("Subject") print "Enter message, end with ^D:" msg = '' while 1: line = sys.stdin.readline() if not line: break msg = msg + line print "Message length is %d" % len(msg) # QQ郵箱默認設置 smtp_server = {'host': 'smtp.qq.com', 'port': None, 'user': from_address, 'passwd': passwd, 'ssl': True} mailer = Mailer() try: mailer.send_mail(smtp_server, from_address, to_address, subject, msg) except MailerException, e: print(e)以上所述就是本文的全部內容了，希望大家能夠喜歡。
