最近有接觸到 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