poyee-data-warehouse/dw_base/utils/email_utils.py

#!/usr/bin/env /usr/bin/python3
# -*- coding:utf-8 -*-

import os
import smtplib
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Union, List

from dw_base.utils.file_utils import get_abs_path


class EmailHandler(object):
    def __init__(self, smtp_server: str, smtp_port: int, username: str, password: str):
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.username = username
        self.password = password

    def send_email(self,
                   to: Union[List[str], str],
                   cc: Union[List[str], str],
                   subject: str,
                   content: str,
                   attachments: Union[List[str], str]):
        smtp = smtplib.SMTP(self.smtp_server, self.smtp_port)
        smtp.login(self.username, self.password)
        if isinstance(to, list):
            to = ','.join(to)
        if cc and isinstance(cc, list):
            cc = ','.join(cc)
        if not attachments:
            attachments = []
        elif isinstance(attachments, str):
            attachments = attachments.split(',')
        message = MIMEMultipart()
        message['From'] = self.username
        message['To'] = to
        if cc:
            message['Cc'] = cc
        message['Subject'] = Header(subject, 'utf-8').encode()
        if content:
            # 添加文本
            message.attach(MIMEText(content, 'plain', 'utf-8'))
        # 添加附件
        for file_path in attachments:
            self.add_attachment(message, file_path)
        smtp.sendmail(self.username, to, message.as_string(unixfrom=True))
        smtp.quit()

    # @staticmethod
    # def add_attachment(message: MIMEMultipart, file_path: str):
    #     local_file = get_abs_path(file_path)
    #     file_name = os.path.basename(file_path)
    #     with open(local_file, 'r') as f:
    #         # 第二个
    #         mime = MIMEBase('text', 'txt', filename=file_name)
    #         # 加上必要的头信息:
    #         mime.add_header('Content-Disposition', 'attachment', filename=file_name)
    #         # mime.add_header('Content-ID', '<0>')
    #         # mime.add_header('X-Attachment-Id', '0')
    #         # 把附件的内容读进来:
    #         mime.set_payload(f.read())
    #         # 用Base64编码:
    #         encoders.encode_base64(mime)
    #         # 添加到MIMEMultipart:
    #         message.attach(mime)

    @staticmethod
    def add_attachment(message: MIMEMultipart, file_path: str):
        local_file = get_abs_path(file_path)
        file_name = os.path.basename(file_path)
        with open(local_file, 'r', encoding='utf-8') as f:
            mime_text = MIMEText(f.read(), 'base64', 'utf-8')
            mime_text["Content-Type"] = 'application/octet-stream'
            mime_text["Content-Disposition"] = 'attachment; filename=%s' % file_name
        message.attach(mime_text)


if __name__ == '__main__':
    pass