# -*- coding: UTF-8 -*-
import ast
import json
import os
import sys
import time
import xlrd
import xlwt
import pandas as pd
from xlutils.copy import copy
from collections import OrderedDict
from aa_demo.base.logger import *
from jsonpath_ng import parse
class xlsx_json_tools:
""" Execl 操作工具 """
def __init__(self, dir_path_name='', read_file_name='', new_file_name='', sheet_name='', datas={}):
"""
Execl 操作工具
:param dir_path_name: 需要保存到的文件夾名稱 路徑示例:項目名/模塊名/data/dir_path_name/xxx.xslx
:param read_file_name: 讀取文件名
:param new_file_name: 存儲文件名
:param sheet_name:execl sheet頁名稱, 如果為空 取 read_file_name 文件名
:param datas: 傳入空字典{},即datas初始值為{} 作用與方法recur_data()
"""
self.read_file_name = read_file_name
self.datas = datas
# 獲取根目錄
proj_path = os.path.abspath(os.path.dirname(__file__))[:os.path.abspath(os.path.dirname(__file__)).find('base')]
# xlsx 文件路徑
self.xlsx_path = os.path.join(proj_path, 'data\\{}\\{}.xlsx'.format("xlsx" if dir_path_name == '' else dir_path_name, self.read_file_name))
self.json_path = os.path.join(proj_path, 'data\\{}\\{}.json'.format("json" if dir_path_name == '' else dir_path_name, self.read_file_name))
self.new_xlsx_path = os.path.join(proj_path, 'data\\{}\\{}.xlsx'.format("xlsx" if dir_path_name == '' else dir_path_name, self.read_file_name if new_file_name == '' else new_file_name))
self.new_json_path = os.path.join(proj_path, 'data\\{}\\{}.json'.format("json" if dir_path_name == '' else dir_path_name, self.read_file_name if new_file_name == '' else new_file_name))
self.sheet_name = self.read_file_name if sheet_name == "" else sheet_name
# 判斷目錄是否存在
self.xlsx_path if os.path.exists(os.path.dirname(self.xlsx_path)) else os.makedirs(os.path.dirname(self.xlsx_path))
self.json_path if os.path.exists(os.path.dirname(self.json_path)) else os.makedirs(os.path.dirname(self.json_path))
# logs.debug("xlsx path : {}".format(self.xlsx_path))
# logs.debug("json path : {}".format(self.json_path))
def recur_data(self, data: dict, result="", is_json_layer='on'):
"""
多維/嵌套字典數據無限遍歷,獲取所有key層和value(key以層級key值拼接)
:param data: 需要遍歷獲取所有層級key/value值的json串
:param result: 層級名稱
:param is_json_layer: 是否逐層獲取,on 逐層獲取key-value 否則不逐層獲取
:return:
"""
if is_json_layer == 'on':
# 使用isinstance檢測數據類型:字典類型
if isinstance(data, dict):
for k, v in data.items():
if result == "":
self.recur_data(v, k, is_json_layer)
else:
self.recur_data(v, result+"_{}".format(k), is_json_layer)
# 列表或元組類型
elif isinstance(data, (list, tuple)):
for i in range(len(data)):
# 自我調用實現無限遍歷, 並判斷是否逐層key拼接命名
self.recur_data(data[i], result, is_json_layer)
else:
# print(result + "=" + str(data))
# self.datas[result] = str(data)
self.datas.setdefault(result, []).append(str(data))
else:
for key, value in data.items():
# self.datas.update({key: [value]})
self.datas.setdefault(key, []).append(str(value))
def read_json_data(self, json_path='', layer_by_layer="on"):
"""
讀取json內容
:param json_path: json文件 路徑+文件名
:param layer_by_layer: 逐層獲取key-value
:return:
"""
if json_path == '':
json_path = self.json_path
# 讀取json文件內容
json_con = json.load(open(json_path, "r", encoding="utf-8"), object_pairs_hook=OrderedDict)
# logs.debug("json_content:\n{}".format(json_con))
# 區分是否為execl轉換的數據
if "execl_content" in json_con.keys():
self.recur_data(data=json_con.get("execl_content"), is_json_layer=layer_by_layer)
else:
self.recur_data(data=json_con, is_json_layer=layer_by_layer)
json_content = self.datas
logs.debug("獲取到的json內容:{}".format(json_content))
return json_content
def write_json_data(self, content):
try:
json.dump(content, open(self.new_json_path, "w+", encoding="utf-8"), ensure_ascii=False, sort_keys=True, indent=4)
logs.info("寫入.json文件成功, 寫入文件:{}".format(self.new_json_path))
except Exception as e:
logs.error("寫入異常原因:{}".format(e))
def read_execl_data(self, read_mode='row', read_num: int = 1) -> list:
"""
讀取execl數據
:param read_mode: 讀取數據方式: row 按行讀取, col 按列讀取
:param read_num: 讀取數據條數 Tips:當讀取條數超過execl條數,則取全部條數
:return: [數據行數, 讀取到的數據集合]
"""
try:
# 打開文件
wb = xlrd.open_workbook(self.xlsx_path)
# 根據sheet名稱獲取sheet頁內容
sheet = wb.sheet_by_name(self.sheet_name)
# 獲取行數、列數
rowNums = sheet.nrows
colNums = sheet.ncols
# logs.debug("行數:{}, 列數:{}".format(rowNums, colNums))
all_contents = []
if read_mode.lower() == "row":
# 獲取第一行內容
# contents = sheet.row_values(0)
# logs.debug("首行內容:{}".format(contents))
# 獲取case條數
runNums = read_num+1 if rowNums > read_num+1 > 1 else rowNums
# logs.debug("run = {}".format(runNums-1))
# 獲取數據
for i in range(0, runNums):
contents = sheet.row_values(i)
all_contents.append(contents)
# logs.debug("所有行List = {}".format(all_contents))
elif read_mode.lower() == "col":
# 獲取第一列內容
# contents = sheet.col_values(0)
# logs.debug("首列內容:{}".format(contents))
# 獲取case條數
runNums = read_num+1 if colNums > read_num+1 > 1 else colNums
# logs.debug("run = {}".format(runNums-1))
# 獲取數據
for i in range(0, runNums):
contents = sheet.col_values(i)
all_contents.append(contents)
# logs.debug("所有列List = {}".format(all_contents))
else:
sys.exit("僅支持[row, col], 請檢查參數:read_mode")
except Exception as e:
runNums = 0
all_contents = None
logs.error("讀取數據出現異常, Exception Casue: {}".format(e))
finally:
logs.debug("獲取到的execl內容:{}".format(all_contents))
return runNums, all_contents
def copy_execl_data(self, read_mode='row', params_name: list = None, params_values: list = None) -> list:
"""
將數據寫入execl Tips:文件不存在則新建,存在則追加(如存在請確保參數名及參數與原數據一一對應,否則會出現覆蓋或錯位填充)
:param read_mode: 寫入方式 [row 按行寫入, col 按列寫入]
:param params_name: 參數名
:param params_values: 參數值 [[], []]
:return:
"""
sys.exit("Execl 寫入文件失敗,請檢查傳入參數是否符合要求") if params_name is None or params_values is None or len(params_name) != len(params_values[0]) else ""
try:
# 打開Execl
wb = xlrd.open_workbook(self.new_xlsx_path)
except FileNotFoundError:
# 新建工作簿
wb = xlwt.Workbook('utf-8')
# 創建工作表
wb.add_sheet(self.sheet_name)
# 保存到新建工作簿
wb.save(self.new_xlsx_path)
# 打開文件
wb = xlrd.open_workbook(self.new_xlsx_path)
except Exception as e:
sys.exit(e)
finally:
sheet_names = wb.sheet_names() # 獲取文件 sheet 列表
if self.sheet_name in sheet_names:
sheet = wb.sheet_by_name(self.sheet_name)
nrows, ncols = sheet.nrows, sheet.ncols
else:
nrows, ncols = 0, 0
# logs.debug("行={}, 列={}".format(nrows, ncols))
# 向 execl 寫入內容, 如果sheetname不存在,則創建副本
cp_wb = copy(wb)
cp_ws = cp_wb.add_sheet(self.sheet_name) if self.sheet_name not in sheet_names else cp_wb.get_sheet(self.sheet_name)
# 根據寫入方式,copy 直接修改單元格內容
if read_mode.lower() == "row":
# 循環寫入keys到表格第一行
for i in range(len(params_name)):
cp_ws.write(0, i, params_name[i], self.__title_style())
# 循環寫入value 到表格第二行
for x in range(len(params_values)):
values = params_values[x]
for y in range(len(values)):
# values 可能為dict 所以需要轉換為str
if nrows == 0:
cp_ws.write(x+1, y, str(values[y]))
else:
cp_ws.write(x+nrows, y, str(values[y]))
elif read_mode.lower() == "col":
# 循環寫入keys到表格第一行
for i in range(len(params_name)):
cp_ws.write(i, 0, params_name[i], self.__title_style())
# 循環寫入value 到表格第二行
for x in range(len(params_values)):
values = params_values[x]
for y in range(len(values)):
# values 可能為dict 所以需要轉換為str
if ncols == 0:
cp_ws.write(y, x+1, str(values[y]))
else:
cp_ws.write(y, x+ncols, str(values[y]))
else:
sys.exit("僅支持[row, col], 請檢查參數:read_mode")
try:
# 保存文件
cp_wb.save(self.new_xlsx_path)
logs.info('Execl 文件寫入成功, 存儲文件名為:{} \n'.format(self.new_xlsx_path))
except FileExistsError:
logs.error('Execl 文件寫入敗, 請關閉名稱為{}文件'.format(self.new_xlsx_path))
except PermissionError:
logs.error('Execl 文件寫入失敗, 請關閉名稱為{}文件'.format(self.new_xlsx_path))
except Exception as e:
logs.error("Execl 文件寫入失敗,失敗原因:{}".format(e))
def write_execl_data(self, params_name: list = None, params_values: list = None) -> list:
"""
將數據寫入execl文件中 (文件存在則覆蓋原數據, 不存在則新建)
:param params_name:
:param params_values:
:return:
"""
sys.exit("Execl 生成文件失敗,請檢查傳入參數是否符合要求") if params_name is None or params_values is None or len(params_name) != len(params_values[0]) else ""
# list轉dataframe,data傳入list格式數據,如:[[],[],[]]
# 寫入數據
df = pd.DataFrame(params_values, columns=params_name)
# 保存到本地excel
try:
df.to_excel(self.new_xlsx_path, index=False, sheet_name=self.sheet_name)
logs.info('Execl 生成文件成功, 存儲文件名為:{} \n'.format(self.new_xlsx_path))
except FileExistsError:
logs.error('Execl 生成文件失敗, 請關閉名稱為{}文件'.format(self.new_xlsx_path))
except PermissionError:
logs.error('Execl 生成文件失敗, 請關閉名稱為{}文件'.format(self.new_xlsx_path))
except Exception as e:
logs.error("Execl 生成文件失敗,失敗原因:{}".format(e))
def read_execl_to_json(self, read_mode='row', read_num=1, is_save_json='off') -> list:
"""
讀取execl數據並轉換為json
:param read_mode: 讀取數據方式: row 按行讀取, col 按列讀取
:param read_num: 讀取數據條數 Tips:當讀取條數超過execl條數,則取全部條數
:param is_save_json: 是否保存到json文件;[on 保存,off 不保存]
:return: [{}, {}]
"""
try:
rd = self.read_execl_data(read_mode=read_mode, read_num=read_num)
runNums = rd[0]
all_contents = rd[1]
# 拆分json結構 keys
r = []
all_data = {}
for j in range(1, runNums):
key_name = all_contents[0]
key_value = all_contents[j]
# 將兩個list(表頭和case行) 拼成dict
s = {}
for x in range(0, len(key_value)):
if key_value[x] != "刪除":
s[key_name[x]] = ast.literal_eval(key_value[x]) if str(key_value[x]).startswith("[") or str(key_value[x]).startswith("{") else key_value[x]
# 將數據存儲到r列表中
r.extend([json.dumps(s, ensure_ascii=False)])
# 將全部數據存儲到all_data中
all_data.setdefault("execl_content", []).append(ast.literal_eval(json.dumps(s, ensure_ascii=False)))
except Exception as e:
r = None
all_data = None
logs.error("讀取數據出現異常, Exception Casue: {}".format(e))
return r
finally:
logs.debug(all_data)
logs.debug(type(all_data))
# 判斷是否需要存儲到文件中
if is_save_json == "on":
# json.dump(all_data, open(self.new_json_path, "w+", encoding="utf-8"), ensure_ascii=False, sort_keys=True, indent=4)
# logs.info("寫入.json文件成功, 寫入文件:{}".format(self.json_path))
self.write_json_data(all_data)
# logs.debug("寫入.json文件內容:{}".format(all_data))
return r
def read_json_to_execl(self, to_execl_mode="on", read_json_layer='on') -> list:
"""
讀取json數據保存到execl
:param to_execl_mode: on 無文件則生成,有則覆蓋, off 無文件則生成,有則追加
:param read_json_layer: on json逐層獲取key-value, off 僅獲取一層key-value
:return:
"""
# 讀取json數據
json_dic = self.read_json_data(layer_by_layer=read_json_layer)
# logs.debug("獲取到的json內容:{}".format(json_dic))
# 獲取key-value
keys_list = list(json_dic.keys())
values_list = list(json_dic.values())
data_line = len(values_list[0]) if values_list != [] else None
line_list = []
# logs.debug('所有層級keys:{}'.format(keys_list))
# logs.debug('所有層級values:{}'.format(values_list))
# logs.debug("json共包含{}行數據信息".format(data_line))
if data_line is not None:
for i in range(data_line):
line = []
for j in range(len(keys_list)):
# logs.debug("{}:{}".format(keys_list[j], values_list[j]))
line.append(values_list[j][i] if values_list[j] is not None else "") # 單行
line_list.extend([line]) # 多行
# logs.debug("獲取單行信息line={}".format(line))
# logs.debug("獲取多行信息line_list={}".format(line_list))
# logs.debug("寫入xlsx文件的內容:{}".format([keys_list, line_list]))
if to_execl_mode == "on":
self.write_execl_data(keys_list, line_list)
else:
self.copy_execl_data(read_mode="row", params_name=keys_list, params_values=line_list)
return [keys_list, line_list]
@staticmethod
def __forward(letter, jump):
"""字母表上的循環迭代器"""
if letter.islower():
start_character = ord('a')
else:
start_character = ord('A')
start = ord(letter) - start_character
offset = ((start + jump) % 26) + start_character
result = chr(offset)
return result
@staticmethod
def __title_style():
"""xlsx表格標題樣式"""
# 新建一個樣式表
style = xlwt.XFStyle()
# 新建一個字體格式對象
font = xlwt.Font()
font.name = '微軟雅黑'
font.bold = True
font.height = 180 # 360
# 把字體放入樣式表
style.font = font
# 新建一個邊框樣式
borders = xlwt.Borders()
# 邊框都是細線
borders.top = xlwt.Borders.THIN
borders.bottom = xlwt.Borders.THIN
borders.left = xlwt.Borders.THIN
borders.right = xlwt.Borders.THIN
# 把邊框給樣式表
style.borders = borders
# 新建一個對齊
alingnment = xlwt.Alignment()
# 垂直和水平對齊
alingnment.horz = xlwt.Alignment.HORZ_CENTER
alingnment.vert = xlwt.Alignment.VERT_CENTER
# 把對齊給樣式表
style.alignment = alingnment
# 設置背景顏色
pattern = xlwt.Pattern()
# 背景顏色
pattern.pattern = xlwt.Pattern.SOLID_PATTERN
pattern.pattern_fore_colour = 48
# 把背景顏色給樣式表
style.pattern = pattern
return style
if __name__ == "__main__":
"""debug"""
params_name = ["name", "age", "addr"]
# params_values = [["zhangsan", 18, "shanghai"], ["lisi", 22, "beijing"], ["wugo", 30, "harbin"]]
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo", sheet_name="demo").read_execl_data(read_mode="row", read_num=0)
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo", sheet_name="demo").read_execl_data(read_mode="col", read_num=0)
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo", sheet_name="demo").write_execl_data(read_mode="row", params_name=params_name, params_values=params_values)
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo", sheet_name="demo").copy_execl_data(read_mode="row", params_name=params_name, params_values=params_values)
# xlsx_json_tools(dir_path_name="demo", read_file_name="need", sheet_name="demo").copy_execl_data(read_mode="col", params_name=params_name, params_values=params_values)
# xlsx_json_tools(dir_path_name="demo", read_file_name="need", sheet_name="demo").write_execl_data(params_name=params_name, params_values=params_values)
# """ json <--> xlsx 轉換 """
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo").read_execl_data(read_mode="row", read_num=0)
xlsx_json_tools(dir_path_name="demo", read_file_name="demo", new_file_name='demo1', sheet_name="demo").read_execl_to_json(read_mode="row", read_num=0, is_save_json="on")
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo", new_file_name='demo2', sheet_names="demo").read_execl_to_json(read_mode="col", read_num=0, is_save_json="on")
# xlsx_json_tools(dir_path_name="demo", read_file_name="demo").read_json_data(layer_by_layer="on.")
xlsx_json_tools(dir_path_name="demo", read_file_name="demo", new_file_name='demo2').read_json_to_execl(to_execl_mode='on', read_json_layer='on.')