Python 如何高效实现 PDF 内容差异对比

最近有接触到 PDF 内容对比，所以分享一下如何用 Python 实现 PDF 内容对比。

1. 安装 PyMuPDF 库

PyMuPDF 提供了丰富的文档操作功能，包括文本/图像提取、页面渲染、文档合并拆分、注释添加等。支持格式包括 PDF、EPUB、XPS 等。它是基于 C 语言库 MuPDF 的 Python 绑定，MuPDF 由 Artifex 公司开发，以高性能和小巧著称。通过 pip install PyMuPDF 安装，但在代码中需通过 import fitz 调用其功能。fitz 是该库的核心模块，fitz 名称源自 MuPDF 的原始渲染引擎 “Fitz”。为保持一致性，PyMuPDF 的 Python 接口沿用了此名称。

pip install pymupdf

import fitz

2. 获取 PDF 内容

fitz.open 是 PyMuPDF（fitz 模块）中用于打开 PDF 或其他支持的文档格式的函数。它返回一个 fitz.Document 对象。
通过 fitz.Document 对象，可以：

• 访问页面：
  使用索引访问文档中的页面，例如 doc[0] 表示第一页。
  每个页面是一个 fitz.Page 对象。
• 获取文档信息：
  获取文档的元数据（如标题、作者、创建时间等）。
  获取文档的页数。

获取 PDF 内容有两种方式：

通过文件路径获取

def get_pdf_content_from_path(pdf_path):"""Get PDF content from a local file path"""pdf = fitz.open(pdf_path)return pdf 

通过 URL 获取

注意通过接口调用获取 repsonce.content 字节类型 content，而不是 response.text 字符串类型 content

属性	response.content	response.text
返回类型	bytes（字节）	str（字符串）
解码	不进行解码，返回原始二进制数据	自动根据 response.encoding 解码
适用场景	处理二进制文件（如图片、PDF 等）	处理文本数据（如 HTML、JSON 等）
手动解码	需要手动解码（如 content.decode(‘utf-8’)）	自动解码，无需额外操作

def get_pdf_content_from_datalake(content_url):"""Get PDF content using content_url"""content = get_content_by_content_url(content_url)try:pdf = fitz.open(filetype="pdf", stream=content)except Exception as e:raise ValueError(f"Failed to open PDF from DataLake for URL: {content_url}. Error: {str(e)}")return pdf

3. 提取 PDF 每页信息

PDF 通常有很多页 content，需要比较每页的 content，前面获取到 fitz.Document，使用索引访问文档中的页面 doc[index] 返回 fitz.Page 对象。
下面是 fitz.Page 的常用属性，我们对比内容只需要用到 get_text() 和 get_pixmap()，通过比较每页的 text 和像素就能找出 PDF 任何细微的差异，包括内容格式，e.g 字体，加粗，高亮，table 布局，图片大小等。

属性/方法	描述
number	当前页面的页码（从 0 开始）。
rect	页面尺寸（矩形区域）。
rotation	页面旋转角度（0、90、180 或 270）。
mediabox	页面媒体框的尺寸。
cropbox	页面裁剪框的尺寸。
get_text()	提取页面文本（支持多种格式，如 “text”、“html”、“json”）。
get_pixmap()	将页面渲染为图像。
search_for()	搜索页面中的文本。
get_images()	获取页面中的嵌入图像信息。
add_annot()	在页面上添加注释。
write()	将页面内容导出为字节流。

其中 get_pixmap() 用于将 PDF 页面渲染为像素图（图像）。它是将 PDF 页面转换为图像格式的核心方法，常用于生成页面的可视化表示或进行图像比较。返回的 fitz.Pixmap 对象包含图像的像素数据和相关信息，常用属性如下：

属性名	描述
samples	图像的原始像素数据（字节流）。
width	图像的宽度（像素）。
height	图像的高度（像素）。
stride	每行像素的字节数。
colorspace	图像的颜色空间（如 RGB、灰度等）。

  # Determine the maximum number of pagesmax_pages = max(len(pdf_base), len(pdf_target))

def extract_page_data(pdf, page_num):"""Extract text and pixel data from a PDF page."""page = pdf[page_num]text = page.get_text()pix = page.get_pixmap()return {"text": text,"pix_samples": pix.samples,"pix_width": pix.width,"pix_height": pix.height,}
def generate_page_data(pdf_base, pdf_target, max_pages, doc_folder):"""Generator to yield page data for multiprocessing."""for page_num in range(max_pages):page_data_base = extract_page_data(pdf_base, page_num)page_data_target = extract_page_data(pdf_target, page_num)yield (page_data_base, page_data_target, page_num, doc_folder)

4. 内容对比

metadata 差异

fitz.Document 对象元数据 metadata 属性，通常包括文档的基本信息，例如标题、作者、创建时间等。如果忽略 metadata 差异，可以忽略此项对比。

以下是 metadata 字典中常见的键及其含义：

键名	描述
title	文档的标题（Title）。
author	文档的作者（Author）。
subject	文档的主题（Subject）。
keywords	文档的关键字（Keywords）。
creator	创建文档的应用程序（Creator）。
producer	生成文档的工具或软件（Producer）。
creationDate	文档的创建日期（Creation Date）。
modDate	文档的最后修改日期（Modification Date）。
trapped	文档是否被标记为“Trapped”（通常为 True 或 False，可能为空）。

compare_metadata(pdf_base.metadata, pdf_target.metadata, result)
def compare_metadata(metadata_base, metadata_target, result):"""Compare PDF metadata"""for key in set(metadata_base.keys()) | set(metadata_target.keys()):if metadata_base.get(key) != metadata_target.get(key):result["metadata_differences"].append(f"Metadata '{key}' differs: pdf_base='{metadata_base.get(key)}', pdf_target='{metadata_target.get(key)}'")

文本对比

ndiff 是 Python 标准库 difflib 中的一个方法，用于逐行比较两个字符串序列，并生成一个可读的差异列表。它特别适合用于文本比较，能够清晰地标记出新增、删除和修改的部分。

difflib.ndiff 的功能

• 输入: 两个字符串序列（通常是通过 splitlines() 分割的多行文本）。
• 输出: 一个迭代器，生成每一行的差异标记。
• 差异标记:
  -：表示在第一个序列中存在，但在第二个序列中不存在的行。
  +：表示在第二个序列中存在，但在第一个序列中不存在的行。
  （空格）：表示两个序列中都存在的行（没有变化）。
  ?：表示上一行的具体差异（通常用于标记字符级别的变化）。

def compare_text_content(page_data_base, page_data_target, page_num, result):"""Compare text content of two pages."""text_base = page_data_base["text"]text_target = page_data_target["text"]if text_base != text_target:result["text_differences"].append(f"Text differs on page {page_num + 1}")diff = list(difflib.ndiff(text_base.splitlines(), text_target.splitlines()))differences = [d for d in diff if d.startswith('+ ') or d.startswith('- ')]if differences:result["text_differences"].append(f"Page {page_num + 1} specific differences: {differences[:5]}...")

可视化对比

比较两个 PDF 页面视觉内容，通过比较页面的像素数据来检测页面之间的视觉差异。

• 页面尺寸比较:
  首先比较两个页面的宽度和高度，如果页面尺寸不同，记录差异并退出函数。
• 像素数据比较:
  将页面的像素数据转换为图像对象。使用 PIL.Image.frombytes 将页面的像素数据转换为 RGB 图像对象。
  使用 ImageChops.difference 计算两个图像的差异，返回一个差异图像，其中每个像素的值表示两个图像对应像素的差异程度。
• 保存差异图像:
  如果发现差异，保存基准页面、目标页面和差异图像到指定的文件夹。
  记录差异信息到 result 字典中。

def compare_visual_content(page_data_base, page_data_target, page_num, doc_folder, result):"""Compare visual content of two pages."""if (page_data_base["pix_width"] != page_data_target["pix_width"] orpage_data_base["pix_height"] != page_data_target["pix_height"]):result["format_differences"].append(f"Page {page_num + 1} size differs: PDF_base={page_data_base['pix_width']}x{page_data_base['pix_height']}, "f"PDF_target={page_data_target['pix_width']}x{page_data_target['pix_height']}")returnimg_base = Image.frombytes("RGB", [page_data_base["pix_width"], page_data_base["pix_height"]], page_data_base["pix_samples"])img_target = Image.frombytes("RGB", [page_data_target["pix_width"], page_data_target["pix_height"]], page_data_target["pix_samples"])diff_img = ImageChops.difference(img_base, img_target)if np.any(np.array(diff_img)):img_base_path = os.path.join(doc_folder, f"page_{page_num + 1}_pdf_base.png")img_target_path = os.path.join(doc_folder, f"page_{page_num + 1}_pdf_target.png")diff_path = os.path.join(doc_folder, f"page_{page_num + 1}_diff.png")img_base.save(img_base_path)img_target.save(img_target_path)diff_img.save(diff_path)result["format_differences"].append(f"differs on page {page_num + 1}: difference image saved at {diff_path}")

5. 提升对比效率

通过哈希值快速判断页面是否相同

通过比较页面内容的哈希值（包括文本和像素数据），如果哈希值相同，则跳过进一步比较。
如果哈希值不同，调用 compare_text_content 和 compare_visual_content 方法分别比较文本和视觉内容。

def hash_page_content(page_data):"""Generate a hash for the page content."""text_hash = hashlib.md5(page_data["text"].encode()).hexdigest()pix_hash = hashlib.md5(page_data["pix_samples"]).hexdigest()return text_hash, pix_hashdef compare_page(page_data_base, page_data_target, page_num, doc_folder):"""Compare a single page for text and visual differences."""result = {"text_differences": [],"format_differences": []}try:# Compare hashes firstbase_hash = hash_page_content(page_data_base)target_hash = hash_page_content(page_data_target)if base_hash == target_hash:return result  # Skip comparison if hashes are identical# Compare text and visual contentcompare_text_content(page_data_base, page_data_target, page_num, result)compare_visual_content(page_data_base, page_data_target, page_num, doc_folder, result)except Exception as e:result["format_differences"].append(f"Failed to compare page {page_num + 1}: {str(e)}")return result

早停机制

如果 PDF 差异页面非常很多，后续的页面差异其实是无意义的，我们可以设定一个差异页面数量的最大值，比如 3 或 5，当发现的差异页面数量达到指定的最大值时，函数会停止进一步的比较。

def compare_page_with_limit(args, diff_page_count, max_diff_pages, lock):"""Compare a single page with early termination."""page_data_base, page_data_target, page_num, doc_folder = argswith lock:if diff_page_count.value >= max_diff_pages:return None  # Skip further processing if limit is reachedpage_result = compare_page(page_data_base, page_data_target, page_num, doc_folder)if page_result["text_differences"] or page_result["format_differences"]:with lock:diff_page_count.value += 1return page_result

多进程机制

如果需要比较的 PDF 文件比较多，我们也可以采用多进程并发比较，提升脚本执行时间。这里可以根据实际情况，是基于 PDF 之间并行，还是基于单个 PDF 页面之间并行。我这边是基于 PDF 页面之间并发执行的，考虑到大多数 PDF 页面达上百页，页面之间并发效率更高。

pool.starmap 是 Python 中 multiprocessing.Pool 提供的一种方法，用于在多进程环境下并行执行函数。它类似于 map 方法，但支持将多个参数传递给目标函数。

这里定义了一个 diff_page_count 共享变量（通过 manager.Value 创建），因为是 int 型，所以在多进程环境下需要使用 lock 来保护它。这是因为 manager.Value 本身并不能保证对其值的操作是原子的（atomic）。
共享变量的非原子操作，对共享变量的操作（如 diff_page_count.value += 1）实际上是由多个步骤组成的：

• 读取当前值。
• 增加值。
• 写回新值。

在多进程环境下，如果多个进程同时执行这些步骤，就可能导致数据竞争（race condition），从而导致共享变量的值不正确。假设两个进程同时读取 diff_page_count.value 的值为 5，然后分别将其加 1 并写回。最终的结果可能是 6 而不是预期的 7，因为两个进程的操作互相覆盖了。使用 lock 可以确保在一个进程修改共享变量时，其他进程必须等待，直到当前进程完成操作并释放锁。这就避免了数据竞争，确保共享变量的值始终正确。

当然如果换成 diff_page_count = manager.list()，它的操作（如添加或删除元素）是线程安全的，底层已经实现了同步机制。因此，多个进程可以安全地向列表中添加元素，而无需显式使用 lock。但是 manager.list() 的操作比直接操作 manager.Value 稍慢，因为它需要处理线程安全。如果性能是关键问题，仍然可以考虑使用 manager.Value 和 lock。

def prepare_output_folder(output_folder, pdf_object_id):"""Prepare the output folder for storing comparison results."""output_folder = os.path.join(constants.OUTPUT_DIR, output_folder)os.makedirs(output_folder, exist_ok=True)doc_folder = os.path.join(output_folder, pdf_object_id.replace(":", "_"))clear_and_create_content_dir(doc_folder)return doc_folderdef compare_pdf(pdf_base_path, pdf_target_path, pdf_object_id, pdf_base_object_url, pdf_target_object_url,is_from_datalake=True, output_folder="pdf_diff_results", max_diff_pages=3):"""Compare two PDF files for content and format differences"""# Prepare output folderdoc_folder = prepare_output_folder(output_folder, pdf_object_id)# Initialize resultresult = {"text_differences": [],"format_differences": [],"metadata_differences": [],"page_count": {"pdf_base": 0, "pdf_target": 0}}# Open PDF filespdf_base = get_pdf_content_from_datalake(pdf_base_object_url) if is_from_datalake else get_pdf_content_from_path(pdf_base_path)pdf_target = get_pdf_content_from_datalake(pdf_target_object_url) if is_from_datalake else get_pdf_content_from_path(pdf_target_path)# Compare page countresult["page_count"]["pdf_base"] = len(pdf_base)result["page_count"]["pdf_target"] = len(pdf_target)# Compare metadata, ignore differences in creation/modification dates# compare_metadata(pdf_base.metadata, pdf_target.metadata, result)# Determine the maximum number of pagesmax_pages = max(len(pdf_base), len(pdf_target))# Compare pages in parallel using a generatorwith Manager() as manager:# Shared counter for tracking pages with differencesdiff_page_count = manager.Value('i', 0)lock = manager.Lock()# Create a pool of worker processeswith Pool() as pool:page_results = pool.starmap(compare_page_with_limit,[(args, diff_page_count, max_diff_pages, lock) for args in generate_page_data(pdf_base, pdf_target, max_pages, doc_folder)])if diff_page_count.value >= max_diff_pages:print(f"Early termination: {diff_page_count.value} pages with differences found, stopping further processing.")pool.terminate()pool.join()# Aggregate resultsfor page_result in page_results:if page_result is None:continue  # Skip if terminated earlyresult["text_differences"].extend(page_result["text_differences"])result["format_differences"].extend(page_result["format_differences"])return result

6. 其他

还有一些其他细节问题，这里就不细说了，一个完整的脚本执行是需要考虑很多因素的，目的就是为了全自动化，减少人工干预成本，提高整体效率。
这里罗列一些：

• 测试数据收集和配置，方便后期定制化执行不同的测试数据集
• 脚本执行过程中的 log，方便 troubleshooting
• 生成测试报告，包括细节信息，汇总信息（total，fail，pass），及其他统计信息，方便 triage
• 部署到 Jenkins 上日常执行，并发送测试报告，方便 CICD