使用Python高效实现PDF内容差异对比的方法详解

目录

• 1. 安装 PyMuPDF 库
• 2. 获取 PDF 内容
• 通过文件路径获取
• 通过 URL 获取
• 3. 提取 PDF 每页信息
• 4. 内容对比
• metadata 差异
• 文本对比
• 可视化对比
• 5. 提升对比效率
• 通过哈希值快速判断页面是否相同
• 早停机制
• 多进程机制
• 6. 其他

1. 安装 PyMuPDF 库

PyMuPDF 提供了丰富的文档操作功能，包括文本/图像提取、页面渲染、文档合并拆分、注释添加等。dqEqwkK支持格式包括 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

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 布局，图片大小等。

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

  # Determine the maximum number of pages
    max_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 字典中常见的键及其含义：

compare_metadata(pdf_base.metadata, pdf_target.metadata, result)
def compare_metadata(metadata_base, metadata_target, result):
    "dqEqwkK""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='{mphpetadata_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 diffeandroidrences:
            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"] or
            page_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']}"
        )
        return

    img_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_hash


def 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 first
        base_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 content
        compare_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 差异页面非常很多，后续的页面dqEqwkK差异其实是无意义的，我们可以设定一个差异页面数量的最大值，比如 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 = args
    with lock:
        if diff_page_count.value >= max_diff_pages:
            return None  # Skip further processing if limit is reached
    page_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 += 1
    return 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_folder


def 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 folder
    doc_folder = prepare_output_folder(output_folder, pdf_object_id)
    
    # Initialize result
    result = {
        "text_differences": [],
        "format_differences": [],
        "metadata_differences": [],
        "page_count": {"pdf_base": 0, "pdf_target": 0}
    }
    
 
    # Open PDF files
    pdf_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 count
    result["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 pages
    max_pages = max(len(pdf_base), len(pdf_target))

    # Compare pages in parallel using a generator
    with Manager() as manager:
        # Shared counter for tracking pages with differences
        diff_page_count = manager.Value('i', 0)
        lock = manager.Lock()
        # Create a pool of worker processes
        with 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 results
    for page_result in page_results:
        if page_result is None:
            continue  # Skip if terminated early
        result["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

以上就是使用Python高效实现PDF内容差异对比的方法详解的详细内容，更多关于Python PDF内容差异对比的资料请关注编程客栈(www.devze.com)其它相关文章！