网站免费关键词如何做,交换友情链接的要求有,国建设银行e路通网站申,淮安做网站.哪家网络公司好?目录
实例要求
原始pdf文件格式
输出xls文件格式
运行界面
完整代码
代码分析
遍历表格
布局界面
控件简介
写入表格
表格排序
事件循环
异步编程 实例要求
使用PySimpleGUI做一个把单位考勤系统导出的pdf文件合并输出Excel的应用#xff0c;故事出自#xff1…
目录
实例要求
原始pdf文件格式
输出xls文件格式
运行界面
完整代码
代码分析
遍历表格
布局界面
控件简介
写入表格
表格排序
事件循环
异步编程 实例要求
使用PySimpleGUI做一个把单位考勤系统导出的pdf文件合并输出Excel的应用故事出自
https://hannyang.blog.csdn.net/article/details/135395946
当时时间紧没有好好做界面且输出csv文件了事。今天趁周六休息把代码做一下升级处理使用库PySimpleGUI做了一个稍微漂亮一点的界面又用pdfplumber直接遍历多个pdf文件得到数据后输出Excel文件比我原本先做合并pdf文件再去取数要快原先的pdf文件合并操作纯粹有点多余。最后又尝试对pdf文件读取函数的改造使用了asyncio异步编程效果非常不错。
下面请听我慢慢道来
原始pdf文件格式 输出xls文件格式 运行界面 完整代码
import xlwt, pyperclip, asyncio, pdfplumber
import os, time, datetime as dt
import PySimpleGUI as sg# 全局变量
table_head 姓名,部门,应到,实到,出勤率,迟到次数,早退次数,加班(分钟)
path, font , (宋体,12)
date, data [], []
DateFormat . . - . .
ErrMessage 错误
SortedType [出勤率排序,加班时长排序,迟到次数排序,早退次数排序]# 定义布局
layout [[sg.Text(昆山分行考勤表,font(,16)),sg.Text(pad(132,10)),sg.Text(请选择考勤文件:,fontfont),sg.Input(key-FOLDER-, enable_eventsTrue, readonlyTrue,fontfont,size18),sg.FolderBrowse(button_text..., enable_eventsTrue, initial_folder./)],[sg.Text(考勤日期,fontfont),sg.Text(DateFormat,key-DATE-,fontfont)],[sg.Table(values,headingstable_head.split(,),key-TABLE-,auto_size_columnsFalse,justificationleft,num_rows10)],[sg.Button(输出Excel文件,size(12,1),pad(15,30)),sg.Button(SortedType[0], enable_eventsTrue,size(10,1),pad(15,30)),sg.Button(SortedType[1], enable_eventsTrue,size(10,1),pad(15,30)),sg.Button(SortedType[2], enable_eventsTrue,size(10,1),pad(15,30)),sg.Button(SortedType[3], enable_eventsTrue,size(10,1),pad(15,30)),sg.Button(退出,size(10,1),pad(15,30))],[sg.StatusBar(,key-BAR-,fontfont,size92)]
]# 读取pdf表格
async def read_table(file):dct dict()with pdfplumber.open(file) as pdf:for page in pdf.pages:tables page.extract_tables(table_settings {})for table in tables:for lst in table:tmp lst[1:]if not any(tmp): continuetmp [tmp[0]]tmp[3:8][tmp[-1]]tmp[0] tmp[0].replace(\n,)tmp[0] tmp[0].split(/)tmp[0] tmp[0][-1]if lst[0]时间:dct[lst[0]] tmp[0]else:dct[,.join([lst[0],tmp[0]])] ,.join(tmp[1:])return dct# 写入xls文件
def write_sheet():global data, date, table_head, ErrMessageif ErrMessage[:2] in (错误,文件): returnmyxl xlwt.Workbook()style xlwt.easyxf(align: wrap yes; align: horiz center; font: bold yes;) sheet myxl.add_sheet(考勤表)wcol [20,40,60,30,30,40,40,40,60]for i,w in enumerate(wcol):sheet.col(i).width w * 80sheet.write_merge(0,0,0,8,出勤统计报表,style)style xlwt.easyxf(borders:top thin; borders:bottom thin; borders:left thin; borders:right thin;) sheet.write_merge(1,1,0,2,考勤日期date[0])for i,head in enumerate([序号]table_head.split(,)):sheet.write(2,i,head,style)for i,row in enumerate(data):for j,col in enumerate([str(i1)]row):sheet.write(3i,j,col,style)for i,t in enumerate(SortedType):if t in ErrMessage:tmp SortedType[i]breakelse: tmp excel_file f昆山分行考勤表{date[0]}({tmp}{strDateTime()}).xlsErrMessage f文件输出为{excel_file}try:myxl.save(excel_file)except:ErrMessage 写入excel文件失败finally:pyperclip.copy(\\.join((os.getcwd(),excel_file)))window[-BAR-].update(ErrMessage)# 获取当前时间
def strDateTime(diff0):now dt.datetime.now()time now dt.timedelta(daysdiff) return f{time.year}{time.month:02}{time.day:02}{time.hour:02}{time.minute:02}{time.second:02}# 选择并处理文件
async def on_text_changed(event, values):global date, data, path, ErrMessagenew_path values[-FOLDER-]window[-FOLDER-].update(new_path.split(/)[-1])if pathnew_path: returnelse: path new_pathpdfs [f for f in os.listdir(path) if f.endswith(.pdf) and not f.startswith(PDFmerged)]nums len(pdfs)if nums0:ErrMessage 错误所选文件夹中没有PDF文件window[-BAR-].update(ErrMessage)window[-DATE-].update(DateFormat)window[-TABLE-].update(values[])returndate, data, sheet [], [], dict()tasks []for pdf in pdfs:tasks.append(read_table(/.join([path,pdf])))ErrMessage f文件读取中(共{nums}个PDF文件)......window[-BAR-].update(ErrMessage)window.refresh()results await asyncio.gather(*tasks)for r in results:dt r.get(时间,None)if dt: date.append(dt)sheet.update(r)if date:window[-DATE-].update(date[-1])for k,v in sheet.items():if k in (时间,姓名,所属组织,普通班个人出勤统计报表,): continuedata.append(,.join([k,v]).split(,))window[-TABLE-].update(valuesdata)persons len(data)departments len(set([d[1] for d in data]))if 0:#len(set(date))!1:data []ErrMessage f错误请检查所选文件存在多个时间段{,.join(set(date))}else:ErrMessage f考勤人数{persons} / 部门数{departments}window[-BAR-].update(ErrMessage)# 表格排序
def on_table_sorted(event, data):global ErrMessageif not data: returnslist [x[-4][:-1], x[-1], x[-3], x[-2]]style slist[SortedType.index(event)]data sorted(data, keylambda x: float(eval(style)), reverseTrue)window[-TABLE-].update(valuesdata)ErrMessage f已按{event}更新window[-BAR-].update(ErrMessage)# 创建窗口
window sg.Window(考勤表汇总, layout, finalizeTrue)# 事件循环
while True:event, values window.read()if event sg.WINDOW_CLOSED or event 退出:breakelif event -FOLDER-:asyncio.run(on_text_changed(event, values))elif event in SortedType:on_table_sorted(event, data)elif event 输出Excel文件:write_sheet()# 关闭窗口
window.close()代码分析
重点代码都用彩色字体加粗标注了
遍历表格
读取代码如下
import pdfplumber
...... with pdfplumber.open(file) as pdf: for page in pdf.pages: tables page.extract_tables(table_settings {}) for table in tables: for lst in table: # 根据表格实际情况来清洗数据 return dct
布局界面
import PySimpleGUI as pg
layout [ [sg.Text(昆山分行考勤表,font(,16)), sg.Text(pad(132,10)), sg.Text(请选择考勤文件:,fontfont), sg.Input(key-FOLDER-, enable_eventsTrue, readonlyTrue,fontfont,size18), sg.FolderBrowse(button_text..., enable_eventsTrue, initial_folder./) ], [sg.Text(考勤日期,fontfont), sg.Text(DateFormat,key-DATE-,fontfont) ], [sg.Table(values, headingstable_head.split(,), key-TABLE-, auto_size_columnsFalse, justificationleft, num_rows10)], [sg.Button(输出Excel文件,size(12,1),pad(15,30)), sg.Button(SortedType[0], enable_eventsTrue,size(10,1),pad(15,30)), sg.Button(SortedType[1], enable_eventsTrue,size(10,1),pad(15,30)), sg.Button(SortedType[2], enable_eventsTrue,size(10,1),pad(15,30)), sg.Button(SortedType[3], enable_eventsTrue,size(10,1),pad(15,30)), sg.Button(退出,size(10,1),pad(15,30))], [sg.StatusBar(,key-BAR-,fontfont,size92)]]
控件简介
除了最常用的Text, Input, Button使用了 FolderBrowse、Table、StatsBar 三个不是最常用的控件分别是文件夹打开框、表格和状态栏。
表格最重要的三个参数 values, headings, auto_size_columns
sg.Table(values, headingstable_head.split(,), auto_size_columnsFalse)
表格数据values和表头headings都列表(分别是二维和一维的)auto_size_columnsFalse建议不要缺省否则列宽不可控各列都自动缩进紧靠在一起。
表格更新数据的方法window[-TABLE-].update(valuesdata)
写入表格
import xlwt
def write_sheet(): global data, date, table_head, ErrMessage if ErrMessage[:2] in (错误,输出): return myxl xlwt.Workbook() style xlwt.easyxf(align: wrap yes; align: horiz center; font: bold yes;) sheet myxl.add_sheet(考勤表) wcol [20,40,60,30,30,40,40,40,60] for i,w in enumerate(wcol): sheet.col(i).width w * 80 sheet.write_merge(0,0,0,8,出勤统计报表,style) style xlwt.easyxf(borders:top thin; borders:bottom thin; borders:left thin; borders:right thin;) sheet.write_merge(1,1,0,2,考勤日期date[0]) for i,head in enumerate([序号]table_head.split(,)): sheet.write(2,i,head,style) for i,row in enumerate(data): for j,col in enumerate([str(i1)]row): sheet.write(3i,j,col,style) for i,t in enumerate(SortedType): if t in ErrMessage: tmp SortedType[i] break else: tmp excel_file f昆山分行考勤表{date[0]}({tmp}{strDateTime()}).xls ErrMessage f输出文件为{excel_file} try: myxl.save(excel_file) except: ErrMessage 写入excel文件失败
注意单格和多个单元格的写入区别 sheet.write() sheet.write_merge()
表格排序
SortedType [出勤率排序,加班时长排序,迟到次数排序,早退次数排序] def on_table_sorted(event, data): global ErrMessage if not data: return slist [x[-4][:-1], x[-1], x[-3], x[-2]] style slist[SortedType.index(event)] data sorted(data, keylambda x: float(eval(style)), reverseTrue) window[-TABLE-].update(valuesdata) ErrMessage f已按{event}更新 window[-BAR-].update(ErrMessage)
虽然经常有人诟病eval()函数的安全性但这里还是用eval()简化表格排序事件否则要多写很多代码。
事件循环
while True: event, values window.read() if event sg.WINDOW_CLOSED or event 退出: break elif event -FOLDER-: asyncio.run(on_text_changed(event, values)) elif event in SortedType: on_table_sorted(event, data) elif event 输出Excel文件: write_sheet()
异步编程
此时请出本篇的主角“异步编程”什么是异步编程呢就是有点多任务操作的意思。
异步编程是一种编程范式它允许某些操作在等待结果时不阻塞整个程序。在传统的同步编程中程序会按照顺序执行一旦遇到需要等待的操作如文件I/O或网络请求整个程序就会被阻塞等待操作完成。而在异步编程中程序并不会因为某个耗时的IO操作而停下其他所有任务而是将这个任务交给系统处理自身继续执行后续的操作等到IO操作完成后系统会通知程序进行下一步的处理。
asyncio
在上一段代码中响应-FOLDER-时使用了asyncio.run()函数
import asyncio ....... ......
while True: event, values window.read() if event sg.WINDOW_CLOSED or event 退出: break elif event -FOLDER-: asyncio.run(on_text_changed(event, values))
asyncio.run运行的这个是异步编程的主函数需要用async def来定义
async def
async def on_text_changed(event, values): ......其它代码略...... tasks [] for pdf in pdfs: tasks.append(read_table(/.join([path,pdf]))) ErrMessage f文件读取中(共{nums}个PDF文件)...... window[-BAR-].update(ErrMessage) window.refresh() results await asyncio.gather(*tasks) for r in results: ......遍历取回的被调异步函数返回值的列表......
await
异步主函数中使用 await asyncio.gather(*tasks) 取回被函数的返回结果返回结果是多个任务的返回值组成的列表而主函数的任务呢就是被调函数组成的列表asks.append(read_table())
同样的被调函数也需要用async def来定义它一般都是文件I/O或网络请求等比较耗时的操作
async def read_table(file): dct dict() with pdfplumber.open(file) as pdf: # 读取pdf文件 I/O操作 return dct 源码和2个例表已绑定上传资源欢迎下载测试。
