软件需求
和银行等金融机构合作过程中，经常会有还款计划、放款文件等定时推送的文件，以常见的分隔符进行分隔开，为了在系统出现问题时，快速查找异常数据，写了一个小工具解析这些文件并写入到excel中。
image-20240403171558023
软件功能
将常见分隔符的文件解析后写入到excel中
软件界面
image-20240403170655979
软件源码
- 使用前请安装openpyxl,
pip install openpyxl，可使用pyinstaller打包成exe文件。
import os
import datetime
import logging
from openpyxl import Workbook
import tkinter as tk
from tkinter import filedialog,messagebox
from tkinter.ttk import *
class FileParser(object):
def __init__(self):
pass
@staticmethod
def getAllFiles(folder,data_str,suffix=".txt"):
'''获取文件路径下的所有文件'''
filepath_list = []
for root,folder_names, file_names in os.walk(folder):
for file_name in file_names:
file_path = root + os.sep + file_name
if data_str != '' or data_str is not None:
if data_str in file_name and os.path.splitext(file_path)[-1]==suffix:
filepath_list.append(file_path)
else:
if os.path.splitext(file_path)[-1]==suffix:
filepath_list.append(file_path)
if len(filepath_list)==0:
messagebox.showinfo('文件不存在',"该目录下不存在包含{}且后缀为{}的文件，请检查！".format(data_str,suffix))
return
file_path = sorted(file_path, key=str.lower)
return filepath_list
def batchParseDataFile(self,data_str,logbox,start_date,end_date,date_format,source_dir,target_dir,source_suffix=".txt",data_sep="|",data_start_row=1,data_end_flag="",start_row=1,start_col=1):
'''
获取数据文件中的有效数据
:param start_date:解析的文件的开始日期
:param end_date:解析的结束日期
:param date_format:日期格式
:param source_dir:需要解析的文件的文件夹路径
:param target_dir:生成excel文件的文件夹路径
:param data_sep:数据分隔符
:param data_start_row:有效数据的开始行
:param data_end_flag:有效数据结束标识
:param start_row:写入excel的开始行
:param start_col:写入excel的开始列
:param table_head:有效数据是否包含表头
'''
self.log = logbox
start=datetime.datetime.strptime(start_date,date_format)
end=datetime.datetime.strptime(end_date,date_format)
source_filelist = FileParser.getAllFiles(source_dir,data_str,source_suffix)
if source_filelist is None or len(source_filelist)==0:
self.log.info("该目录{}下不存在文件".format(source_dir))
return
if start > end:
self.log.info("开始日期{}大于结束日期{}".format(start_date,end_date))
return
files = []
null_files = []
starttime = datetime.datetime.now()
while start<=end:
for fname in source_filelist:
if start.strftime(date_format) in fname:
if os.stat(fname).st_size==0:
null_files.append(fname)
else:
files.append(fname)
basename = os.path.basename(fname)
file_date = start.strftime(date_format)
self.log.info("{}日期的{}文件正在解析".format(file_date,basename))
target_basename = basename.split(".")[0]+".xlsx" if file_date in basename else basename.split(".")[0]+"_"+file_date+".xlsx"
target_fname = os.path.join(target_dir,target_basename)
self.log.info("生成{}日期的excel文件{}".format(file_date,target_basename))
self.parseDataFile(fname,target_fname,data_sep=data_sep,data_start_row=data_start_row,data_end_flag=data_end_flag,start_row=start_row,start_col=start_col)
start+=datetime.timedelta(days=1)
endtime=datetime.datetime.now()
costtime = round((endtime - starttime).total_seconds(),2)
if len(files)==0:
self.log.info("{}目录不存在{}-{}时间区间内的目标文件".format(source_dir,start_date,end_date))
else:
messagebox.showinfo('解析完成提示',"文件解析完成，{}-{}时间段内合计{}个文件，空文件{}个，解析{}个有数据文件,，耗时{}秒".format(start_date,end_date,len(files)+len(null_files),len(null_files),len(files),costtime))
def parseDataFile(self,source_dir,target_dir,data_sep="|",data_start_row=1,data_end_flag="",start_row=1,start_col=1):
'''
获取数据文件中的有效数据
:param source_dir:数据文件绝对路径
:param target_dir:生成excel文件的绝对路径
:param data_sep:数据分隔符
:param data_start_row:有效数据的开始行
:param data_end_flag:有效数据结束标识
:param start_row:写入excel的开始行
:param start_col:写入excel的开始列
:param table_head:有效数据是否包含表头
'''
data = self.__getFileData(source_dir,data_sep,data_start_row,data_end_flag)
fname,basename = self.__writeToExcel(source_dir,target_dir,data,start_row,start_col)
return fname,basename
def __getFileData(self,filename,data_sep,data_start_row,data_end_flag):
data = []
try:
if os.stat(filename).st_size==0:
self.log.info("{}文件为空".format(os.path.basename(filename)))
with open(filename,"r",encoding="utf-8") as f:
line_index = 1
for line in f:
if line:
line = line.replace("\n","")
if line_index >= data_start_row and line != data_end_flag:
data.append(line.split(data_sep))
if line == data_end_flag:
break
else:
break
line_index = line_index+1
except Exception as e:
self.log.info("{}解析错误".format(filename))
return data
def __writeToExcel(self,source_file,target_file,data,start_row,start_col):
basename =None
try:
if os.stat(source_file).st_size==0 or data is None or len(data)==0:
self.log.info("{}文件为空，未生成对应的excel文件".format(source_file))
else:
with open(target_file,"w+",encoding="utf-8") as f:
pass
basename = os.path.basename(target_file)
# self.__write_col(filename,basename,col=start_col,data=data)
self.__write2ExcelOpenexl(target_file,basename,row=start_row,col=start_col,data=data)
self.log.info("数据写入excel完成，文件路径{}".format(target_file))
except Exception as e:
self.log.info("写入文件{}错误".format(e))
return target_file,basename
def __write2ExcelOpenexl(self,excelname,sheet,row,col,data=None):
wb = Workbook()
ws = wb.create_sheet(sheet, 0)
for y,i in enumerate(data):
for x,n in enumerate(i):
ws.cell( y + col, x + row, n )
wb.save(excelname)
wb.close()
class LoggerBox(tk.Text):
def write(self, message):
self.insert("end", message)
class APP(object):
def __init__(self,window):
window.title('通用对账文件解析软件')
#获取屏幕尺寸计算参数，使窗口显示再屏幕中央
screen_width = window.winfo_screenwidth()
screen_height = window.winfo_screenheight()
width = 750
height = 800
# window_size = '%dx%d+%d+%d' % (width, height, (screen_width-width)/2, (screen_height-height)/2)
window_size = f'{width}x{height}+{round((screen_width-width)/2)}+{round((screen_height-height)/2)}'
window.geometry(window_size)
z = tk.Label(window,anchor = 'center',bg='white',justify = 'center', width=500, height=500)
z.place(x = 0, y = 0)
a = tk.Label(window,anchor = 'center',text="对账文件解析软件",relief="raised", font=('TimesNewRoman', 30),justify = 'center', width=32, height=1)
a.place(x = 50, y = 30)
self.source_dir = tk.StringVar()
self.source_dir.set("数据文件路径")
b = tk.Label(window,anchor='nw',textvariable=self.source_dir,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
b.place(x = 50, y = 100)
self.data_str = tk.StringVar()
self.data_str.set("文件包含字符串")
c = tk.Label(window,anchor='nw',textvariable=self.data_str,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
c.place(x = 50, y = 150)
self.suffix = tk.StringVar()
self.suffix.set("数据文件后缀")
d = tk.Label(window,anchor='nw',textvariable=self.suffix,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
d.place(x = 400, y = 150)
self.target_dir = tk.StringVar()
self.target_dir.set("文件存放路径")
f = tk.Label(window,anchor='nw',textvariable=self.target_dir,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
f.place(x = 50, y = 200)
self.data_sep = tk.StringVar()
self.data_sep.set("数据分隔符")
g = tk.Label(window,anchor='nw',textvariable=self.data_sep,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
g.place(x = 50, y = 250)
self.start_date = tk.StringVar()
self.start_date.set("开始日期")
h = tk.Label(window,anchor='nw',textvariable=self.start_date,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
h.place(x = 50, y = 300)
self.end_date = tk.StringVar()
self.end_date.set("结束日期")
i = tk.Label(window,anchor='nw',textvariable=self.end_date,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
i.place(x = 400, y = 300)
self.date_format = tk.StringVar()
self.date_format.set("日期格式")
j = tk.Label(window,anchor='nw',textvariable=self.date_format,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
j.place(x = 50, y = 350)
self.data_start_row = tk.StringVar()
self.data_start_row.set("数据开始行")
k = tk.Label(window,anchor='nw',textvariable=self.data_start_row,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
k.place(x = 50, y = 400)
self.data_end_flag = tk.StringVar()
self.data_end_flag.set("数据结束标志")
l = tk.Label(window,anchor='nw',textvariable=self.data_end_flag,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
l.place(x = 50, y = 450)
self.start_row = tk.StringVar()
self.start_row.set("excel写入行")
m = tk.Label(window,anchor='nw',textvariable=self.start_row,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
m.place(x = 50, y = 500)
self.start_col = tk.StringVar()
self.start_col.set("excel写入列")
n = tk.Label(window,anchor='nw',textvariable=self.start_col,fg='black',bg='white',font=('TimesNewRoman',12),justify='center',width=50,height=1)
n.place(x = 400, y = 500)
self.source_dir = tk.StringVar()
entry_source_dir = tk.Entry(window, textvariable=self.source_dir).place(x=200, y=100)
tk.Button(window,text="选择源文件夹",command=lambda:self.select_dir(self.source_dir),width=15,height=1).place(x=400, y=100)
self.data_str = tk.StringVar()
entry_source_dir = tk.Entry(window, textvariable=self.data_str).place(x=200, y=150)
self.suffix = tk.StringVar()
self.suffix.set(".txt")
entry_source_dir = tk.Entry(window, textvariable=self.suffix).place(x=550, y=150)
self.target_dir = tk.StringVar()
entry_target_dir = tk.Entry(window, textvariable=self.target_dir).place(x=200, y=200)
tk.Button(window,text="选择存放文件夹",command=lambda:self.select_dir(self.target_dir),width=15,height=1).place(x=400, y=200)
self.data_sep = tk.StringVar()
self.data_sep.set("|")
entry_data_sep = tk.Entry(window, textvariable=self.data_sep).place(x=200, y=250)
self.date_format = tk.StringVar()
self.date_format.set("%Y%m%d")
entry_date_format = tk.Entry(window, textvariable=self.date_format).place(x=200, y=350)
now=datetime.datetime.now()
self.start_date = tk.StringVar()
self.start_date.set(now.strftime(self.date_format.get()))
entry_start_date = tk.Entry(window, textvariable=self.start_date).place(x=200, y=300)
self.end_date = tk.StringVar()
self.end_date.set(now.strftime(self.date_format.get()))
entry_end_date = tk.Entry(window, textvariable=self.end_date).place(x=550, y=300)
self.data_start_row = tk.IntVar()
self.data_start_row.set(int(1))
entry_data_start_row = tk.Entry(window, textvariable=self.data_start_row).place(x=200, y=400)
self.data_end_flag = tk.StringVar()
entry_data_end_flag = tk.Entry(window, textvariable=self.data_end_flag).place(x=200, y=450)
self.start_row = tk.IntVar()
self.start_row.set(int(1))
entry_start_row = tk.Entry(window, textvariable=self.start_row).place(x=200, y=500)
self.start_col = tk.IntVar()
self.start_col.set(int(1))
entry_start_col = tk.Entry(window, textvariable=self.start_col).place(x=550, y=500)
n = tk.Button(window, text='开始解析', width=90,height=1, command=self.parse_file)
n.place(x = 50,y = 550)
logBox = LoggerBox(window, width=90,height=10)
logBox.place(x=50,y=600)
self.log = logging.getLogger("log")
self.log.setLevel(logging.INFO)
handler = logging.StreamHandler(logBox)
self.log.addHandler(handler)
def select_dir(self,dir):
dir.set(filedialog.askdirectory())
def parse_file(self):
try:
parser = FileParser()
parser.batchParseDataFile(self.data_str.get(),self.log,self.start_date.get(),self.end_date.get(),self.date_format.get(),self.source_dir.get(),self.target_dir.get(),self.suffix.get(),self.data_sep.get(),self.data_start_row.get(),self.data_end_flag.get(),self.start_row.get(),self.start_col.get())
except Exception as e:
messagebox.showinfo('错误信息',e)
if __name__=="__main__":
window = tk.Tk()
app = APP(window)
window.mainloop()
本文由【产品经理不是经理】 gzh 同步发布，欢迎关注
网友评论