Excel数据导入mysql通用代码

深碍√TFBOYSˉ_ 2023-10-05 08:52 28阅读 0赞

#!/usr/bin/env python
    #-*- coding:utf-8 -*-
    '''
    @Time    : 2020/6/18
    @Author  : Frank Su
    @File    : excelToMysql.py
    '''
    
    import pymysql
    import xlrd
    from utils import sql_mysql
    from datetime import datetime
    from xlrd import xldate_as_tuple
    
    
    def excel_mysql(mysql_db,path_excel):
        # 读取excel中内容到数据库
        workbook = xlrd.open_workbook(path_excel)
        # 获取表中所有的sheet名称
        sheetNameList = workbook.sheet_names()
    
        dataList = []
        # 定义表名和字段信息
        name_table = ""
        params = ""
        # 循环将每个sheet导入指定数据库表中
        for sheetName in sheetNameList:
            name_table = ""
            params = ""
            # 通过excel名称和sheet名称来确定插入的数据库表
            print(sheetName)
            if path_excel.find("执法信息") != -1:
                name_table = "law_enforcement_info"
                params = "number, case_name, case_date, case_stage, filing_agency,Confiscated_amount,Execution_amount, cause_case, product_name, des_case"
            elif path_excel.find("上海召回") != -1:
                name_table = "recall_shanghai"
                params = "number, product_name, manufacturer, causeofDefect, account_recall"
            elif path_excel.find("监抽") != -1 and path_excel.find("地方") != -1 and sheetName.find("月") != -1:
                name_table = "supervision_local"
                params = "product_class,name,batch_all,batch_qualified,rate_pass,batch_unqualified,rate_fail,item_fail,city_src,date_check,date_release,url"
            sheet = workbook.sheet_by_name(sheetName)
            data_list = []
            nrows = sheet.nrows  # 行数
            ncols = sheet.ncols  # 列数
    
            all_content = []
            for i in range(1,nrows):
                row_content = []
                for j in range(ncols):
                    ctype = sheet.cell(i, j).ctype  # 表格的数据类型
                    cell = sheet.cell_value(i, j)
                    if ctype == 2 and cell % 1 == 0:  # 如果是整数
                        cell = str(int(cell))
                    elif ctype == 2 and cell % 1 != 0:  # 如果是百分比小数
                        cell = str("%.1f%%" % (cell * 100))
                    elif ctype == 3:
                        # 转成datetime对象
                        date = datetime(*xldate_as_tuple(cell, 0))
                        cell = date.strftime('%Y/%m/%d')
                    elif ctype == 4:
                        cell = True if cell == 1 else False
                    if j < 11:
                        row_content.append(cell)
    
                # print('[' + ','.join("'" + str(element) + "'" for element in row_content) + ']')
                all_content.append(row_content)
            dataList.append(all_content)
            for data in all_content:
    
                mysql_db.insertByList(name_table,params,data)
        print(dataList[0])
    
    
    
    if __name__ == "__main__":
        # 获取数据库链接
        mydb = sql_mysql.MysqldbHelper()
        # 定义excel表格路径
        path_excel = "F:/1resource/申腾/工作/2020上半年上海召回信息汇总表.xlsx"
    
        excel_mysql(mydb, path_excel)

下面这个是sql\_mysql的源码

#!/usr/bin/env python
    #-*- coding:utf-8 -*-
    '''
    @Time    : 2019/12/23
    @Author  : Frank Su
    @File    : sql_mysql.py
    '''
    
    #!/usr/bin/python3
    
    # -*- coding: utf-8 -*-
    import pymysql
    import re
    import time
    
    class MysqldbHelper(object): # 继承object类所有方法
    
    
    
    
        def __init__(self):
    
            # 构造方法：
            self.config = {
                'db': 'zhijian',
                'host': '',  # 你的数据库ip
                'port': 3306,
                'user': 'root',  # 用户名
                'passwd': 'root',  # 密码
                'charset': 'utf8',
                'cursorclass': pymysql.cursors.DictCursor
            }
            self.db = self.config['db']
            self.host = self.config['host']
            self.username = self.config['user']
            self.password = self.config['passwd']
            self.port = self.config['port']
            self.con = None
            self.cur = None
            try:
                self.con = pymysql.connect(**self.config)
                self.con.autocommit(1)
                # 所有的查询，都在连接 con 的一个模块 cursor 上面运行的
                self.cur = self.con.cursor()
            except:
                None
                #print("DataBase connect error,please check the db config.")
        # 建立连接
        def _conn(self):
            try:
                self.con = pymysql.connect(**self.config)
                self.con.autocommit(1)
                # 所有的查询，都在连接 con 的一个模块 cursor 上面运行的
                self.cur = self.con.cursor()
                return True
            except:
                #print("DataBase connect error,please check the db config.")
                return False
        # 测试链接是否正常
        def conn_test(self):
            try:
                #print('conn_test ping()')
                self.con.ping(reconnect=True)
                return True
            except pymysql.MySQLError as e:
                #print('Error Message: {}'.format(e))
                self._conn()
        # 重新连接
        def _reConn(self, num=28800, stime=3):  # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......
            _number = 0
            _status = True
            while _status and _number <= num:
                try:
                    self.con.ping()  # cping 校验连接是否异常
                    _status = False
                except:
                    if self._conn() == True:  # 重新连接,成功退出
                        _status = False
                        break
                    _number += 1
                    time.sleep(stime)  # 连接不成功,休眠3秒钟,继续循环，知道成功或重试次数结束
        # 关闭数据库连接
        def close(self):
            if not  self.con:
                self.con.close()
            else:
                None
                #print("DataBase doesn't connect,close connectiong error;please check the db config.")
    
        # 创建数据库
        def createDataBase(self,DB_NAME):
            # 创建数据库
            self.cur.execute('CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci' % DB_NAME)
            self.con.select_db(DB_NAME)
            #print('creatDatabase:' + DB_NAME)
    
        # 选择数据库
        def selectDataBase(self,DB_NAME):
            self.con.select_db(DB_NAME)
    
        # 获取数据库版本号
        def getVersion(self):
            self.cur.execute("SELECT VERSION()")
            return self.getOneData()
    
        # 获取上个查询的结果
        def getOneData(self):
            # 取得上个查询的结果，是单个结果
            data = self.cur.fetchone()
            return data
    
        # 创建数据库表
        def creatTable(self, tablename, attrdict, constraint):
            """创建数据库表
                args：
                    tablename  ：表名字
                    attrdict   ：属性键值对,{'book_name':'varchar(200) NOT NULL'...}
                    constraint ：主外键约束,PRIMARY KEY(`id`)
            """
            #　判断表是否存在
            if self.isExistTable(tablename):
                #print("%s is exit" % tablename)
                return
            sql = ''
            sql_mid = '`id` bigint(11) NOT NULL AUTO_INCREMENT,'
            for attr,value in attrdict.items():
                sql_mid = sql_mid + '`'+attr + '`'+' '+ value+','
            sql = sql + 'CREATE TABLE IF NOT EXISTS %s ('%tablename
            sql = sql + sql_mid
            sql = sql + constraint
            sql = sql + ') ENGINE=InnoDB DEFAULT CHARSET=utf8'
            #print('creatTable:'+sql)
            self.executeCommit(sql)
    
        def executeSql(self,sql=''):
            """执行sql语句，针对读操作返回结果集
    
                args：
                    sql  ：sql语句
            """
    
            try:
                self.cur.execute(sql)
                records = self.cur.fetchall()
                return records
            except pymysql.Error as e:
                error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1])
                print(error)
    
        def executeCommit(self,sql=''):
            """执行数据库sql语句，针对更新,删除,事务等操作失败时回滚
    
            """
            try:
                self.cur.execute(sql)
                self.con.commit()
            except pymysql.Error as e:
                self.con.rollback()
                error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1])
                print("error:", error)
                return error
    
        def insertByList(self, tablename,params, dataList):
            """创建数据库表
    
                args：
                    tablename  ：表名字
                    paramsList        ：以，间隔的list列表
            """
            # 测试连接是否断开，并重新连接
            self.conn_test()
            sql_res = ""
            str = ""
            print(dataList)
            # for data_list in dataList:
            str = ""
            for index,data in enumerate(dataList):
                if index != len(dataList) - 1:
                    str += ("'" + data + "',")
                else:
                    str += ("'" + data + "'")
            print(str)
            sql = "insert into " + tablename + " (" + params + ")" + " values(" + str + ");"
            # sql_res += sql
            print('_insert:'+sql)
            try:
                self.executeCommit(sql)
            except Exception as e:
                # None
                print(e)
    
        def insert(self, tablename, params):
            """创建数据库表
    
                args：
                    tablename  ：表名字
                    key        ：属性键
                    value      ：属性值
            """
            # 测试连接是否断开，并重新连接
            self.conn_test()
            key = []
            value = []
            for tmpkey, tmpvalue in params.items():
                key.append(tmpkey)
                if isinstance(tmpvalue, str):
                    value.append("\'" + tmpvalue + "\'")
                else:
                    value.append(tmpvalue)
    
            try:
                attrs_sql = '(' + ','.join(key) + ')'
                values_sql = ' values(' + ','.join(value) + ')'
            except Exception as e:
                # None
                print(e)
                # print("****************************")
    
            sql = 'insert into %s' % tablename
            sql = sql + attrs_sql + values_sql
            print('_insert:' + sql)
            try:
                self.executeCommit(sql)
            except Exception as e:
                # None
                print(e)
    
        def select(self, tablename, cond_dict='', order='', fields='*'):
            """查询数据
    
                args：
                    tablename  ：表名字
                    cond_dict  ：查询条件
                    order      ：排序条件
    
                example：
                    #print mydb.select(table)
                    #print mydb.select(table, fields=["name"])
                    #print mydb.select(table, fields=["name", "age"])
                    #print mydb.select(table, fields=["age", "name"])
            """
            consql = ' '
            if cond_dict!='':
                for k, v in cond_dict.items():
                    # 如果出现is null字段则单独处理
                    if v == "is null":
                        consql = consql + k +' is null'+ ' and'
                    else:
                        consql = consql+'`'+k +'`'+ '=' + '"'+v + '"' + ' and'
            consql = consql + ' 1=1 '
            if fields == "*":
                sql = 'select * from %s where ' % tablename
            else:
                if isinstance(fields, list):
                    fields = ",".join(fields)
                    sql = 'select %s from %s where ' % (fields, tablename)
                else:
                    None
                    #print("fields input error, please input list fields.")
            sql = sql + consql + order
            ##print('select:' + sql)
            return self.executeSql(sql)
    
    
        def selectCount(self, tablename, cond_dict='', order='', fields='*'):
            """查询数据数目
    
                args：
                    tablename  ：表名字
                    cond_dict  ：查询条件
                    order      ：排序条件
    
                example：
                    #print mydb.select(table)
                    #print mydb.select(table, fields=["name"])
                    #print mydb.select(table, fields=["name", "age"])
                    #print mydb.select(table, fields=["age", "name"])
            """
            consql = ' '
            if cond_dict!='':
                for k, v in cond_dict.items():
                    consql = consql+'`'+k +'`'+ '=' + '"'+v + '"' + ' and'
            consql = consql + ' 1=1 '
            if fields == "*":
                sql = 'select count(*) from %s where ' % tablename
            else:
                if isinstance(fields, list):
                    fields = ",".join(fields)
                    sql = 'select %s from %s where ' % (fields, tablename)
                else:
                    None
                    #print("fields input error, please input list fields.")
            sql = sql + consql + order
            print('select:' + sql)
            return self.executeSql(sql)
    
        def insertMany(self,table, attrs, values):
            """插入多条数据
    
                args：
                    tablename  ：表名字
                    attrs        ：属性键
                    values      ：属性值
    
                example：
                    table='test_mysqldb'
                    key = ["id" ,"name", "age"]
                    value = [[101, "liuqiao", "25"], [102,"liuqiao1", "26"], [103 ,"liuqiao2", "27"], [104 ,"liuqiao3", "28"]]
                    mydb.insertMany(table, key, value)
            """
            values_sql = ['%s' for v in attrs]
            attrs_sql = '('+','.join(attrs)+')'
            values_sql = ' values('+','.join(values_sql)+')'
            sql = 'insert into %s'% table
            sql = sql + attrs_sql + values_sql
            #print('insertMany:'+sql)
            try:
                #print(sql)
                for i in range(0,len(values),20000):
                        self.cur.executemany(sql,values[i:i+20000])
                        self.con.commit()
            except pymysql.Error as e:
                self.con.rollback()
                error = 'insertMany executemany failed! ERROR (%s): %s' %(e.args[0],e.args[1])
                #print(error)
    
        def delete(self, tablename, cond_dict):
            """删除数据
    
                args：
                    tablename  ：表名字
                    cond_dict  ：删除条件字典
    
                example：
                    params = {"name" : "caixinglong", "age" : "38"}
                    mydb.delete(table, params)
    
            """
            # 测试连接是否断开，并重新连接
            self.conn_test()
            consql = ' '
            if cond_dict!='':
                for k, v in cond_dict.items():
                    if isinstance(v, str):
                        v = "\'" + v + "\'"
                    consql = consql + tablename + "." + k + '=' + v + ' and '
            consql = consql + ' 1=1 '
            sql = "DELETE FROM %s where%s" % (tablename, consql)
            #print(sql)
            return self.executeCommit(sql)
    
        def update(self, tablename, attrs_dict, cond_dict):
            """更新数据
    
                args：
                    tablename  ：表名字
                    attrs_dict  ：更新属性键值对字典
                    cond_dict  ：更新条件字典
    
                example：
                    params = {"name" : "caixinglong", "age" : "38"}
                    cond_dict = {"name" : "liuqiao", "age" : "18"}
                    mydb.update(table, params, cond_dict)
    
            """
            # 测试连接是否断开，并重新连接
            self.conn_test()
            attrs_list = []
            consql = ' '
            for tmpkey, tmpvalue in attrs_dict.items():
                attrs_list.append("`" + tmpkey + "`" + "=" +"\'" + tmpvalue + "\'")
            attrs_sql = ",".join(attrs_list)
            print("attrs_sql:", attrs_sql)
            if cond_dict!='':
                for k, v in cond_dict.items():
                    if isinstance(v, str):
                        v = "\'" + v + "\'"
                    consql = consql + "`" + tablename +"`." + "`" + k + "`" + '=' + v + ' and '
            consql = consql + ' 1=1 '
            sql = "UPDATE %s SET %s where%s" % (tablename, attrs_sql, consql)
            print(sql)
            return self.executeCommit(sql)
    
        def dropTable(self, tablename):
            """删除数据库表
    
                args：
                    tablename  ：表名字
            """
            sql = "DROP TABLE  %s" % tablename
            self.executeCommit(sql)
    
        def deleteTable(self, tablename):
            """清空数据库表
    
                args：
                    tablename  ：表名字
            """
            sql = "DELETE FROM %s" % tablename
            #print("sql=",sql)
            self.executeCommit(sql)
    
        def isExistTable(self, tablename):
            """判断数据表是否存在
    
                args：
                    tablename  ：表名字
    
                Return:
                    存在返回True，不存在返回False
            """
            sql = "select * from %s" % tablename
            result = self.executeCommit(sql)
            if result is None:
                return True
            else:
                if re.search("doesn't exist", result):
                    return False
                else:
                    return True
    
    
    
    if __name__ == "__main__":
        a = 1
        # 定义数据库访问参数
        # config = {
        #     'host': '47.100.55.66',
        #     'port': 3306,
        #     'user': 'root',
        #     'passwd': 'Su19951212@',
        #     'charset': 'utf8',
        #     'cursorclass': pymysql.cursors.DictCursor
        # }
        #
        # 初始化打开数据库连接
        mydb = MysqldbHelper()
    
        # 打印数据库版本
        #print(mydb.getVersion())
    
        # # 创建数据库
        # DB_NAME = input('输入要创建的数据库名：')
        # mydb.createDataBase(DB_NAME)
        #
        # # 选择数据库
        # #print("========= 选择数据库%s ===========" % DB_NAME)
        # mydb.selectDataBase(DB_NAME)
        #
        # #创建表
        # TABLE_NAME = input('输入要创建的数据表名：')
        # #print("========= 选择数据表%s ===========" % TABLE_NAME)
        # '''例如
        #     CREATE TABLE `lisi` (
        #       `id` bigint(11) NOT NULL auto_increment,
        #       `name` varchar(30) NOT NULL,
        #       PRIMARY KEY  (`id`)
        #     ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
        # '''
        # #print("========= 设置表的字段和属性 ===========" )
        # attrdict = {
        #     'name':'varchar(30) NOT NULL'
        # }
        # constraint = "PRIMARY KEY(`id`)"
        # mydb.creatTable(TABLE_NAME,attrdict,constraint)
        # #print('首次创建以后，请手动刷新一下数据库表页面。')
        #
        # # 插入纪录
        # #print("========= 单条数据插入 ===========")
        # params = {}
        # for i in range(5):
        #     params.update({"name":"testuser"+str(i)}) # 生成字典数据，循环插入
        #     #print(params)
        #     mydb.insert(TABLE_NAME, params)
        #     #print("")
        #
        # # 批量插入数据
        # #print("========= 多条数据同时插入 ===========")
        # insert_values = []
        # for i in range(5):
        #     # values.append((i,"testuser"+str(i)))
        #     insert_values.append([u"测试用户"+str(i)]) # 插入中文数据
        # #print(insert_values)
        # insert_attrs = ["name"]
        # mydb.insertMany(TABLE_NAME,insert_attrs, insert_values)
        #
        # # 数据查询
        # #print("========= 数据查询 ===========")
        # #print(mydb.select(TABLE_NAME, fields=["id", "name"]))
        # #print(mydb.select(TABLE_NAME, cond_dict = {'name':'测试用户1'},fields=["id", "name"]))
        # #print(mydb.select(TABLE_NAME, cond_dict = {'name':'测试用户2'},fields=["id", "name"],order="order by id desc"))
        #
        # # 删除数据
        # #print("========= 删除数据 ===========")
        # delete_params = {"name": "测试用户2"}
        # mydb.delete(TABLE_NAME, delete_params)
        #
        # # 更新数据
        # #print("========= 更新数据 ===========")
        # update_params = {"name": "测试用户99"}   # 需要更新为什么值
        # update_cond_dict = {"name": "测试用户3"}  # 更新执行的查询条件
        # mydb.update(TABLE_NAME, update_params, update_cond_dict)
        #
        # # 删除表数据
        # #print("========= 删除表数据 ===========")
        # mydb.deleteTable(TABLE_NAME)
        #
        # # 删除表
        # #print("========= 删除表===================")
        # mydb.dropTable(TABLE_NAME)

你需要将代码中文件路径改为自己的就可以，这个仅供大家参考，我这里只是粗略的写了一下，也可以在我这个基础上进一步优化，