使用datacompy比较两个列表

需求: 判断DB的数据与EXCEL的数据是否完全一致

该需求用到的知识有点多

• pandas读取SQL
• pandas读取EXCEL
• datacompy比较列表
• pandas写EXCEL

开发前准备

pip install pymysql
pip install pandas
pip install sqlalchemy
pip install datacompy
pip install openpyxl

开发代码

• 连接DB并获取数据

    def do_db(self):
        engine = create_engine("mysql+pymysql://root:" + self.passwd + "@" + self.host + ":" + self.port + "/" + self.db)
        sql = "select lot_no as '批号' from ZM_TBL_DOMESTIC_GINNED_COTTON where DELETE_FLAG = '0'"

        self.df1 = pd.read_sql_query(sql, engine)
        print(self.df1)

• 读取EXCEL

    def do_excel(self):
        self.df2 = pd.read_excel(self.file_name, usecols=[0], sheet_name="Sheet1", keep_default_na=False, converters={'批号': str})
        self.df2 = self.df2.drop_duplicates() 
        print(self.df2)

PS：这里需要注意的是，使用datacompy比较的两个列表中不能又重复的数据，所以要使用self.df2.drop_duplicates()去重

• 比较列表,并将差异存入EXCEL

    def dict_compare(self):
        self.do_db()
        self.do_excel()

        compare = datacompy.Compare(self.df1, self.df2, join_columns=['批号'])

        # print(compare.matches())  # 最后判断是否相等，返回 bool
        # print(compare.report())  # 打印报告详情，返回 string
        # print(compare.report(sample_count=5000))  # 打印报告详情，返回 string

        df1_unq_rows = compare.df1_unq_rows
        df2_unq_rows = compare.df2_unq_rows

        writer = pd.ExcelWriter(self.file_name, engine='openpyxl')
        writer.book = load_workbook(self.file_name)
        df1_unq_rows.to_excel(writer, sheet_name='EXCEL缺少的数据')
        df2_unq_rows.to_excel(writer, sheet_name="DB缺少的数据")
        writer.save()
        writer.close()

查看datacompy文档

完整代码

#!/usr/bin/python3
# -*- encoding: utf-8 -*-
'''
@File        :检查.py
@Time        :2020/10/26 10:39:06
@Author      :He
@Software    :vsCode
'''


import pymysql
import time
import datetime
import uuid
import os
from sqlalchemy import create_engine
import pandas as pd
import datacompy
from openpyxl import load_workbook


class mysql_class:
    def __init__(self):

        self.host = 'IP'
        self.port = '端口'
        self.passwd = '密码'
        self.user = 'root'
        self.db = ''
        self.file_name = 'EXCEL.xlsx'

    def do_db(self):
        engine = create_engine("mysql+pymysql://root:" + self.passwd + "@" + self.host + ":" + self.port + "/" + self.db)
        sql = "select lot_no as '批号' from ZM_TBL_DOMESTIC_GINNED_COTTON where DELETE_FLAG = '0'"

        self.df1 = pd.read_sql_query(sql, engine)
        print(self.df1)

    def do_excel(self):
        self.df2 = pd.read_excel(self.file_name, usecols=[0], sheet_name="Sheet1", keep_default_na=False, converters={'批号': str})
        self.df2 = self.df2.drop_duplicates()
        print(self.df2)

    def getCurrentTime(self):
        return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))

    def dict_compare(self):
        self.do_db()
        self.do_excel()

        compare = datacompy.Compare(self.df1, self.df2, join_columns=['批号'])

        df1_unq_rows = compare.df1_unq_rows
        df2_unq_rows = compare.df2_unq_rows

        writer = pd.ExcelWriter(self.file_name, engine='openpyxl')
        writer.book = load_workbook(self.file_name)
        df1_unq_rows.to_excel(writer, sheet_name='EXCEL缺少的数据')
        df2_unq_rows.to_excel(writer, sheet_name="DB缺少的数据")
        writer.save()
        writer.close()


if __name__ == "__main__":
    os.chdir(os.path.abspath(os.path.dirname(__file__)))

    starttime = datetime.datetime.now()
    print(starttime)

    mysql_class = mysql_class()
    mysql_class.dict_compare()

    endtime = datetime.datetime.now()
    print(endtime)

    print('\n数据处理成功!所用时间为：' + str((endtime - starttime).seconds))