# Python自动化办公:用pypdf库5分钟搞定PDF批量合并与拆分(附完整代码)
如果你每天的工作都离不开PDF文件,那你一定经历过这样的场景:财务季度结束,需要把几十份独立的报销单合并成一份报告;法务部门发来上百页的合同,你只需要提取其中的关键章节;或者每个月都要把销售团队的业绩报表拆分成单独的文件发给每个人。这些重复性的操作不仅耗时,还容易出错,尤其是在处理大量文件时,手动操作几乎是一场噩梦。
我刚开始工作时,也曾经用鼠标一个个点击“合并”、“另存为”,直到有一次因为漏掉了一个文件,差点耽误了重要的项目交付。从那以后,我就开始寻找自动化解决方案,而Python的pypdf库彻底改变了我的工作方式。它不仅仅是一个工具,更像是一个得力的数字助手,能把原本需要几小时的工作压缩到几分钟内完成,而且准确率100%。
今天,我们就来深入探讨如何用pypdf库实现PDF的批量合并与拆分,我会分享一些在实际工作中总结出来的高效代码模板和避坑经验,让你也能快速上手,把时间花在更有价值的事情上。
## 1. 环境准备与pypdf库基础
在开始编写自动化脚本之前,我们需要先搭建好Python环境。我推荐使用Python 3.8或更高版本,因为这些版本对现代库的支持更好,而且pypdf库在这些版本上经过了充分测试。
### 1.1 安装pypdf库
安装pypdf非常简单,只需要一条命令。但这里有个细节需要注意:pypdf是PyPDF2的现代继任者,从2023年开始,PyPDF2的开发已经合并到pypdf项目中。这意味着如果你之前用过PyPDF2,现在应该切换到pypdf,因为它有更好的维护和更多新功能。
打开你的终端或命令提示符,执行以下命令:
```bash
pip install pypdf
```
如果你需要处理加密的PDF文件(比如有密码保护的文档),可以安装包含加密功能的完整版本:
```bash
pip install "pypdf[crypto]"
```
安装完成后,可以通过以下代码验证安装是否成功:
```python
import pypdf
print(f"pypdf版本: {pypdf.__version__}")
```
> 注意:有些教程可能还在使用`PyPDF2`这个旧的包名,如果你看到`import PyPDF2`的代码,需要将其改为`import pypdf`。虽然pypdf保持了向后兼容性,但使用新的导入方式能确保你用到最新的功能和修复。
### 1.2 理解PDF处理的基本概念
在开始编码前,了解几个核心概念会让后面的学习更顺畅:
- **PdfReader**:这是读取PDF文件的主要类。你可以把它想象成一个PDF文件的“阅读器”,通过它可以访问文件中的所有页面、文本和元数据。
- **PdfWriter**:用于创建或修改PDF文件的类。当你需要合并多个PDF、添加新页面或保存修改时,就会用到它。
- **PageObject**:代表PDF中的单个页面。每个页面都可以被读取、修改或复制到其他PDF中。
这三个类构成了pypdf库的核心,几乎所有的PDF操作都围绕它们展开。下面这个简单的例子展示了如何读取一个PDF文件的基本信息:
```python
from pypdf import PdfReader
# 读取PDF文件
reader = PdfReader("example.pdf")
# 获取页面数量
page_count = len(reader.pages)
print(f"PDF共有 {page_count} 页")
# 读取第一页的文本内容
first_page = reader.pages[0]
text_content = first_page.extract_text()
print(f"第一页内容预览: {text_content[:200]}...") # 只显示前200个字符
# 获取文档元数据
metadata = reader.metadata
print(f"文档标题: {metadata.title}")
print(f"作者: {metadata.author}")
print(f"创建日期: {metadata.creation_date}")
```
这个基础框架是后续所有复杂操作的基础。在实际工作中,我建议你先用一个小型的测试PDF文件运行这段代码,确保环境配置正确,然后再进行更复杂的操作。
## 2. PDF批量合并:从基础到高级技巧
合并PDF可能是办公自动化中最常见的需求之一。无论是合并多个报告、整合不同部门的文档,还是将扫描的图片PDF组合成完整文件,pypdf都能轻松应对。
### 2.1 基础合并:按顺序合并多个文件
最基本的合并场景是按特定顺序将多个PDF文件合并成一个。假设你有三个季度报告:`Q1_report.pdf`、`Q2_report.pdf`和`Q3_report.pdf`,需要将它们合并成年度报告。
下面是一个完整的解决方案:
```python
from pypdf import PdfWriter
import os
def merge_pdfs_simple(pdf_files, output_path):
"""
简单合并多个PDF文件
参数:
pdf_files: PDF文件路径列表,按合并顺序排列
output_path: 合并后文件的保存路径
"""
merger = PdfWriter()
for pdf_file in pdf_files:
try:
merger.append(pdf_file)
print(f"已添加: {pdf_file}")
except Exception as e:
print(f"添加文件 {pdf_file} 时出错: {e}")
# 可以选择跳过出错的文件继续处理
continue
# 写入合并后的文件
with open(output_path, "wb") as output_file:
merger.write(output_file)
print(f"合并完成!文件已保存至: {output_path}")
merger.close()
# 使用示例
if __name__ == "__main__":
# 定义要合并的文件列表
files_to_merge = [
"reports/Q1_report.pdf",
"reports/Q2_report.pdf",
"reports/Q3_report.pdf"
]
# 检查文件是否存在
existing_files = [f for f in files_to_merge if os.path.exists(f)]
if len(existing_files) != len(files_to_merge):
missing = set(files_to_merge) - set(existing_files)
print(f"警告:以下文件不存在: {missing}")
# 执行合并
merge_pdfs_simple(existing_files, "annual_report_2024.pdf")
```
这段代码的核心是`PdfWriter`的`append()`方法,它会按顺序将整个PDF文件添加到合并器中。`try-except`块确保了即使某个文件有问题,程序也不会完全崩溃,而是跳过该文件继续处理。
### 2.2 高级合并:选择性合并页面
有时候,我们不需要合并整个文件,只需要每个文件中的特定页面。比如,从多个合同中只提取签名页,或者从长文档中只合并目录和摘要部分。
pypdf提供了`merge()`方法来实现更精细的控制:
```python
from pypdf import PdfReader, PdfWriter
def merge_specific_pages(pdf_specs, output_path):
"""
从多个PDF中选择性合并特定页面
参数:
pdf_specs: 列表,每个元素是元组 (文件路径, 页面范围)
页面范围可以是:
- 单个数字: 3 (第3页,0-based索引)
- 范围: (1, 5) # 第1到第5页(包含)
- 列表: [0, 2, 4] # 第0、2、4页
output_path: 输出文件路径
"""
writer = PdfWriter()
for pdf_path, page_spec in pdf_specs:
reader = PdfReader(pdf_path)
# 根据页面规格确定要添加的页面
if isinstance(page_spec, int):
# 单个页面
pages_to_add = [reader.pages[page_spec]]
elif isinstance(page_spec, tuple) and len(page_spec) == 2:
# 页面范围
start, end = page_spec
pages_to_add = reader.pages[start:end+1]
elif isinstance(page_spec, list):
# 页面列表
pages_to_add = [reader.pages[i] for i in page_spec]
else:
raise ValueError(f"无效的页面规格: {page_spec}")
# 添加页面到写入器
for page in pages_to_add:
writer.add_page(page)
print(f"从 {pdf_path} 添加了 {len(pages_to_add)} 页")
# 保存合并后的文件
with open(output_path, "wb") as output_file:
writer.write(output_file)
print(f"选择性合并完成!输出文件: {output_path}")
writer.close()
# 实际应用示例:合并合同的关键页
if __name__ == "__main__":
# 假设我们有三份合同,每份都需要提取封面、条款页和签名页
contract_specs = [
("contract_1.pdf", [0, 2, 15]), # 第0、2、15页
("contract_2.pdf", [0, 3, 18]), # 第0、3、18页
("contract_3.pdf", [0, 2, 16]), # 第0、2、16页
]
merge_specific_pages(contract_specs, "combined_contract_key_pages.pdf")
```
这种选择性合并在实际工作中特别有用,因为它允许你精确控制最终文档的内容,避免无关信息的干扰。
### 2.3 批量合并文件夹中的所有PDF
当需要处理大量文件时,手动列出每个文件路径是不现实的。这时,我们可以编写一个自动扫描文件夹并合并所有PDF的脚本:
```python
import os
import glob
from pypdf import PdfWriter
from datetime import datetime
def merge_all_pdfs_in_folder(folder_path, output_path=None, sort_by="name"):
"""
合并指定文件夹中的所有PDF文件
参数:
folder_path: 包含PDF文件的文件夹路径
output_path: 输出文件路径(如果为None,则自动生成)
sort_by: 排序方式,可选 "name"(按文件名)或 "date"(按修改时间)
"""
# 查找所有PDF文件
pdf_pattern = os.path.join(folder_path, "*.pdf")
pdf_files = glob.glob(pdf_pattern)
if not pdf_files:
print(f"在 {folder_path} 中未找到PDF文件")
return
# 根据指定方式排序
if sort_by == "name":
pdf_files.sort()
elif sort_by == "date":
pdf_files.sort(key=os.path.getmtime)
else:
print(f"未知的排序方式: {sort_by},将按文件名排序")
pdf_files.sort()
print(f"找到 {len(pdf_files)} 个PDF文件:")
for i, f in enumerate(pdf_files, 1):
print(f" {i}. {os.path.basename(f)}")
# 如果没有指定输出路径,自动生成
if output_path is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
folder_name = os.path.basename(os.path.normpath(folder_path))
output_path = f"merged_{folder_name}_{timestamp}.pdf"
# 执行合并
merger = PdfWriter()
for pdf_file in pdf_files:
try:
merger.append(pdf_file)
except Exception as e:
print(f"跳过文件 {pdf_file}(错误: {e})")
continue
with open(output_path, "wb") as output_file:
merger.write(output_file)
merger.close()
print(f"\n成功合并 {len(pdf_files)} 个文件到: {output_path}")
print(f"文件大小: {os.path.getsize(output_path) / 1024:.2f} KB")
# 使用示例:合并下载文件夹中的所有PDF
if __name__ == "__main__":
# Windows系统
# downloads_folder = os.path.join(os.path.expanduser("~"), "Downloads")
# macOS/Linux系统
downloads_folder = os.path.join(os.path.expanduser("~"), "Downloads")
if os.path.exists(downloads_folder):
merge_all_pdfs_in_folder(downloads_folder, sort_by="date")
else:
print("下载文件夹不存在")
```
这个脚本的亮点在于它的灵活性:可以按文件名或修改时间排序,自动处理错误文件,还能根据文件夹名称和时间戳自动生成输出文件名。我在实际工作中经常用它来整理从不同渠道收到的零散文档。
## 3. PDF智能拆分:多种策略应对不同场景
与合并相反,拆分PDF是将一个大文件分解成多个小文件。这在分享文档、提取特定内容或准备邮件附件时非常有用。
### 3.1 按固定页数拆分
最简单的拆分方式是按固定页数将文档分成若干部分。比如,将一个100页的手册每20页拆分成一个文件,方便分章节阅读或发送。
```python
from pypdf import PdfReader, PdfWriter
import os
def split_by_fixed_pages(input_pdf, pages_per_file, output_folder="split_output"):
"""
按固定页数拆分PDF
参数:
input_pdf: 输入PDF文件路径
pages_per_file: 每个输出文件包含的页数
output_folder: 输出文件保存的文件夹
"""
# 创建输出文件夹
os.makedirs(output_folder, exist_ok=True)
# 读取输入文件
reader = PdfReader(input_pdf)
total_pages = len(reader.pages)
print(f"输入文件: {input_pdf}")
print(f"总页数: {total_pages}")
print(f"每 {pages_per_file} 页拆分为一个文件")
# 计算需要拆分成多少个文件
num_files = (total_pages + pages_per_file - 1) // pages_per_file # 向上取整
# 逐个创建输出文件
for file_index in range(num_files):
writer = PdfWriter()
# 计算当前文件的页面范围
start_page = file_index * pages_per_file
end_page = min(start_page + pages_per_file, total_pages)
# 添加页面到当前文件
for page_num in range(start_page, end_page):
writer.add_page(reader.pages[page_num])
# 生成输出文件名
base_name = os.path.splitext(os.path.basename(input_pdf))[0]
output_filename = f"{base_name}_part_{file_index + 1:03d}.pdf"
output_path = os.path.join(output_folder, output_filename)
# 保存文件
with open(output_path, "wb") as output_file:
writer.write(output_file)
print(f"创建文件: {output_filename} (页 {start_page + 1}-{end_page})")
writer.close()
print(f"\n拆分完成!共创建 {num_files} 个文件,保存在: {output_folder}")
# 使用示例
if __name__ == "__main__":
# 将一个大型手册每25页拆分成一个文件
split_by_fixed_pages(
input_pdf="product_manual.pdf",
pages_per_file=25,
output_folder="manual_parts"
)
```
这种拆分方式特别适合处理大型文档,比如技术手册、培训材料或长篇报告。通过控制`pages_per_file`参数,你可以根据实际需求调整每个文件的大小。
### 3.2 按书签/大纲拆分
很多PDF文件包含书签(也称为大纲或目录),这些书签通常对应文档的章节结构。按书签拆分可以智能地将文档按章节分解,这是最符合阅读习惯的拆分方式。
```python
from pypdf import PdfReader, PdfWriter
def split_by_bookmarks(input_pdf, output_folder="bookmark_split"):
"""
按书签拆分PDF文件
参数:
input_pdf: 输入PDF文件路径
output_folder: 输出文件夹路径
"""
import os
os.makedirs(output_folder, exist_ok=True)
reader = PdfReader(input_pdf)
# 获取文档的书签(大纲)
if not reader.outline:
print("该PDF文件没有书签,无法按书签拆分")
return
print(f"找到 {len(reader.outline)} 个书签")
# 书签结构可能是嵌套的,这里我们展平处理
def flatten_outline(outline_items, level=0):
flat_items = []
for item in outline_items:
if isinstance(item, list):
# 这是一个嵌套的书签组
flat_items.extend(flatten_outline(item, level + 1))
else:
# 单个书签
flat_items.append((item, level))
return flat_items
flat_outline = flatten_outline(reader.outline)
# 为每个书签创建独立的PDF文件
for i, (bookmark, level) in enumerate(flat_outline):
# 获取书签指向的页面
try:
# 书签对象通常有title和page属性
title = getattr(bookmark, 'title', f'书签_{i+1}')
page_obj = bookmark.page
# 找到页面索引
if hasattr(page_obj, 'page_number'):
page_index = page_obj.page_number
else:
# 尝试其他方式获取页面索引
page_index = 0 # 默认值
# 在实际代码中,这里需要更复杂的逻辑来获取准确的页面索引
# 为了示例简化,我们假设书签指向第i页
page_index = i
# 创建只包含该页面的PDF
writer = PdfWriter()
writer.add_page(reader.pages[page_index])
# 清理文件名(移除非法字符)
safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()
if not safe_title:
safe_title = f"section_{i+1}"
output_path = os.path.join(output_folder, f"{safe_title}.pdf")
with open(output_path, "wb") as output_file:
writer.write(output_file)
print(f"创建: {safe_title}.pdf (原书签: {title})")
writer.close()
except Exception as e:
print(f"处理书签时出错: {e}")
continue
print(f"\n按书签拆分完成!文件保存在: {output_folder}")
# 使用示例
if __name__ == "__main__":
split_by_bookmarks("technical_documentation.pdf", "chapters")
```
> 注意:PDF的书签结构可能比较复杂,不同的PDF生成工具创建的书签格式可能不同。上面的代码提供了一个基本框架,在实际使用中可能需要根据具体的PDF文件调整书签解析逻辑。
### 3.3 按文本内容智能拆分
最智能的拆分方式是根据文档内容自动识别拆分点。比如,你可以设置规则:每当出现"第X章"或"Section X"这样的文本时,就在此处拆分。
```python
from pypdf import PdfReader, PdfWriter
import re
def split_by_content_pattern(input_pdf, pattern, output_folder="content_split"):
"""
根据文本内容模式拆分PDF
参数:
input_pdf: 输入PDF文件路径
pattern: 正则表达式模式,匹配时开始新文件
output_folder: 输出文件夹
"""
import os
os.makedirs(output_folder, exist_ok=True)
reader = PdfReader(input_pdf)
total_pages = len(reader.pages)
# 编译正则表达式
regex = re.compile(pattern, re.IGNORECASE)
# 存储拆分点(页面索引)
split_points = [0] # 总是从第0页开始
print("扫描文档内容,寻找拆分点...")
for page_idx in range(total_pages):
page = reader.pages[page_idx]
text = page.extract_text()
if text and regex.search(text):
print(f"在第 {page_idx + 1} 页找到匹配模式")
split_points.append(page_idx)
# 添加最后一个拆分点(文档结束)
split_points.append(total_pages)
print(f"找到 {len(split_points) - 2} 个拆分点")
# 根据拆分点创建文件
for i in range(len(split_points) - 1):
start_page = split_points[i]
end_page = split_points[i + 1]
# 如果起始页和结束页相同,跳过(空部分)
if start_page >= end_page:
continue
writer = PdfWriter()
# 添加该部分的所有页面
for page_idx in range(start_page, end_page):
writer.add_page(reader.pages[page_idx])
# 生成文件名
output_filename = f"section_{i+1:03d}_pages_{start_page+1}-{end_page}.pdf"
output_path = os.path.join(output_folder, output_filename)
with open(output_path, "wb") as output_file:
writer.write(output_file)
print(f"创建: {output_filename} ({end_page - start_page} 页)")
writer.close()
print(f"\n按内容拆分完成!共创建 {len(split_points) - 2} 个部分")
# 使用示例:按章节标题拆分
if __name__ == "__main__":
# 匹配中文章节标题(如 "第一章"、"第1章"、"第一节" 等)
chapter_pattern = r'第[一二三四五六七八九十\d]+章'
# 或者匹配英文章节标题
# chapter_pattern = r'Chapter\s+\d+|Section\s+\d+'
split_by_content_pattern(
input_pdf="book.pdf",
pattern=chapter_pattern,
output_folder="book_chapters"
)
```
这种基于内容的拆分方式非常强大,但它依赖于PDF文本提取的质量。有些扫描版的PDF可能无法提取文本,这时就需要考虑其他拆分策略。
## 4. 实战应用:构建完整的PDF处理工作流
掌握了合并和拆分的基本技能后,我们可以将这些技术组合起来,构建完整的自动化工作流。下面我将分享几个在实际工作中非常实用的综合案例。
### 4.1 案例一:自动化报告生成系统
假设你每月需要生成一份综合报告,报告由以下几个部分组成:
1. 封面页(固定模板)
2. 执行摘要(从数据库生成的PDF)
3. 各部门详细报告(多个独立的PDF文件)
4. 附录(包括数据表格和参考资料)
我们可以创建一个自动化脚本来处理这个工作流:
```python
import os
import glob
from datetime import datetime
from pypdf import PdfWriter
def generate_monthly_report(month, year, department_reports_folder, output_folder="reports"):
"""
生成月度综合报告
参数:
month: 月份(1-12)
year: 年份
department_reports_folder: 存放各部门报告的文件夹
output_folder: 输出文件夹
"""
os.makedirs(output_folder, exist_ok=True)
# 1. 准备报告组件
components = []
# 封面页
cover_page = f"templates/cover_{month:02d}_{year}.pdf"
if os.path.exists(cover_page):
components.append(cover_page)
else:
# 使用通用封面
components.append("templates/cover_generic.pdf")
# 执行摘要(假设已生成)
executive_summary = f"generated/exec_summary_{year}_{month:02d}.pdf"
if os.path.exists(executive_summary):
components.append(executive_summary)
# 各部门报告(按特定顺序)
department_order = ["sales", "marketing", "engineering", "finance", "hr"]
for dept in department_order:
dept_pattern = os.path.join(department_reports_folder, f"*{dept}*_{year}_{month:02d}.pdf")
dept_files = glob.glob(dept_pattern)
if dept_files:
# 如果有多个文件,取最新的
dept_files.sort(key=os.path.getmtime, reverse=True)
components.append(dept_files[0])
# 附录
appendix_files = [
"appendices/data_tables.pdf",
"appendices/references.pdf"
]
for appendix in appendix_files:
if os.path.exists(appendix):
components.append(appendix)
# 2. 合并所有组件
merger = PdfWriter()
print(f"正在生成 {year}年{month}月报告,包含以下组件:")
for i, comp in enumerate(components, 1):
if os.path.exists(comp):
merger.append(comp)
print(f" {i}. {os.path.basename(comp)}")
else:
print(f" {i}. {os.path.basename(comp)} (缺失)")
# 3. 保存报告
report_filename = f"monthly_report_{year}_{month:02d}.pdf"
report_path = os.path.join(output_folder, report_filename)
with open(report_path, "wb") as output_file:
merger.write(output_file)
merger.close()
# 4. 生成报告元数据文件(可选)
metadata = {
"generated_date": datetime.now().isoformat(),
"report_period": f"{year}-{month:02d}",
"components": [os.path.basename(c) for c in components if os.path.exists(c)],
"total_pages": len(merger.pages) if hasattr(merger, 'pages') else "未知"
}
# 这里可以添加代码将元数据保存为JSON或写入数据库
print(f"\n报告生成完成: {report_path}")
print(f"文件大小: {os.path.getsize(report_path) / 1024 / 1024:.2f} MB")
return report_path
# 使用示例
if __name__ == "__main__":
# 生成2024年3月的报告
report = generate_monthly_report(
month=3,
year=2024,
department_reports_folder="department_reports",
output_folder="monthly_reports"
)
# 可以进一步处理报告,比如发送邮件或上传到云存储
print(f"报告已保存,路径: {report}")
```
这个系统可以设置为定时任务(比如每月第一天自动运行),完全自动化整个报告生成过程。
### 4.2 案例二:合同管理系统中的PDF处理
在合同管理工作中,经常需要处理大量的PDF文件。下面是一个简化的合同处理系统,展示了如何将pypdf集成到实际业务中:
```python
import os
import shutil
from datetime import datetime
from pypdf import PdfReader, PdfWriter
class ContractProcessor:
"""合同PDF处理器"""
def __init__(self, contracts_folder="contracts"):
self.contracts_folder = contracts_folder
self.processed_folder = os.path.join(contracts_folder, "processed")
self.archive_folder = os.path.join(contracts_folder, "archive")
# 创建必要的文件夹
for folder in [self.processed_folder, self.archive_folder]:
os.makedirs(folder, exist_ok=True)
def extract_signature_pages(self, contract_path, output_folder="signatures"):
"""
从合同中提取签名页
参数:
contract_path: 合同文件路径
output_folder: 签名页保存文件夹
"""
os.makedirs(output_folder, exist_ok=True)
reader = PdfReader(contract_path)
contract_name = os.path.splitext(os.path.basename(contract_path))[0]
# 策略1:查找包含"签名"或"signature"的页面
signature_pages = []
for page_idx, page in enumerate(reader.pages):
text = page.extract_text()
if text:
# 查找签名相关的关键词
keywords = ["签名", "签字", "signature", "sign here", "authorized by"]
if any(keyword.lower() in text.lower() for keyword in keywords):
signature_pages.append(page_idx)
# 策略2:如果没找到,假设最后几页是签名页
if not signature_pages and len(reader.pages) > 0:
# 取最后3页作为签名页
signature_pages = list(range(max(0, len(reader.pages) - 3), len(reader.pages)))
# 提取并保存签名页
if signature_pages:
writer = PdfWriter()
for page_idx in signature_pages:
writer.add_page(reader.pages[page_idx])
output_path = os.path.join(output_folder, f"{contract_name}_signatures.pdf")
with open(output_path, "wb") as output_file:
writer.write(output_file)
writer.close()
print(f"从 {contract_name} 提取了 {len(signature_pages)} 个签名页")
return output_path
else:
print(f"在 {contract_name} 中未找到签名页")
return None
def batch_process_contracts(self, pattern="*.pdf"):
"""
批量处理合同文件夹中的所有合同
参数:
pattern: 文件匹配模式
"""
contract_files = glob.glob(os.path.join(self.contracts_folder, pattern))
print(f"找到 {len(contract_files)} 个合同文件")
processing_log = []
for contract_file in contract_files:
try:
# 跳过已处理的文件
if "processed" in contract_file or "archive" in contract_file:
continue
contract_name = os.path.basename(contract_file)
print(f"\n处理合同: {contract_name}")
# 1. 提取签名页
signature_file = self.extract_signature_pages(contract_file)
# 2. 提取合同基本信息
reader = PdfReader(contract_file)
metadata = {
"filename": contract_name,
"pages": len(reader.pages),
"processed_date": datetime.now().isoformat(),
"signature_pages_extracted": signature_file is not None
}
# 3. 添加到处理日志
processing_log.append(metadata)
# 4. 移动原始文件到归档文件夹
archive_path = os.path.join(self.archive_folder, contract_name)
shutil.move(contract_file, archive_path)
print(f"合同已归档: {archive_path}")
except Exception as e:
print(f"处理合同 {contract_file} 时出错: {e}")
continue
# 保存处理日志
log_file = os.path.join(self.processed_folder, f"processing_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
# 这里可以添加代码将processing_log保存为JSON文件
print(f"\n批量处理完成!处理了 {len(processing_log)} 个合同")
return processing_log
def merge_contracts_by_party(self, party_name, output_folder="merged_contracts"):
"""
按签约方合并合同
参数:
party_name: 签约方名称
output_folder: 输出文件夹
"""
os.makedirs(output_folder, exist_ok=True)
# 在合同文件夹中查找包含特定签约方名称的文件
party_pattern = f"*{party_name}*.pdf"
party_contracts = glob.glob(os.path.join(self.contracts_folder, party_pattern))
if not party_contracts:
# 也在归档文件夹中查找
party_contracts = glob.glob(os.path.join(self.archive_folder, party_pattern))
if party_contracts:
merger = PdfWriter()
for contract in party_contracts:
merger.append(contract)
output_filename = f"contracts_with_{party_name}_{datetime.now().strftime('%Y%m%d')}.pdf"
output_path = os.path.join(output_folder, output_filename)
with open(output_path, "wb") as output_file:
merger.write(output_file)
merger.close()
print(f"合并了 {len(party_contracts)} 个与 {party_name} 相关的合同")
print(f"输出文件: {output_path}")
return output_path
else:
print(f"未找到与 {party_name} 相关的合同")
return None
# 使用示例
if __name__ == "__main__":
processor = ContractProcessor("company_contracts")
# 批量处理所有合同
log = processor.batch_process_contracts()
# 按签约方合并合同
merged = processor.merge_contracts_by_party("ABC公司", "party_contracts")
print("\n合同处理流程完成!")
```
这个合同处理器展示了如何将PDF处理技术应用到具体的业务场景中。通过提取签名页、按签约方合并合同等功能,可以显著提高合同管理工作的效率。
### 4.3 性能优化与错误处理
当处理大量或大型PDF文件时,性能可能成为一个问题。以下是一些优化技巧和错误处理策略:
```python
import time
from pypdf import PdfReader, PdfWriter
import psutil
import gc
def optimize_pdf_processing(input_pdf, output_pdf):
"""
优化PDF处理性能的示例
参数:
input_pdf: 输入文件路径
output_pdf: 输出文件路径
"""
start_time = time.time()
start_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
print(f"开始处理: {input_pdf}")
print(f"初始内存使用: {start_memory:.2f} MB")
try:
# 策略1:使用with语句确保文件正确关闭
with open(input_pdf, "rb") as input_file:
reader = PdfReader(input_file)
# 策略2:分批处理大型文件
total_pages = len(reader.pages)
batch_size = 50 # 每批处理50页
writer = PdfWriter()
for batch_start in range(0, total_pages, batch_size):
batch_end = min(batch_start + batch_size, total_pages)
print(f"处理页面 {batch_start + 1} 到 {batch_end}...")
for page_idx in range(batch_start, batch_end):
page = reader.pages[page_idx]
# 这里可以添加页面处理逻辑
# 例如:提取文本、添加水印等
writer.add_page(page)
# 策略3:定期垃圾回收
if batch_end % 100 == 0:
gc.collect()
current_memory = psutil.Process().memory_info().rss / 1024 / 1024
print(f" 当前内存使用: {current_memory:.2f} MB")
# 策略4:使用临时文件处理大型输出
import tempfile
with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp_file:
tmp_path = tmp_file.name
writer.write(tmp_file)
# 策略5:验证输出文件
if os.path.getsize(tmp_path) > 0:
shutil.move(tmp_path, output_pdf)
print(f"文件已保存: {output_pdf}")
else:
print("错误:输出文件为空")
os.unlink(tmp_path)
writer.close()
except Exception as e:
print(f"处理过程中出错: {e}")
# 策略6:详细的错误信息
import traceback
traceback.print_exc()
return False
finally:
# 策略7:确保资源清理
if 'writer' in locals():
writer.close()
end_time = time.time()
end_memory = psutil.Process().memory_info().rss / 1024 / 1024
print(f"处理完成!")
print(f"耗时: {end_time - start_time:.2f} 秒")
print(f"内存使用变化: {end_memory - start_memory:.2f} MB")
print(f"输出文件大小: {os.path.getsize(output_pdf) / 1024 / 1024:.2f} MB")
return True
# 错误处理装饰器
def handle_pdf_errors(func):
"""处理PDF操作错误的装饰器"""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except FileNotFoundError as e:
print(f"文件未找到错误: {e}")
return None
except PermissionError as e:
print(f"权限错误: {e}")
return None
except Exception as e:
print(f"处理PDF时发生未知错误: {e}")
# 记录详细错误信息
error_log = {
"function": func.__name__,
"args": args,
"kwargs": kwargs,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
# 这里可以添加代码将error_log保存到文件或数据库
return None
return wrapper
# 使用错误处理装饰器
@handle_pdf_errors
def safe_pdf_operation(pdf_path, operation="read"):
"""安全的PDF操作函数"""
if operation == "read":
reader = PdfReader(pdf_path)
return len(reader.pages)
elif operation == "info":
reader = PdfReader(pdf_path)
return {
"pages": len(reader.pages),
"metadata": reader.metadata,
"encrypted": reader.is_encrypted
}
else:
raise ValueError(f"未知操作: {operation}")
# 使用示例
if __name__ == "__main__":
# 测试性能优化
if os.path.exists("large_document.pdf"):
optimize_pdf_processing("large_document.pdf", "optimized_output.pdf")
# 测试错误处理
result = safe_pdf_operation("nonexistent.pdf", "read")
if result is None:
print("安全处理了错误情况")
```
这些优化策略和错误处理技巧可以帮助你构建更健壮、更高效的PDF处理系统,特别是在处理企业级应用时尤为重要。
在实际工作中,我发现最有效的学习方式是将这些代码片段根据具体需求进行组合和修改。每个业务场景都有其独特性,没有一套代码能解决所有问题。但有了这些基础模块,你可以像搭积木一样构建出适合自己工作流的自动化工具。
我刚开始使用pypdf时,也遇到过各种问题:编码错误、内存不足、文件损坏等等。但每次解决问题的过程都让我对这个库有了更深的理解。现在,PDF处理对我来说不再是繁琐的手工劳动,而是一个可以轻松自动化的流程。希望这些经验对你有所帮助,让你也能从重复的PDF操作中解放出来,专注于更有创造性的工作。