Python异步编程|PySimpleGUI界面读取PDF转换Excel

简介: 实例要求:使用PySimpleGUI库做一个把单位考勤系统导出的pdf文件合并输出Excel表格的应用,库pdfplumber直接遍历多个pdf文件,得到数据后输出xks文件,同时对pdf文件读取函数的进行改造,使用了asyncio异步编程效果非常不错。

pp.png


实例要求

使用PySimpleGUI库做一个把单位考勤系统导出的pdf文件合并输出Excel表格的应用,库pdfplumber直接遍历多个pdf文件,得到数据后输出xks文件,同时对pdf文件读取函数的进行改造,使用了asyncio异步编程效果非常不错。

下面请听我慢慢道来:

原始pdf文件格式

image.gif


输出xls文件格式

image.gif


运行界面

image.gif


完整代码

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("请选择考勤文件:",font=font),
     sg.Input(key="-FOLDER-", enable_events=True, readonly=True,font=font,size=18),
     sg.FolderBrowse(button_text='...', enable_events=True, initial_folder='./')
     ],
    [sg.Text("考勤日期:",font=font),
     sg.Text(DateFormat,key='-DATE-',font=font)
     ],
    [sg.Table(values='',
              headings=table_head.split(','),
              key='-TABLE-',
              auto_size_columns=False,
              justification='left',
              num_rows=10)],
    [sg.Button("输出Excel文件",size=(12,1),pad=(15,30)),
     sg.Button(SortedType[0], enable_events=True,size=(10,1),pad=(15,30)),
     sg.Button(SortedType[1], enable_events=True,size=(10,1),pad=(15,30)),
     sg.Button(SortedType[2], enable_events=True,size=(10,1),pad=(15,30)),
     sg.Button(SortedType[3], enable_events=True,size=(10,1),pad=(15,30)),
     sg.Button("退出",size=(10,1),pad=(15,30))],
    [sg.StatusBar('',key="-BAR-",font=font,size=92)]
]
# 读取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): continue
                    tmp = [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, 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(i+1)]+row):
            sheet.write(3+i,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文件失败!'
    finally:
        pyperclip.copy('\\'.join((os.getcwd(),excel_file)))
        window['-BAR-'].update(ErrMessage)
# 获取当前时间
def strDateTime(diff=0):
    now = dt.datetime.now()
    time = now + dt.timedelta(days=diff)    
    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, ErrMessage
    new_path = values["-FOLDER-"]
    window["-FOLDER-"].update(new_path.split('/')[-1])
    if path==new_path: return
    else: path = new_path
    pdfs = [f for f in os.listdir(path) if f.endswith('.pdf') and not f.startswith('PDFmerged')]
    nums = len(pdfs)
    if nums==0:
        ErrMessage = '错误:所选文件夹中没有PDF文件!'
        window['-BAR-'].update(ErrMessage)
        window['-DATE-'].update(DateFormat)
        window['-TABLE-'].update(values=[])
        return
    date, 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 ('时间','姓名,所属组织','普通班个人出勤统计报表,'): continue
        data.append(','.join([k,v]).split(','))
        window['-TABLE-'].update(values=data)
    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 ErrMessage
    if not data: return
    slist = ['x[-4][:-1]', 'x[-1]', 'x[-3]', 'x[-2]']
    style = slist[SortedType.index(event)]
    data = sorted(data, key=lambda x: float(eval(style)), reverse=True)
    window['-TABLE-'].update(values=data)
    ErrMessage = f'已按{event}更新!'
    window['-BAR-'].update(ErrMessage)
# 创建窗口
window = sg.Window("考勤表汇总", layout, finalize=True)
# 事件循环
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()
# 关闭窗口
window.close()

image.gif

代码分析

重点代码都用彩色字体加粗标注了:

遍历表格

读取代码如下:

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("请选择考勤文件:",font=font),

    sg.Input(key="-FOLDER-", enable_events=True, readonly=True,font=font,size=18),

    sg.FolderBrowse(button_text='...', enable_events=True, initial_folder='./')

    ],

   [sg.Text("考勤日期:",font=font),

    sg.Text(DateFormat,key='-DATE-',font=font)

    ],

   [sg.Table(values='',

             headings=table_head.split(','),

             key='-TABLE-',

             auto_size_columns=False,

             justification='left',

             num_rows=10)],

   [sg.Button("输出Excel文件",size=(12,1),pad=(15,30)),

    sg.Button(SortedType[0], enable_events=True,size=(10,1),pad=(15,30)),

    sg.Button(SortedType[1], enable_events=True,size=(10,1),pad=(15,30)),

    sg.Button(SortedType[2], enable_events=True,size=(10,1),pad=(15,30)),

    sg.Button(SortedType[3], enable_events=True,size=(10,1),pad=(15,30)),

    sg.Button("退出",size=(10,1),pad=(15,30))],

   [sg.StatusBar('',key="-BAR-",font=font,size=92)]]

控件简介

除了最常用的Text, Input, Button,使用了 FolderBrowse、Table、StatsBar三个不是最常用的控件,分别是文件夹打开框、表格和状态栏。

表格最重要的三个参数: values, headings, auto_size_columns

sg.Table(values='', headings=table_head.split(','), auto_size_columns=False)

表格数据values和表头headings都列表(分别是二维和一维的),auto_size_columns=False建议不要缺省,否则列宽不可控,各列都自动缩进紧靠在一起。

表格更新数据的方法:window['-TABLE-'].update(values=data)

写入表格

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(i+1)]+row):

           sheet.write(3+i,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, key=lambda x: float(eval(style)), reverse=True)

   window['-TABLE-'].update(values=data)

   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运行的这个是异步编程的主函数,需要用asyncdef来定义:

async def

asyncdefon_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 = awaitasyncio.gather(*tasks)

   for r in results:

      ......遍历取回的被调异步函数返回值的列表......

await

异步主函数中使用 awaitasyncio.gather(*tasks) 取回被函数的返回结果,返回结果是多个任务的返回值组成的列表;而主函数的任务呢就,是被调函数组成的列表:asks.append(read_table())

同样的,被调函数也需要用asyncdef来定义,它一般都是文件I/O或网络请求等比较耗时的操作:

asyncdefread_table(file):

   dct = dict()

   with pdfplumber.open(file) as pdf:

       # 读取pdf文件 I/O操作

   return dct


目录
相关文章
|
2月前
|
数据采集 数据可视化 数据挖掘
利用Python自动化处理Excel数据:从基础到进阶####
本文旨在为读者提供一个全面的指南,通过Python编程语言实现Excel数据的自动化处理。无论你是初学者还是有经验的开发者,本文都将帮助你掌握Pandas和openpyxl这两个强大的库,从而提升数据处理的效率和准确性。我们将从环境设置开始,逐步深入到数据读取、清洗、分析和可视化等各个环节,最终实现一个实际的自动化项目案例。 ####
305 10
|
3月前
|
Python
Python中的异步编程:使用asyncio和aiohttp实现高效网络请求
【10月更文挑战第34天】在Python的世界里,异步编程是提高效率的利器。本文将带你了解如何使用asyncio和aiohttp库来编写高效的网络请求代码。我们将通过一个简单的示例来展示如何利用这些工具来并发地处理多个网络请求,从而提高程序的整体性能。准备好让你的Python代码飞起来吧!
152 2
|
7天前
|
文字识别 BI
【图片型PDF】批量识别扫描件PDF指定区域局部位置内容,将识别内容导出Excel表格或批量改名文件,基于阿里云OCR对图片型PDF识别改名案例实现
在医疗和政务等领域,图片型PDF文件(如病历、报告、公文扫描件)的处理需求广泛。通过OCR技术识别这些文件中的文字信息,提取关键内容并保存为表格,极大提高了信息管理和利用效率。本文介绍一款工具——咕嘎批量OCR系统,帮助用户快速处理图片型PDF文件,支持区域识别、内容提取、导出表格及批量改名等功能。下载工具后,按步骤选择处理模式、进行区域采样、批量处理文件,几分钟内即可高效完成数百个文件的处理。
49 8
|
1月前
|
数据可视化 数据挖掘 大数据
1.1 学习Python操作Excel的必要性
学习Python操作Excel在当今数据驱动的商业环境中至关重要。Python能处理大规模数据集,突破Excel行数限制;提供丰富的库实现复杂数据分析和自动化任务,显著提高效率。掌握这项技能不仅能提升个人能力,还能为企业带来价值,减少人为错误,提高决策效率。推荐从基础语法、Excel操作库开始学习,逐步进阶到数据可视化和自动化报表系统。通过实际项目巩固知识,关注新技术,为职业发展奠定坚实基础。
|
3月前
|
API 调度 开发者
探索Python中的异步编程:从asyncio到Trio
本文将带你深入Python异步编程的心脏地带,从asyncio的基本概念到Trio的高级特性,我们将一起揭开Python异步编程的神秘面纱,并探讨它们如何改变我们的编程方式。
|
3月前
|
API 开发者 Python
探索Python中的异步编程:Asyncio与Tornado的对决
在这个快节奏的世界里,Python开发者面临着一个挑战:如何让代码跑得更快?本文将带你走进Python异步编程的两大阵营——Asyncio和Tornado,探讨它们如何帮助我们提升性能,以及在实际应用中如何选择。我们将通过一场虚拟的“对决”,比较这两个框架的性能和易用性,让你在异步编程的战场上做出明智的选择。
|
3月前
|
测试技术 Python
Python中的异步编程与`asyncio`库
Python中的异步编程与`asyncio`库
|
3月前
|
并行计算 调度 开发者
探索Python中的异步编程:从基础到实战
在Python的世界里,异步编程是一种让程序运行更加高效、响应更快的技术。本文不仅会介绍异步编程的基本概念和原理,还将通过具体代码示例展示如何在Python中实现异步操作。无论你是初学者还是有经验的开发者,都能从中获益,了解如何运用这一技术优化你的项目。
|
3月前
|
数据处理 Python
探索Python中的异步编程:从基础到实战
在Python的世界中,“速度”不仅是赛车手的追求。本文将带你领略Python异步编程的魅力,从原理到实践,我们不单单是看代码,更通过实例感受它的威力。你将学会如何用更少的服务器资源做更多的事,就像是在厨房里同时烹饪多道菜而不让任何一道烧焦。准备好了吗?让我们开始这场技术烹饪之旅。
|
3月前
|
数据采集 调度 Python
探索Python中的异步编程:从基础到高级
【10月更文挑战第36天】在Python的世界中,异步编程是提升程序性能和响应速度的重要工具。本文将带你深入了解Python异步编程的核心概念,包括事件循环、协程与异步IO,并逐步展示如何在实际项目中应用这些概念来编写更高效、可扩展的代码。通过理论讲解与实践案例的结合,我们将一起构建一个异步Web爬虫,以直观感受异步编程的强大之处。

热门文章

最新文章

推荐镜像

更多