实现自动化操作Excel功能@自动发送邮件功能(图)

两天前，一位网友向我发送了一条私人消息，说他希望我帮助他实现批量发送电子邮件并自动定时发送点对点的功能。只有聊天后才知道为什么网民迫切需要上述功能。他的需求是这样的：

1.从excel文件中读取5K电子邮件地址，并将相同的电子邮件发送到这些地址，但是每次仅发送20电子邮件；

2.这些电子邮件一次只能点对点发送，这意味着接收电子邮件的每个人只会看到发件人的电子邮件地址。

@实现自动操作Excel功能

class handleExcel:
    def __init__(self, fileName=None):  # 打开文件或者新建文件（如果不存在的话)
        ...
    def save(self, new_filename=None):  # save excel file
        '''保存文件'''
        ...
    def close(self):  # close excel file
        '''关闭文件'''
        ...
    def add_sheet(self, sheetname=None):
        '''添加工作表'''
        ...
    def copy_sheet(self, srcsheet, destsheet=None, before=None): 
        '''复制工作表'''
        ...
    def delete_sheet(self, sheet):
        '''删除工作表'''
        ...
    def get_rows(self, sheet):
        '''获取有效使用行数'''
        ...
    def get_cols(self, sheet):
        '''获取有效使用列数'''
        ...
    def read_cell(self, sheet, row, col):  
        '''读单元格数据'''
        ...
    def write_cell(self, sheet, row, col, value):  
        '''写单元格数据'''
        ...
    def get_range(self, sheet, row1, col1, row2, col2):  
        '''获取某一区域的数据'''
        ...
    def copy_range(self, sheet, row1, col1, row2, col2, tgt_row, tgt_col):
        '''复制一块区域'''
        ...
    def cut_range(self, sheet, row1, col1, row2, col2, tgt_row, tgt_col):
        '''移动一块区域'''
        ...
    def clear_range(self, sheet, row1, col1, row2, col2):
        '''清除指定区域内容'''
        ...
    def add_picture(self, sheet, pic_name, left, top, width, height): 
        '''添加图片'''
        ...
    def del_row(self, sheet, row):
        '''删除指定行'''
        ...
    def del_col(self, sheet, col):
        '''删除指定列'''
        ...

if __name__ == '__main__':
    xls = handleExcel(r'C:\xxoo\mail.xlsx')
    rows = xls.get_rows('Sheet1')
    cols = xls.get_cols('Sheet1')
    print(rows)
    print(cols)
    mail_list = []
    for i in range(1, rows):
        cell_value = xls.read_cell('Sheet1', i, 1)
        mail_list.append(cell_value)
        print(cell_value)
    print(mail_list)
    xls.close()

@自动邮件发送功能

#!/usr/bin/env python
# -*-coding:utf-8-*-
import threading
from auto_send_mail.handle_excel import handleExcel
__author__ = 'SamWoo'
import mimetypes
import os
import smtplib
import time
import schedule
from email import encoders
from email.header import Header
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import parseaddr, formataddr
from os.path import getsize
class EmailManager:
    def __init__(self, **kwargs):
        '''
        constructor
        :param kwargs:Variable paramete
        '''
        self.kwargs = kwargs
        self.smtp_server = 'smtp.qq.com'
        self.MAX_FILE_SIZE = 10 * 1024 * 1024
    def __get_cfg(self, key, throw=True):
        '''
        get the configuration file based on the key
        :param key:
        :param throw:
        :return:
        '''
        cfg = self.kwargs.get(key)
        if throw == True and (cfg is None or cfg == ''):
            raise Exception("The configuration can't be empty", 'utf-8')
        return cfg
    def __init_cfg(self):
        self.msg_from = self.__get_cfg('msg_from')
        self.password = self.__get_cfg('password')
        self.msg_to = ';'.join(self.__get_cfg('msg_to'))
        self.msg_subject = self.__get_cfg('msg_subject')
        self.msg_content = self.__get_cfg('msg_content')
        self.msg_date = self.__get_cfg('msg_date')
        # attachment
        self.attach_file = self.__get_cfg('attach_file', throw=False)
    def login_server(self):
        '''
        login server
        :return:
        '''
        server = smtplib.SMTP_SSL(self.smtp_server, 465)
        server.set_debuglevel(1)
        server.login(self.msg_from, self.password)
        return server
    def get_main_msg(self):
        '''
        suject content
        :return:
        '''
        msg = MIMEMultipart()
        # message content
        msg.attach(MIMEText(self.msg_content, 'plain', 'utf-8'))
        msg['From'] = self._format_addr('Sam <%s>' % self.msg_from)
        msg['To'] = self._format_addr('To <%s>' % self.msg_to)
        msg['Subject'] = Header(self.msg_subject, 'utf-8')
        msg['Date'] = self.msg_date
        # attachment content
        attach_file = self.get_attach_file()
        if attach_file is not None:
            msg.attach(attach_file)
        return msg
    def get_attach_file(self):
        '''
        generate mail attachment content
        :return:
        '''
        if self.attach_file is not None and self.attach_file != '':
            try:
                if getsize(self.attach_file) > self.MAX_FILE_SIZE:
                    raise Exception('The attachment is too large and the upload failed!!')
                with open(self.attach_file, 'rb') as file:
                    ctype, encoding = mimetypes.guess_type(self.attach_file)
                    if ctype is None or encoding is not None:
                        ctype = 'application/octet-stream'
                    maintype, subtype = ctype.split('/', 1)
                    mime = MIMEBase(maintype, subtype)
                    mime.set_payload(file.read())
                    # set header
                    mime.add_header('Content-Disposition', 'attachment',
                                    filename=os.path.basename(self.attach_file))
                    mime.add_header('Content-ID', '<0>')
                    mime.add_header('X-Attachment-Id', '0')
                    # set the attachment encoding rules
                    encoders.encode_base64(mime)
                    return mime
            except Exception as e:
                print('%s......' % e)
                return None
        else:
            return None
    def _format_addr(self, s):
        name, addr = parseaddr(s)
        return formataddr((Header(name, 'utf-8').encode(), addr))
    def send(self):
        try:
            # initialize the configuration file
            self.__init_cfg()
            # log on to the SMTP server and verify authorization
            server = self.login_server()
            # mail content
            msg = self.get_main_msg()
            # send mail
            server.sendmail(self.msg_from, self.msg_to, msg.as_string())
            server.quit()
            print("Send succeed!!")
        except smtplib.SMTPException:
            print("Error:Can't send this email!!")
def get_mail_address():
    global mail_list, i
    xls = handleExcel(r'C:\xxoo\mail.xlsx')
    rows = xls.get_rows('Sheet1')
    mail_list = []
    for i in range(1, rows+1):
        cell_value = xls.read_cell('Sheet1', i, 1)
        mail_list.append(cell_value)
        print(cell_value)
    print(mail_list)
    xls.close()
    return mail_list
def send_email(mail_list):
    global manager
    mail_cfg = {'msg_from': 'xxxxx@qq.com',
                'password': 'xxxxx',
                'msg_to': ['xxxx@qq.com'],
                'msg_subject': 'Python Auto Send Email Test',
                'msg_content': 'Hi, boy! Just do it, Python!',
                'attach_file': r'.\font.zip',
                'msg_date': time.ctime()
                }
    for mail in mail_list:
        mail_cfg['msg_to'] = [mail]
        manager = EmailManager(**mail_cfg)
        print('Now send email to {}'.format(mail_cfg['msg_to']))
        manager.send()
def run_thread(func):
    mail_list = get_mail_address()
    num = len(mail_list) // 4
    list1 = mail_list[0:num]
    list2 = mail_list[num:num * 2]
    list3 = mail_list[num * 2:num * 3]
    list4 = mail_list[num * 3:]
    #开启4条线程分批发送500封相同邮件
    threading.Thread(target=func, args=(list1,)).start()
    threading.Thread(target=func, args=(list2,)).start()
    threading.Thread(target=func, args=(list3,)).start()
    threading.Thread(target=func, args=(list4,)).start()
if __name__ == "__main__":
    schedule.every(10).minutes.do(run_threaded, send_email)
    while True:
        schedule.run_pending()
        time.sleep(1)

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/shumachanpin/article-358820-1.html