快捷導(dǎo)航

Python基于smtplib實(shí)現(xiàn)異步發(fā)送郵件服務(wù)

更新時(shí)間：2015年05月28日 12:11:09 投稿：hebedich

這篇文章主要介紹了Python基于smtplib實(shí)現(xiàn)異步發(fā)送郵件服務(wù),需要的朋友可以參考下

基于smtplib包制作而成，但在實(shí)踐中發(fā)現(xiàn)一個(gè)不知道算不算是smtplib留的一個(gè)坑，在網(wǎng)絡(luò)斷開的情況下發(fā)送郵件時(shí)會(huì)拋出一個(gè)socket.gaierror的異常，但是smtplib中并沒(méi)有捕獲這個(gè)異常，導(dǎo)致程序會(huì)因這個(gè)異常終止，因此代碼中針對(duì)這部分的異常進(jìn)行處理，確保不會(huì)異常終止。

#!/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):
    """
    發(fā)送郵件主程序
    :param smtp_server: dict 郵件服務(wù)器設(shè)置
      :keyword host: string smtp服務(wù)器地址
      :keyword port: int smtp服務(wù)器端口號(hào)
      :keyword user: string 用戶名
      :keyword passwd: string 密碼
      :keyword ssl: bool 是否啟用ssl,默認(rèn)False
      :keyword timeout: int 超時(shí)時(shí)間,默認(rèn)10s
    :param from_address: 發(fā)件人郵箱
    :param to_address: 收件人郵箱
    :param subject: 郵件標(biāo)題
    :param body: 郵件內(nèi)容
    :param files: 附件
    :raise: NetworkError/MailerException
    """
    # 格式化郵件內(nèi)容
    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')
    # 格式化郵件數(shù)據(jù)
    msg['From'] = self._format_address(from_address)
    msg['To'] = ', '.join(self._format_list(to_address))
    msg['subject'] = self._encode_utf8(subject)

    # 構(gòu)造附件數(shù)據(jù)
    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)

    # 沒(méi)有輸入端口則使用默認(rèn)端口
    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)
      # 開啟調(diào)試模式
      # server.set_debuglevel(1)

      # 如果存在用戶名密碼則嘗試登錄
      if user and passwd:
        server.login(user, passwd)

      # 發(fā)送郵件
      server.sendmail(from_address, to_address, msg.as_string())

      logging.debug('Mail sent success.')

      # 關(guān)閉stmp連接
      server.quit()

    except socket.gaierror, e:
      """ 網(wǎng)絡(luò)無(wú)法連接 """
      logging.exception(e)
      raise NetworkError(e)

    except smtplib.SMTPServerDisconnected, e:
      """ 網(wǎng)絡(luò)連接異常 """
      logging.exception(e)
      raise NetworkError(e)

    except smtplib.SMTPException, e:
      """ 郵件發(fā)送異常 """
      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的頭部數(shù)據(jù)
    :param s: string 待格式化數(shù)據(jù)
    :return: 格式化后的數(shù)據(jù)
    """
    return Header(s, 'utf-8').encode()

  def _encode_utf8(self, s):
    """
    格式化成utf-8編碼
    :param s: string 待格式化數(shù)據(jù)
    :return: string 格式化后的數(shù)據(jù)
    """
    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):
  """ 郵件發(fā)送異常類 """
  pass


class NetworkError(MailerException):
  """ 網(wǎng)絡(luò)異常類 """
  pass

# test for @qq.com
if __name__ == '__main__':
  import sys

  def prompt(prompt):
    """
    接收終端輸入的數(shù)據(jù)
    """
    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郵箱默認(rèn)設(shè)置
  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)

以上所述就是本文的全部?jī)?nèi)容了，希望大家能夠喜歡。

您可能感興趣的文章: