Python自动化办公:5分钟搞定PDF批量转Excel(附完整代码)
Python自动化办公:5分钟搞定PDF批量转Excel(附完整代码)
你是不是也遇到过这样的场景?财务部门发来几百份供应商的PDF格式对账单,需要你手动把里面的金额、日期、编号一个个敲进Excel表格;或者导师甩给你一个文件夹,里面是上百篇文献的PDF摘要,要求你整理成结构化的数据表。面对这种重复、枯燥且极易出错的任务,很多人第一反应是硬着头皮上,结果往往是眼睛看花、手指敲麻,效率低下不说,还难免出错。
其实,对于这类有明确模式的批量文档处理,Python可以成为你的“数字员工”。它不知疲倦,不会出错,一旦写好脚本,就能把原本需要数小时甚至数天的工作,压缩到几分钟内完成。今天,我们就来深入聊聊,如何用Python真正实现“5分钟搞定PDF批量转Excel”,不仅仅是跑通一个Demo,而是构建一个健壮、灵活、适合你真实办公场景的自动化工具。我们将从核心库的选择讲起,逐步深入到表格识别、数据清洗、异常处理,最后封装成一个拿来即用的脚本。无论你是编程新手,还是希望优化现有流程的开发者,都能在这里找到实用的解决方案。
1. 核心工具选型:为什么是pdfplumber和openpyxl?
在Python的生态里,处理PDF的库不少,但各有侧重。PyPDF2 和 PyMuPDF 功能强大,但在提取表格和精确文本定位上不够直观;tabula-py 专精于表格提取,但依赖Java环境,配置稍显麻烦。综合易用性、纯Python实现以及对表格结构的解析能力,pdfplumber 成为了我们的首选。
pdfplumber 的优势在于它能以相对直观的方式,提供页面上的字符、线、矩形等底层元素,让我们不仅能拿到文本,还能获取其坐标、字体等信息。这对于从非标准格式的PDF中“抠”出数据至关重要。而输出端,我们选择 openpyxl,因为它能很好地读写.xlsx格式,支持现代Excel的所有功能,如单元格样式、公式、图表(虽然我们这里用不到),且API清晰。
注意:如果你的PDF是扫描件(图片格式),上述所有基于文本提取的库都将失效。你需要先使用OCR(光学字符识别)技术,例如
pytesseract配合pdf2image库将PDF页面转为图片再识别文本,这完全是另一个技术路径,复杂度更高。
安装这两个库非常简单,只需一行命令:
pip install pdfplumber openpyxl
为了让你对它们的组合有一个全局认识,我们先看一个最简单的数据流转示意图:
PDF批量处理核心流程
- 定位输入:指定存放大量PDF文件的文件夹路径。
- 遍历文件:使用
os模块列出所有PDF文件,生成待处理列表。 - 单文件解析:对每个PDF,用
pdfplumber打开,根据其结构(是纯文本、表格还是混合布局)选择提取策略。 - 数据提取与转换:从解析出的原始数据中,清洗、筛选出我们需要的信息,并整理成列表或字典的结构。
- 写入Excel:将上一步整理好的结构化数据,通过
openpyxl按行或按列写入Excel工作表,一个PDF文件通常对应Excel中的一行。 - 循环与保存:循环处理所有PDF,最后统一保存Excel文件。
这个流程看似线性,但每个环节都有值得深挖的细节和可能遇到的“坑”,接下来我们就分步拆解。
2. 从简单到复杂:三种典型的PDF数据提取策略
不是所有的PDF都一样,因此没有一种提取方法能通吃所有场景。根据PDF内容的排版,我们大致可以分为三类,并采取不同的策略。
2.1 策略一:提取所有文本(适用于无结构纯文本)
这是最基础的需求。假设PDF里是连续的段落文字,没有明显的表格框线,你只是想获取全部文字内容。用 pdfplumber 可以轻松实现。
import pdfplumber
def extract_all_text(pdf_path):
"""提取PDF中的所有文本,返回一个字符串"""
all_text = ""
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
# extract_text() 方法会尝试保持文本的大致顺序
page_text = page.extract_text()
if page_text:
all_text += page_text + "\n" # 添加分页符
return all_text
但 extract_text() 有时顺序会乱。如果你需要更精确的控制,比如想按行提取,可以使用 extract_words(),它会返回每个单词及其坐标:
def extract_words_by_line(pdf_path):
"""按行提取单词,返回二维列表"""
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
words = page.extract_words(x_tolerance=2, y_tolerance=2)
# 根据y坐标(纵坐标)对单词进行分组,同一行的y坐标相近
lines = {}
for word in words:
line_key = round(word['top']) # 用top坐标取整作为行标识
lines.setdefault(line_key, []).append(word['text'])
# 按行排序并输出
for line_key in sorted(lines.keys()):
print(" ".join(lines[line_key]))
这里用到的 x_tolerance 和 y_tolerance 参数很重要,它们定义了在多远的距离内,单词可以被认为是同一个“文本对象”的一部分或属于同一行,适当调整可以改善提取效果。
2.2 策略二:提取表格数据(PDF中的标准表格)
这是最常见的自动化需求。发票、报表、数据清单通常都以表格形式呈现。pdfplumber 的 extract_tables() 方法对此有很好的支持。
def extract_tables_to_list(pdf_path):
"""提取PDF中的所有表格,返回一个三维列表 [page][table][row][cell]"""
all_tables = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
# 检测并提取当前页的所有表格
tables = page.extract_tables()
for table in tables:
all_tables.append(table)
# 你也可以通过table_settings参数调整检测灵敏度
# tables = page.extract_tables(table_settings={"vertical_strategy": "lines", "horizontal_strategy": "lines"})
return all_tables
extract_tables() 返回的 table 是一个二维列表,可以直接对应到Excel的行和列。这是将PDF表格迁移到Excel最直接的桥梁。但PDF表格千奇百怪,合并单元格、虚线边框、无边框表格都会影响识别精度。这时就需要调整 table_settings。
一个更健壮的表格提取函数可能长这样:
def extract_tables_robust(pdf_path, page_num=None):
"""增强版的表格提取,尝试多种策略"""
tables_found = []
strategies = [
{"vertical_strategy": "lines", "horizontal_strategy": "lines"},
{"vertical_strategy": "text", "horizontal_strategy": "text"},
{"vertical_strategy": "explicit", "horizontal_strategy": "explicit"},
]
with pdfplumber.open(pdf_path) as pdf:
pages = [pdf.pages[page_num]] if page_num is not None else pdf.pages
for page in pages:
for strategy in strategies:
try:
tables = page.extract_tables(table_settings=strategy)
if tables and any(any(row for row in table) for table in tables):
tables_found.extend(tables)
break # 找到一种可用的策略就跳出
except Exception as e:
continue
return tables_found
2.3 策略三:基于坐标的关键信息抓取(混合布局)
最复杂的情况是:你需要的信息散落在PDF的各个角落,比如从一份合同PDF里提取“合同编号”、“签署日期”、“总金额”。这些信息没有固定的表格承载,但通常有固定的标签(如“合同编号:”)或相对固定的页面位置。
这时,我们需要结合文本提取和坐标判断。思路是:先提取所有文本块及其边界框(bbox),然后通过关键词或坐标范围来定位目标。
def extract_info_by_keyword(pdf_path, keyword_mappings):
"""
根据关键词映射提取信息。
keyword_mappings: 字典,如 {"合同编号": None, "总金额": None}
函数会查找关键词后的文本作为值。
"""
result = {key: None for key in keyword_mappings}
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
# 提取单词和其bbox
words = page.extract_words(extra_attrs=["fontname", "size"])
text = " ".join([w['text'] for w in words])
for keyword in keyword_mappings.keys():
if keyword in text:
# 简单的逻辑:找到关键词,取它后面的一个词(实际应用需更复杂的解析)
idx = text.find(keyword)
# 这里只是一个示例,真实场景需要更精细的文本切片和正则表达式
# 例如,使用正则表达式 r'合同编号:\s*(\S+)'
pass
return result
对于位置固定的情况,可以直接使用坐标裁剪页面:
def extract_by_bbox(pdf_path, bbox_dict):
"""
通过预设的边界框提取信息。
bbox_dict: 字典,键为信息名,值为 (x0, top, x1, bottom) 元组
"""
result = {}
with pdfplumber.open(pdf_path) as pdf:
page = pdf.pages[0] # 假设信息在第一页
for name, bbox in bbox_dict.items():
cropped = page.within_bbox(bbox)
result[name] = cropped.extract_text()
return result
3. 构建健壮的批量处理框架
掌握了单文件提取技术后,我们要将其扩展为批量处理系统。这个框架需要处理文件遍历、错误处理、进度反馈、结果汇总等工程问题。
3.1 文件遍历与任务调度
首先,一个可靠的遍历函数是基础。它应该能过滤出PDF文件,并处理可能的嵌套文件夹。
import os
from pathlib import Path
def find_pdf_files(directory_path, recursive=True):
"""
查找目录下的所有PDF文件。
Args:
directory_path: 根目录路径
recursive: 是否递归查找子目录
Returns:
list: 完整的PDF文件路径列表
"""
pdf_files = []
if recursive:
# 使用 pathlib 更现代的方式
for pdf_path in Path(directory_path).rglob("*.pdf"):
pdf_files.append(str(pdf_path.resolve()))
else:
for item in os.listdir(directory_path):
full_path = os.path.join(directory_path, item)
if os.path.isfile(full_path) and item.lower().endswith('.pdf'):
pdf_files.append(full_path)
return sorted(pdf_files) # 排序便于追踪
3.2 核心处理循环与错误处理
批量处理中最怕一个文件出错导致整个任务崩溃。我们必须将每个文件的处理放在独立的 try-except 块中。
import traceback
from openpyxl import Workbook
from datetime import datetime
def batch_process_pdfs_to_excel(pdf_dir, output_excel_path, extraction_function):
"""
批量处理PDF的核心函数。
Args:
pdf_dir: PDF文件夹路径
output_excel_path: 输出的Excel文件路径
extraction_function: 自定义的数据提取函数,接收pdf_path,返回一个字典。
"""
pdf_paths = find_pdf_files(pdf_dir)
if not pdf_paths:
print(f"在目录 {pdf_dir} 中未找到PDF文件。")
return
wb = Workbook()
ws = wb.active
ws.title = "Extracted_Data"
# 写入表头。假设 extraction_function 返回的字典键作为表头。
# 我们先处理第一个文件来获取表头(如果所有文件结构一致)
headers_written = False
headers = []
success_count = 0
error_log = []
for idx, pdf_path in enumerate(pdf_paths, start=1):
print(f"正在处理 ({idx}/{len(pdf_paths)}): {os.path.basename(pdf_path)}")
try:
# 调用用户提供的提取函数
data_dict = extraction_function(pdf_path)
if not headers_written:
headers = list(data_dict.keys())
ws.append(headers) # 写入表头
headers_written = True
# 确保数据顺序与表头一致
row_data = [data_dict.get(h, "") for h in headers]
ws.append(row_data)
success_count += 1
except pdfplumber.exceptions.PDFSyntaxError:
error_msg = f"文件损坏或不是有效PDF: {pdf_path}"
error_log.append(error_msg)
print(f" 错误: {error_msg}")
except KeyError as e:
error_msg = f"提取字段错误 {e} 于文件: {pdf_path}"
error_log.append(error_msg)
print(f" 错误: {error_msg}")
except Exception as e:
error_msg = f"处理文件 {pdf_path} 时发生未知错误: {str(e)}"
error_log.append(error_msg)
print(f" 错误: {error_msg}")
# 打印详细堆栈信息用于调试
traceback.print_exc()
# 保存工作簿
wb.save(output_excel_path)
print(f"\n处理完成!成功: {success_count}, 失败: {len(error_log)}")
# 如果有错误,将日志写入文本文件
if error_log:
log_path = output_excel_path.replace('.xlsx', '_error_log.txt')
with open(log_path, 'w', encoding='utf-8') as f:
f.write(f"批量处理错误日志 {datetime.now()}\n")
f.write("="*50 + "\n")
for log in error_log:
f.write(log + "\n")
print(f"错误日志已保存至: {log_path}")
这个框架将处理逻辑(extraction_function)与批量引擎解耦,复用性非常高。你需要做的,就是根据你的PDF格式,编写那个特定的 extraction_function。
4. 实战:编写一个完整的发票信息提取脚本
让我们把所有知识串联起来,解决一个实际问题:从一个文件夹的发票PDF中,提取“发票号码”、“开票日期”、“购买方”、“价税合计”四个字段,并保存到Excel。
假设我们的发票PDF中,这些信息的位置相对固定,或者有明确的标签文字。我们将采用关键词定位为主,坐标定位为辅的混合策略。
首先,定义我们的提取函数:
import re
import pdfplumber
def extract_invoice_info(pdf_path):
"""
从发票PDF中提取特定信息。
这是一个示例函数,实际规则需要根据你的PDF样本调整。
"""
result = {
"发票号码": "",
"开票日期": "",
"购买方名称": "",
"价税合计(大写)": "",
"文件名": os.path.basename(pdf_path)
}
full_text = ""
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
full_text += page.extract_text() + "\n"
# 使用正则表达式匹配信息
# 示例规则,极其需要根据实际发票样式调整!!!
# 匹配发票号码,通常是“发票号码”后的一串数字
invoice_no_match = re.search(r'发票号码[::]?\s*([0-9]+)', full_text)
if invoice_no_match:
result["发票号码"] = invoice_no_match.group(1)
# 匹配开票日期,格式可能是YYYY年MM月DD日或YYYY-MM-DD
date_match = re.search(r'开票日期[::]?\s*(\d{4}[-年]\d{1,2}[-月]\d{1,2}日?)', full_text)
if date_match:
result["开票日期"] = date_match.group(1)
# 匹配购买方,通常在“购买方”或“购货单位”之后的多行文本内
# 这是一个简化版,复杂情况需要更精细的上下文分析
buyer_match = re.search(r'购买方[::]?\s*[\n\r]*([^\n\r]+(?:公司|厂|店))', full_text)
if buyer_match:
result["购买方名称"] = buyer_match.group(1).strip()
# 匹配价税合计大写金额
amount_match = re.search(r'价税合计.*?[((]大写[))].*?([零壹贰叁肆伍陆柒捌玖拾佰仟万亿元角分整]+)', full_text)
if amount_match:
result["价税合计(大写)"] = amount_match.group(1)
return result
然后,准备一个配置文件(比如 config.py),定义你的提取规则,这样不用修改主代码:
# config.py
EXTRACTION_RULES = {
"发票号码": {
"type": "regex",
"pattern": r'发票号码[::]?\s*([0-9]+)',
"group": 1
},
"开票日期": {
"type": "regex",
"pattern": r'开票日期[::]?\s*(\d{4}[-年]\d{1,2}[-月]\d{1,2}日?)',
"group": 1
},
# ... 更多规则
}
最后,主程序文件 main.py 将一切组合起来:
# main.py
import os
from pathlib import Path
from batch_processor import batch_process_pdfs_to_excel # 假设批量处理函数保存在这里
from invoice_extractor import extract_invoice_info # 假设提取函数保存在这里
if __name__ == "__main__":
# 用户配置区
PDF_SOURCE_DIR = "./invoices" # 你的发票PDF文件夹
OUTPUT_EXCEL_PATH = "./发票汇总.xlsx" # 输出Excel文件
print("开始批量提取发票信息...")
batch_process_pdfs_to_excel(
pdf_dir=PDF_SOURCE_DIR,
output_excel_path=OUTPUT_EXCEL_PATH,
extraction_function=extract_invoice_info # 传入我们的定制提取函数
)
print(f"数据已保存至: {os.path.abspath(OUTPUT_EXCEL_PATH)}")
运行这个 main.py,脚本就会自动扫描 ./invoices 文件夹下的所有PDF,应用 extract_invoice_info 函数提取信息,并将结果和任何错误日志保存下来。第一次运行时,你很可能需要根据少数样本的输出结果,反复调整 extract_invoice_info 函数中的正则表达式或逻辑,直到它能准确捕捉到你的数据。这个过程被称为“规则调优”,是此类自动化任务成功的关键。
5. 性能优化与高级技巧
当处理成千上万个PDF,或者每个PDF页数很多时,效率就变得重要了。这里有几个提升性能的点。
并行处理:如果处理每个PDF是独立的IO和计算密集型任务,可以使用多进程来加速。Python的 concurrent.futures 模块很方便。
from concurrent.futures import ProcessPoolExecutor, as_completed
def batch_process_parallel(pdf_paths, extraction_function, max_workers=4):
"""使用多进程并行处理PDF"""
results = {}
with ProcessPoolExecutor(max_workers=max_workers) as executor:
future_to_pdf = {executor.submit(extraction_function, path): path for path in pdf_paths}
for future in as_completed(future_to_pdf):
pdf_path = future_to_pdf[future]
try:
results[pdf_path] = future.result()
except Exception as exc:
results[pdf_path] = f"生成异常: {exc}"
return results
注意:并行处理时,写入Excel需要小心,因为多个进程不能同时写入同一个文件。通常的做法是让每个进程将结果先保存到临时文件或队列中,最后由主进程统一汇总写入。
缓存与增量处理:对于定期运行的任务,可以记录已处理过的文件哈希值(如MD5),下次运行时跳过未修改的文件,实现增量更新。
处理加密PDF:如果PDF有密码,pdfplumber.open() 需要提供 password 参数。你可以准备一个密码列表或通过其他方式获取密码。
内存管理:对于超大PDF,使用 pdfplumber.open(pdf_path, laparams={'line_margin': 0.5}) 中的 laparams 参数可以调整解析策略,有时能减少内存占用。处理完每一页后,及时关闭文件句柄(使用 with 语句保证)。
最后,别忘了给你的脚本添加一些人性化的功能,比如进度条。tqdm 库可以轻松实现:
pip install tqdm
from tqdm import tqdm
# 在批量处理循环中
for pdf_path in tqdm(pdf_paths, desc="处理PDF"):
# ... 处理逻辑
自动化脚本的价值在于一次投入,长期受益。在项目初期,花时间构建一个健壮、可配置、有良好错误处理的框架,远比写一个只能在一台电脑上跑一次的“脆皮”脚本要划算得多。当你下次再遇到“把XXX个PDF里的YYY信息弄到Excel里”这种需求时,你只需要调整或编写那个核心的 extraction_function,然后套用现成的批量处理框架,五分钟?也许真的就够了。
更多推荐



所有评论(0)