Python Excel 处理函数 以及 OracleHelper 工具类

Python Excel 处理函数 以及 OracleHelper 工具类

def read_excel(file_path,index=0):
    import xlrd
    book = xlrd.open_workbook(file_path)
    sheet = book.sheet_by_index(index)
    return sheet

def excel_to_json(file_path,columns,index=0,header_index=0):
    import json
    sheet = read_excel(file_path,index)
    
    rows = sheet.nrows
    cols = sheet.ncols 
    print(sheet.name,rows)
    header = sheet.row_values(0)  # 获取第一行的值 
    print(header)
    data = []   
    for i in range(rows):
        if i == header_index:
            continue
        a = []
        row = sheet.row_values(i) 
        for  c in columns:
            idx = header.index(c)
            a.append(row[idx] ) 
        data.append(a)
    return data

实际项目：

from OracleHelper import OracleHelper
def read_excel(file_path,index=0):
    import xlrd
    book = xlrd.open_workbook(file_path)
    sheet = book.sheet_by_index(index)
    return sheet

def excel_to_json(file_path,columns,index=0,header_index=0):
    import json
    sheet = read_excel(file_path,index)
    
    rows = sheet.nrows
    cols = sheet.ncols 
    print(sheet.name,rows)
    header = sheet.row_values(0)  # 获取第一行的值 
    print(header)
    data = []   
    for i in range(rows):
        if i == header_index:
            continue
        a = []
        row = sheet.row_values(i) 
        for  c in columns:
            idx = header.index(c)
            a.append(row[idx] ) 
        data.append(a)
    return data

def jnh_yf():
    filepath = r"C:UserswangjunDesktop2021年终嘉年华数据2022云帆合约计划.xlsx"
    columns = ['CUST_NAME','AUTH_ID','近3个月月均','建议合约类型']
    sheet_index = 1
    a = excel_to_json(filepath,columns,sheet_index,0)
    
    helper = OracleHelper("IP","1521","USERNAME","PASSWORD","DB")
    # a = helper.get_data("select * from t_user")
    # print(a) 
    for c in a:
        print(c[1])
        helper.exec('''INSERT INTO ta_config01 (cust_id, activity_name, product, info) VALUES ( :1, :2, :3, :4)''' ,(
            int(c[1]),'yunfan',c[3],c[2]
        ))  # 客户ID，活动名称，产品，信息
    
    # print(a)  # 返回的是一个列表
jnh_yf()

OracleHelper.py

import cx_Oracle
class OracleHelper:
    def __init__(self, host, port, user, pwd, dbname):
        self.host = host
        self.port = port
        self.user = user
        self.pwd = pwd
        self.dbname = dbname

    def __str__(self):
        return "host:%s,port:%s,user:%s,pwd:%s,dbname:%s" % (self.host, self.port, self.user, self.pwd, self.dbname)

    def __repr__(self):
        return "host:%s,port:%s,user:%s,pwd:%s,dbname:%s" % (self.host, self.port, self.user, self.pwd, self.dbname)

    def get_conn(self):
        """
        获取一个数据库连接
        :return:
        """ 
        try:
            conn = cx_Oracle.connect(self.user, self.pwd, self.host + ":" + self.port + "/" + self.dbname)
            return conn
        except Exception as e:
            print("数据库连接失败：", e)

    def close_conn(self, conn):
        """
        关闭数据库连接
        :param conn:
        :return:
        """
        try:
            if conn:
                conn.close()
        except Exception as e:
            print("关闭数据库连接失败：", e)

    def get_data(self, sql):
        """
        查询数据
        :param sql:
        :return:
        """
        try:
            conn = self.get_conn()
            cur = conn.cursor()
            cur.execute(sql)
            data = cur.fetchall()
            self.close_conn(conn)
        except Exception as e:
            print("查询数据失败：", e)
        finally:
            return data
    def exec(self,sql,data=None):
        print(sql,data)
        conn = self.get_conn()
        cur = conn.cursor()
        if data:
            cur.execute(sql,data)
        else:
            cur.execute(sql)
        conn.commit()
        cur.close()