Python代码段

• EXCEL导出操作

def lists_to_xls(rows, xls, header=None):
    columns = rows[0]
    range_columns = range(len(columns))
    data = rows[1:]
    startrow = 0
    if header:
        startrow = 1
    df = pandas.DataFrame(rows[1:], columns=columns)
    # Create a Pandas Excel writer using XlsxWriter as the engine.
    writer = pandas.ExcelWriter(xls, engine='xlsxwriter')
    book = writer.book
    # Convert the dataframe to an XlsxWriter Excel object.
    sheet_name = '明细表'
    df.to_excel(writer, sheet_name=sheet_name, index=False, startrow=startrow)
    # Get the xlsxwriter worksheet object.
    worksheet = writer.sheets[sheet_name]
    # 计算列宽并设置
    widths = [0 for i in range_columns]
    for row in rows:
        for i, v in enumerate(row):
            ch = len(v.encode('utf8'))
            if ch > widths[i]:
                widths[i] = ch
    # Set the column width and format
    for i, v in enumerate(widths):
        worksheet.set_column(i, i, widths[i])
    # 标题，单元格合并及格式设置
    if header:
        fmt = book.add_format({
            'bold': True
        })
        fmt.set_font_size(24)
        fmt.set_font_color('#FF0000')
        worksheet.merge_range(0, 0, 0, len(columns) - 1, header, cell_format=fmt)
    writer.save()
    print('导出文件到', xls)

• IMAP接收特定邮件

# 测试代码
mail = IMap()
conn = mail.login()
accs = mail.get_content(conn)
mail.loginout(conn)
    
    
class IMap:
    def __init__(self):
        self.user_id = '11@xx.com'
        self.password = '****'
        self.imap_server = 'imap.xxx.com'
        self.imap_server_port = 993

    def login(self):
        """
        登录邮件服务器
        :param
        :return: imap连接
        """
        try:
            serv = imaplib.IMAP4_SSL(self.imap_server, self.imap_server_port)
            serv.login(self.user_id, self.password)
            return serv
        except Exception as e:
            print('邮箱登录失败:', e)
            exit(1)

    @staticmethod
    def loginout(conn):
        """
        登出邮件服务器
        :param conn: imap连接
        :return:
        """
        conn.close()
        conn.logout()

    @staticmethod
    def get_content(conn):
        """
        获取指定邮件，解析内容
        :param conn: imap连接
        :return: 数据内容
        """
        # 在连接服务器后，搜索之前，需要选择邮箱，默认select(mailbox='INBOX', readonly=False)
        conn.select()
        # 筛选符合条件的邮件，根据发件人过滤
        ret, data = conn.search(None, '(FROM "xx@yy.com")')
        # 邮件列表
        email_list = data[0].split()
        if len(email_list) == 0:
            print('收件箱为空，已退出')
            exit(1)
        pi = len(email_list) - 1
        print('邮件总数:', pi + 1)
        accts = []
        last_day = None
        while pi >= 0:
            pr, pd = conn.fetch(email_list[pi], '(RFC822)')
            pi -= 1
            if pr != 'OK':
                print('接收邮件{0}失败'.format(pi))
                break
            data = pd[0][1].decode('utf8')
            msg = email.message_from_string(data)
            sub = email.header.decode_header(msg["Subject"])[0][0].decode("utf-8")
            date = msg.get('date')
            ti = date.find(' +0800')
            if ti > 0:
                date = date[0:ti].strip()
            date = datetime.datetime.strptime(date, '%a, %d %b %Y %H:%M:%S')
            # 前一天的不再处理
            if last_day and (date.year != last_day.year or date.month != last_day.month or date.day != last_day.day):
                break
            # 主题匹配
            if not sub.endswith('xxxxxx'):
                continue
            sender = email.header.decode_header(msg["From"])[0][0]

            last_day = date
            # 通过walk可以遍历出所有的内容
            for part in msg.walk():
                if part.is_multipart():
                    continue
                # 内容类型
                content_type = part.get_content_type()
                # 如果是附件，这里就会取出附件的文件名，以下两种方式都可以获取
                # name = part.get_param("name")
                name = part.get_filename()
                # 附件
                if name:
                    print('发现附件', name)
                else:
                    # 文本内容
                    txt = part.get_payload(decode=False)
                    # 分别解释text/html和text/plain两种类型，纯文本解释起来较简单，两种类型内容一致
                    if content_type == 'text/html':
                        print('发现html内容')
                    elif content_type == 'text/plain':
                        # 纯文本格式为bytes，不同邮件服务器较统一
                        accts.append(txt)
        return accts