pdfplumber：PDF文本与表格数据提取

Python 办公自动化专题 · 从PDF中精准提取结构化数据

专题：Python 自动化办公系统学习

关键词：Python, 自动化办公, pdfplumber, PDF提取, 表格提取, 数据提取, 文本解析, Python办公

一、pdfplumber概述

什么是pdfplumber

pdfplumber 是一个基于 pdfminer.six 构建的 Python 库，专注于从 PDF 文件中精确提取文本和表格数据。与传统的 PDF 解析库相比，pdfplumber 提供了更细粒度的访问控制，允许开发者获取每个字符的精确位置坐标（x0, y0, x1, y1）、字体名称、字号大小等信息，这使得它特别适合需要保留文档结构布局的提取场景。

pdfplumber 的核心优势在于其对 PDF 内部对象的完整暴露。它不像其他库那样只提供"黑盒"式的文本提取，而是将 PDF 的页面分解为字符（Char）、矩形（Rect）、线（Line）、图像（Image）等基础对象，让开发者可以根据实际需求灵活组合使用。这种设计思路使得 pdfplumber 在处理复杂排版、多栏布局、带表格的文档时表现出色。

与其他PDF库的对比

特性	pdfplumber	PyPDF2	camelot	tabula-py
文本提取精度	高（字符级坐标）	中	低	低
表格提取	支持（可调参）	不支持	专精（Lattice/Stream）	专精
可视化调试	内置 to_image	不支持	支持	有限
字符级访问	完整支持	不支持	不支持	不支持
学习曲线	中等	简单	中等	中等
中文支持	良好	一般	良好	良好

安装方法

# 基础安装
pip install pdfplumber

# 如需OCR支持，同时安装
pip install pdfplumber pytesseract pillow

# 如需配合pandas进行数据处理
pip install pdfplumber pandas openpyxl

基本概念：PDF对象模型

pdfplumber 将 PDF 文档按照层次结构组织。最顶层是 PDF 文档对象，包含若干页面（Page）。每个页面由一系列基础对象组成，主要包括：Char（字符，最小的文本单元，包含字体、大小、坐标等属性）、Rect（矩形，通常用于表示表格边框或背景色块）、Line（线段，用于绘制表格线或分隔线）。理解这一对象模型是高效使用 pdfplumber 的关键，因为几乎所有的高级功能（如表格提取、区域文本提取）都是基于对这些基础对象的筛选和组合实现的。

import pdfplumber

# 打开PDF文件
with pdfplumber.open("sample.pdf") as pdf:
    # 获取页面总数
    print(f"总页数: {len(pdf.pages)}")

    # 获取第一页
    first_page = pdf.pages[0]

    # 查看页面尺寸
    print(f"页面尺寸: {first_page.width} x {first_page.height}")

    # 提取页面纯文本
    text = first_page.extract_text()
    print(text[:500])

# 深入探索页面对象
with pdfplumber.open("sample.pdf") as pdf:
    page = pdf.pages[0]

    # 查看页面上所有字符对象
    chars = page.chars
    print(f"字符数量: {len(chars)}")
    print("第一个字符属性:")
    for key, value in chars[0].items():
        print(f"  {key}: {value}")

    # 查看矩形对象（如表格边框）
    rects = page.rects
    print(f"矩形数量: {len(rects)}")

    # 查看线段对象
    lines = page.lines
    print(f"线段数量: {len(lines)}")

二、文本提取

提取纯文本

pdfplumber 最基础的功能就是提取PDF中的纯文本内容。通过页面对象的 extract_text() 方法，开发者可以快速获取页面上的所有文本。然而需要注意的是，这个方法的输出结果依赖于PDF内部文本对象的排列顺序，有时候文本顺序可能不符合人类阅读习惯（尤其是在多栏布局或复杂排版的情况下）。pdfplumber 内部会尝试对文本进行排序，但在某些场景下仍需要手动干预。

import pdfplumber

with pdfplumber.open("report.pdf") as pdf:
    # 提取所有页面的文本
    all_text = ""
    for i, page in enumerate(pdf.pages):
        text = page.extract_text()
        all_text += f"\n\n=== 第{i+1}页 ===\n\n{text}"

    # 写入纯文本文件
    with open("report.txt", "w", encoding="utf-8") as f:
        f.write(all_text)

    print("文本提取完成，已保存到 report.txt")

保留位置文本（按区域提取）

在实际项目中，我们往往只需要提取页面特定区域内的文本，例如发票的票头区域、合同的签署区域或者报表的表头部分。pdfplumber 允许通过坐标（x0, y0, x1, y1）指定裁剪区域，然后仅提取该区域内的文本。坐标原点位于页面左上角，x 轴向右为正，y 轴向下为正，单位为点（point，1/72英寸）。这种按区域提取的能力在处理结构化文档时极为有用。

import pdfplumber

with pdfplumber.open("invoice.pdf") as pdf:
    page = pdf.pages[0]

    # 定义裁剪区域（左，上，右，下）
    # 例如：提取发票右上角的发票号码区域
    invoice_number_bbox = (page.width * 0.6, 30, page.width - 20, 80)
    cropped = page.within_bbox(invoice_number_bbox)
    invoice_number = cropped.extract_text()
    print(f"发票号码区域文本: {invoice_number}")

    # 提取页面左上角的标题区域
    title_bbox = (50, 20, page.width - 50, 100)
    title_area = page.within_bbox(title_bbox)
    title_text = title_area.extract_text()
    print(f"标题区域: {title_text}")

    # 排除页面页脚区域
    body_area = page.within_bbox((0, 0, page.width, page.height - 50))
    body_text = body_area.extract_text()

按关键字提取

在合同审核、报告分析等场景中，常常需要提取特定关键字附近的文本内容。通过结合 pdfplumber 的字符级坐标信息和正则表达式，可以实现精准的关键字定位和上下文提取。这种方法比简单的全文搜索更为强大，因为它不仅知道关键字"在哪里"，还知道它在页面上的精确位置。

import pdfplumber
import re

def extract_after_keyword(page, keyword, chars_after=200):
    """在页面中查找关键字，并提取关键字后的指定字符数文本"""
    text = page.extract_text()
    if not text:
        return None

    match = re.search(keyword, text)
    if match:
        start = match.end()
        return text[start:start + chars_after]
    return None

def extract_between_keywords(page, start_kw, end_kw):
    """提取两个关键字之间的所有文本"""
    text = page.extract_text()
    if not text:
        return None

    pattern = re.escape(start_kw) + "(.*?)" + re.escape(end_kw)
    match = re.search(pattern, text, re.DOTALL)
    if match:
        return match.group(1).strip()
    return None

with pdfplumber.open("contract.pdf") as pdf:
    page = pdf.pages[0]

    # 提取"合同金额"后面的内容
    amount_text = extract_after_keyword(page, "合同金额")
    print(f"合同金额信息: {amount_text}")

    # 提取"违约责任"和"争议解决"之间的内容
    liability = extract_between_keywords(page, "违约责任", "争议解决")
    print(f"违约责任条款: {liability}")

文本排序策略

PDF 中的文本对象顺序不一定与视觉阅读顺序一致，特别是在多栏布局或表格混排的页面中。pdfplumber 提供了 extract_text() 方法的 x_tolerance 和 y_tolerance 参数，用于控制文本合并的策略。x_tolerance 决定了在同一行中多个字符之间的最大允许间距（超过该值则认为分属不同单词），y_tolerance 则决定了多行文本之间的垂直间距阈值。合理调整这些参数可以显著改善提取结果的准确性。对于复杂的多栏布局，通常需要先手动将页面分割为多个区域，再分别提取各区域的文本。

with pdfplumber.open("multi_column.pdf") as pdf:
    page = pdf.pages[0]

    # 默认提取（可能顺序错乱）
    default_text = page.extract_text()
    print("默认提取:", default_text[:200])

    # 调整x_tolerance，控制同一行字符合并的容差
    tuned_text = page.extract_text(x_tolerance=3, y_tolerance=3)
    print("调整参数后:", tuned_text[:200])

    # 手动分栏：分别提取左栏和右栏
    mid_x = page.width / 2
    left_col = page.within_bbox((0, 0, mid_x, page.height))
    right_col = page.within_bbox((mid_x, 0, page.width, page.height))

    left_text = left_col.extract_text()
    right_text = right_col.extract_text()

    # 按阅读顺序拼接（先左后右）
    ordered_text = left_text + "\n\n" + right_text
    print("分栏提取后:", ordered_text[:300])

三、表格提取

find_tables方法详解

表格提取是 pdfplumber 最强大的功能之一。页面对象的 find_tables() 方法会自动检测页面上的表格结构，返回 Table 对象的列表。每个 Table 对象包含 rows 和 cells 属性，可以方便地访问表格的每一行和每一格。pdfplumber 的表格检测算法基于分析页面上的矩形和线段对象：它会寻找由直线或矩形边框构成的网格结构，然后从这些结构中推断出表格的行列划分。

需要注意的是，pdfplumber 的表格提取效果很大程度上依赖于 PDF 文件中表格结构的清晰程度。对于有明确边框线的表格（Lattice 型），pdfplumber 通常能获得非常准确的提取结果；对于没有边框线、仅靠对齐暗示的表格（Stream 型），则需要调整参数或采用更复杂的策略。

import pdfplumber

with pdfplumber.open("table_report.pdf") as pdf:
    page = pdf.pages[0]

    # 查找页面上所有表格
    tables = page.find_tables()
    print(f"检测到 {len(tables)} 个表格")

    # 提取第一个表格
    if tables:
        table = tables[0]

        # 获取表格原始数据（列表的列表）
        data = table.extract()
        for row_idx, row in enumerate(data):
            print(f"第{row_idx}行: {row}")

        # 获取表格的边界坐标
        bbox = table.bbox
        print(f"表格位置: x0={bbox[0]}, y0={bbox[1]}, x1={bbox[2]}, y1={bbox[3]}")

        # 获取表格的行数和列数
        print(f"表格尺寸: {len(table.rows)}行 x {len(table.columns)}列")

表格参数调优

pdfplumber 的 find_tables() 方法提供了多个参数来适应不同的表格样式。vertical_strategy 和 horizontal_strategy 分别控制垂直和水平表格线的检测策略，可选值包括 "lines"（仅使用线段）、"rects"（仅使用矩形）、"explicit"（手动指定）、"intersections"（线段的交叉点）和 "text"（基于文本对齐）。当表格边框不完整或完全没有边框时，使用 "text" 策略可以通过分析文本的水平和垂直对齐来推断表格结构，这在处理无边框表格时非常关键。

with pdfplumber.open("table_no_border.pdf") as pdf:
    page = pdf.pages[0]

    # 默认策略：基于线段检测
    tables_default = page.find_tables()

    # 基于文本对齐推断表格结构（适用于无边框表格）
    tables_text = page.find_tables(
        vertical_strategy="text",
        horizontal_strategy="text"
    )

    # 自定义表格线检测参数
    tables_tuned = page.find_tables(
        vertical_strategy="lines",
        horizontal_strategy="lines",
        # 最小表格线长度（过滤短线段）
        min_words_vertical=5,
        min_words_horizontal=3,
        # 交点合并容差
        intersection_tolerance=5,
        # 联合边缘检测（处理不连续的边框）
        join_tolerance=3,
        # 边缘检测的边距
        edge_min_length=10,
    )

    if tables_tuned:
        data = tables_tuned[0].extract()
        for row in data:
            print(row)

表格数据清洗

提取出的表格原始数据通常包含各种噪音：空白单元格、多余的换行符、前后空格、错误的字符编码等。因此，对提取结果进行清洗是必不可少的一步。常见的清洗操作包括去除首尾空白字符、替换特殊空白符、合并多行文本、处理空值和去除完全空的行。这些清洗操作最好封装成可复用的函数，以保持主逻辑的简洁性。

def clean_table_data(raw_data):
    """清洗表格数据：去除空白、修复格式"""
    cleaned = []
    for row in raw_data:
        # 去除每个单元格的前后空白
        clean_row = []
        for cell in row:
            if cell is None:
                clean_row.append("")
            else:
                # 去除多余换行和空白
                cell = cell.replace("\n", " ").strip()
                clean_row.append(cell)

        # 跳过完全空的行
        if any(cell for cell in clean_row):
            cleaned.append(clean_row)

    return cleaned

with pdfplumber.open("report.pdf") as pdf:
    page = pdf.pages[0]
    tables = page.find_tables()

    if tables:
        raw = tables[0].extract()
        print("原始数据:")
        for row in raw:
            print(row)

        cleaned = clean_table_data(raw)
        print("\n清洗后数据:")
        for row in cleaned:
            print(row)

复杂表格处理

实际工作中的表格往往具有复杂的结构：合并单元格、跨页表格、不规则表头、混合了图片和文本的单元格等。处理合并单元格时，需要识别哪些单元格跨越了多行或多列，然后根据上下文填充缺失的标签。跨页表格则需要将多个页面上的表格片段拼接起来，关键在于识别表头的重复模式并去除重复。对于更为复杂的情况（如表格内嵌图片），可能需要结合 OCR 技术或人工干预。

def merge_table_pages(pdf_path):
    """合并跨页表格，自动处理重复表头"""
    all_rows = []
    header_row = None

    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            tables = page.find_tables()
            if not tables:
                continue

            table = tables[0]
            data = table.extract()

            if not data:
                continue

            if header_row is None:
                # 第一页：保存表头和数据
                header_row = data[0]
                all_rows.extend(data[1:])
            else:
                # 后续页：检查第一行是否与表头相同
                first_row = data[0]
                if first_row == header_row:
                    # 跳过重复表头，只取数据行
                    all_rows.extend(data[1:])
                else:
                    # 表头不一致，全部保留
                    all_rows.extend(data)

    return [header_row] + all_rows if header_row else all_rows

# 使用示例
merged_data = merge_table_pages("long_table_report.pdf")
print(f"合并后共 {len(merged_data)} 行数据")
for row in merged_data[:5]:
    print(row)

四、可视化调试

to_image 调试工具

pdfplumber 提供了内置的可视化调试工具 to_image()，可以将 PDF 页面渲染为图像，并在图像上叠加显示各种调试信息。这对于理解页面的结构布局、验证表格检测结果、调试坐标参数非常有用。to_image() 方法返回一个 PageImage 对象，支持在图像上绘制矩形框、标注文本位置、标记表格边界等操作。调试完成后，可以保存为 PNG 图片进行人工审核。

import pdfplumber

with pdfplumber.open("sample.pdf") as pdf:
    page = pdf.pages[0]

    # 将页面渲染为图像
    im = page.to_image(resolution=150)

    # 在图像上标记所有检测到的表格
    tables = page.find_tables()
    for table in tables:
        # 用红色框标记表格边界
        im.draw_rect(table.bbox, "red")
        for row in table.rows:
            # 用蓝色框标记每一行
            for cell in row.cells:
                im.draw_rect(cell, "blue", fill="blue", fill_opacity=0.1)

    # 保存调试图像
    im.save("debug_table.png")
    print("调试图像已保存到 debug_table.png")

矩形、线和字符的可视化

在对 PDF 结构进行深度分析时，将页面上的矩形、线段和字符对象可视化地呈现出来，可以帮助我们理解 PDF 的内部构造方式。例如，表格的边框可能由多个矩形拼接而成，也可能由四条独立的线段构成。通过可视化这些基础对象，可以直观地看到表格线是否连续、矩形是否完整覆盖了表格区域，从而选择最合适的提取策略。

with pdfplumber.open("complex_layout.pdf") as pdf:
    page = pdf.pages[0]
    im = page.to_image(resolution=150)

    # 用绿色标记所有矩形对象
    for rect in page.rects:
        bbox = (rect["x0"], rect["y0"], rect["x1"], rect["y1"])
        im.draw_rect(bbox, "green")

    # 用黄色标记所有线段
    for line in page.lines:
        if line["height"] > 0:  # 垂直线
            im.draw_line(
                (line["x0"], line["top"], line["x1"], line["bottom"]),
                "yellow", stroke_width=2
            )
        else:  # 水平线
            im.draw_line(
                (line["x0"], line["top"], line["x1"], line["bottom"]),
                "yellow", stroke_width=2
            )

    # 用白色点标记所有字符位置
    for char in page.chars[::10]:  # 每10个字符标记一个以减少密度
        bbox = (char["x0"], char["top"], char["x1"], char["bottom"])
        im.draw_rect(bbox, "white", fill="white")

    im.save("debug_objects.png")
    print("页面对象可视化已保存")

页面结构分析

可视化调试的最终目的是理解页面的结构层次，从而为提取策略的制定提供依据。通过分析字符的分布密度、矩形和线段的位置关系，可以判断页面的布局类型（单栏、双栏、混合等）、识别页眉页脚区域、发现隐藏的表格结构。这些洞察对于编写健壮的提取代码至关重要。建议在开发提取脚本的第一步，先对代表性页面进行可视化分析，然后再编写提取逻辑。

def analyze_page_structure(page):
    """分析页面结构并输出统计信息"""
    print("========== 页面结构分析 ==========")
    print(f"页面尺寸: {page.width:.1f} x {page.height:.1f}")

    chars = page.chars
    rects = page.rects
    lines = page.lines
    images = page.images

    print(f"字符总数: {len(chars)}")
    print(f"矩形总数: {len(rects)}")
    print(f"线段总数: {len(lines)}")
    print(f"图片总数: {len(images)}")

    # 分析字体使用情况
    fonts = set()
    for char in chars:
        fonts.add(char["fontname"])
    print(f"使用字体: {fonts}")

    # 分析字号分布
    sizes = {}
    for char in chars:
        size = round(char["size"], 1)
        sizes[size] = sizes.get(size, 0) + 1
    print(f"字号分布: {dict(sorted(sizes.items()))}")

    # 检测表头区域（使用字号变化）
    if chars:
        avg_size = sum(c["size"] for c in chars) / len(chars)
        large_chars = [c for c in chars if c["size"] > avg_size * 1.2]
        if large_chars:
            min_y = min(c["top"] for c in large_chars)
            max_y = max(c["bottom"] for c in large_chars)
            print(f"疑似标题区域: y=[{min_y:.1f}, {max_y:.1f}]")

    return {
        "chars": len(chars),
        "rects": len(rects),
        "lines": len(lines),
        "images": len(images),
        "fonts": fonts,
        "sizes": sizes,
    }

with pdfplumber.open("unknown.pdf") as pdf:
    info = analyze_page_structure(pdf.pages[0])

五、字符与对象分析

字符属性详解

pdfplumber 中的 Char 对象是 PDF 中最基本的文本单元，提供了极为丰富的属性信息。每个 Char 对象除了包含字符本身的文本内容（text）外，还记录了其精确的边界框坐标（x0, y0, x1, y1）、字体名称（fontname）、字号（size）、字体是否加粗（fontname 中的 Bold 标识）、渲染模式（rendering_mode）、字符间距（width）等信息。这些细粒度的属性使得开发者可以实现许多高级功能，如基于字体大小识别标题层级、基于字体名称区分正文和注释、基于位置坐标重建阅读顺序等。

通过分析字符属性的分布模式，我们可以自动识别文档的结构元素：大号字体的文本通常为标题或表头，等宽字体的文本可能是代码或数据，特定的颜色可能表示批注或修改标记。这种基于字符属性的文档结构分析，是实现智能化 PDF 提取的基础。

def analyze_chars_detail(chars):
    """详细分析字符属性"""
    print(f"共 {len(chars)} 个字符")
    print("\n=== 字体统计 ===")
    font_stats = {}
    for c in chars:
        fn = c["fontname"]
        if fn not in font_stats:
            font_stats[fn] = {"count": 0, "sizes": set(), "text": []}
        font_stats[fn]["count"] += 1
        font_stats[fn]["sizes"].add(round(c["size"], 1))
        if len(font_stats[fn]["text"]) < 10:
            font_stats[fn]["text"].append(c["text"])

    for fn, stats in font_stats.items():
        print(f"  字体 '{fn}': {stats['count']}个字符, 字号: {sorted(stats['sizes'])}")
        print(f"    示例文本: {''.join(stats['text'][:5])}")

    print("\n=== 字号分布 ===")
    size_dist = {}
    for c in chars:
        sz = round(c["size"], 1)
        size_dist[sz] = size_dist.get(sz, 0) + 1
    for sz in sorted(size_dist):
        bar = "█" * min(size_dist[sz] // 10, 50)
        print(f"  {sz:4.1f}pt: {size_dist[sz]:5d}个 {bar}")

# 使用示例
with pdfplumber.open("document.pdf") as pdf:
    chars = pdf.pages[0].chars
    analyze_chars_detail(chars)

矩形分析

PDF 中的矩形对象（Rect）通常表示表格的单元格背景、边框色块、按钮或图标区域。通过分析页面上的矩形对象，我们可以了解表格的结构布局、识别高亮区域、检测表单字段位置。矩形的属性包括四条边的坐标、填充颜色（stroking_color, non_stroking_color）、线宽（linewidth）等。在表格提取的上下文中，矩形分析可以帮助我们判断表格边框的完整性和连续性。

def analyze_rects(page):
    """分析页面上的矩形对象"""
    rects = page.rects
    if not rects:
        print("页面上没有矩形对象")
        return

    print(f"矩形数量: {len(rects)}")

    # 矩形尺寸分析
    widths = [r["x1"] - r["x0"] for r in rects]
    heights = [r["y1"] - r["y0"] for r in rects]

    print(f"矩形宽度范围: {min(widths):.1f} - {max(widths):.1f}")
    print(f"矩形高度范围: {min(heights):.1f} - {max(heights):.1f}")

    # 查找表格区域（大尺寸矩形）
    page_area = page.width * page.height
    large_rects = [r for r in rects
                   if (r["x1"] - r["x0"]) * (r["y1"] - r["y0"]) > page_area * 0.01]
    print(f"大尺寸矩形（>1%页面）: {len(large_rects)}个")

    # 检查矩形填充色
    filled = [r for r in rects if r.get("non_stroking_color")]
    if filled:
        print(f"有填充色的矩形: {len(filled)}个（可能是表格表头背景）")

    # 检测可能的表格行间隔
    y_positions = sorted(set(round(r["y0"]) for r in large_rects))
    if len(y_positions) > 2:
        gaps = [y_positions[i+1] - y_positions[i] for i in range(len(y_positions)-1)]
        print(f"行间隔: 最小值={min(gaps)}, 最大值={max(gaps)}, 平均={sum(gaps)/len(gaps):.1f}")

    return large_rects

with pdfplumber.open("invoice.pdf") as pdf:
    analyze_rects(pdf.pages[0])

线分析

线段（Line）是 PDF 表格结构的另一关键组成元素。表格的水平线和垂直线分别由水平线段和垂直线段表示。通过分析线段的起始和结束坐标，可以重构出表格的网格结构。pdfplumber 在 find_tables() 内部就使用了类似的线段分析算法。当默认的表格检测效果不佳时，手动分析线段分布可以帮助我们理解问题所在，从而有针对性地调整参数。

def analyze_lines(page):
    """分析页面上的线段对象"""
    lines = page.lines
    if not lines:
        print("页面上没有线段对象")
        return

    # 区分水平线和垂直线
    h_lines = [l for l in lines if abs(l["y0"] - l["y1"]) < 0.5]
    v_lines = [l for l in lines if abs(l["x0"] - l["x1"]) < 0.5]

    print(f"水平线段: {len(h_lines)}条")
    print(f"垂直线段: {len(v_lines)}条")

    # 水平线Y坐标聚类（识别表格行）
    h_y_positions = sorted(set(round(l["y0"], 1) for l in h_lines))
    if len(h_y_positions) > 2:
        print(f"水平线Y坐标层次: {len(h_y_positions)}个不同位置")
        print(f"  Y范围: {min(h_y_positions):.1f} - {max(h_y_positions):.1f}")

        # 检查是否形成均匀网格（表格特征）
        gaps = [h_y_positions[i+1] - h_y_positions[i] for i in range(len(h_y_positions)-1)]
        avg_gap = sum(gaps) / len(gaps)
        uniformity = max(gaps) - min(gaps)
        if uniformity < avg_gap * 0.3:
            print(f"  → 水平线分布均匀，疑似表格行（平均行距: {avg_gap:.1f}）")

    # 垂直线X坐标聚类（识别表格列）
    v_x_positions = sorted(set(round(l["x0"], 1) for l in v_lines))
    if len(v_x_positions) > 1:
        print(f"垂直线X坐标层次: {len(v_x_positions)}个不同位置")
        print(f"  X范围: {min(v_x_positions):.1f} - {max(v_x_positions):.1f}")

    return h_lines, v_lines

with pdfplumber.open("table_report.pdf") as pdf:
    analyze_lines(pdf.pages[0])

文本块组合

在实际场景中，PDF 中的一个"段落"可能由多个独立的字符或文本行对象组成。通过分析字符的位置和间距，我们可以将分散的字符重新组合成有意义的文本块、段落或段落组。pdfplumber 的 extract_text() 方法内部已经做了基本的文本组合工作，但在需要更精细控制时（如保留段落边界、识别列表项、检测缩进层次），需要开发者自行实现文本块的组合逻辑。

def group_chars_into_words(chars, x_tolerance=2):
    """将字符按水平位置组合成单词"""
    if not chars:
        return []

    # 按行（Y坐标相近）分组
    sorted_chars = sorted(chars, key=lambda c: (-c["top"], c["x0"]))

    lines = []
    current_line = [sorted_chars[0]]
    for c in sorted_chars[1:]:
        # 如果Y坐标差小于容差，视为同一行
        if abs(c["top"] - current_line[-1]["top"]) < 5:
            current_line.append(c)
        else:
            lines.append(current_line)
            current_line = [c]
    if current_line:
        lines.append(current_line)

    # 将每行的字符组合成文本
    result = []
    for line in lines:
        text = ""
        prev_x = line[0]["x0"]
        for c in line:
            # 如果字符间距超过容差，插入空格
            if c["x0"] - prev_x > x_tolerance:
                text += " "
            text += c["text"]
            prev_x = c["x1"]
        result.append({
            "text": text,
            "y": line[0]["top"],
            "x0": line[0]["x0"],
            "x1": line[-1]["x1"],
            "size": max(c["size"] for c in line),
            "font": line[0]["fontname"],
        })

    return result

with pdfplumber.open("document.pdf") as pdf:
    chars = pdf.pages[0].chars
    words = group_chars_into_words(chars)
    for w in words[:20]:
        print(f"[{w['y']:6.1f}] {w['text']}")

六、数据清洗与结构化

提取数据转pandas DataFrame

PDF 数据提取的最终目标通常是将数据导入到结构化的数据分析工具中进行进一步处理。pandas DataFrame 是 Python 数据科学生态中的核心数据结构。将 pdfplumber 提取的表格数据转换为 DataFrame 后，就可以利用 pandas 强大的数据操作功能进行清洗、转换、分析和可视化。转换过程通常包括指定列名（可能来自表格的第一行）、处理数据类型（将字符串转换为数值类型）、处理缺失值等步骤。

import pdfplumber
import pandas as pd

with pdfplumber.open("financial_report.pdf") as pdf:
    page = pdf.pages[0]
    tables = page.find_tables()

    if tables:
        # 提取表格原始数据
        raw_data = tables[0].extract()

        # 第一行为列名
        headers = raw_data[0]
        data_rows = raw_data[1:]

        # 创建 DataFrame
        df = pd.DataFrame(data_rows, columns=headers)

        print("数据框信息:")
        print(df.info())
        print("\n前5行:")
        print(df.head())

        # 保存为 CSV
        df.to_csv("financial_data.csv", index=False, encoding="utf-8-sig")
        print("数据已保存到 financial_data.csv")

清洗规则

从PDF提取的数据往往存在各种质量问题，需要制定系统的清洗规则。常见的清洗操作包括：去除金额字段中的货币符号和千分位逗号以便转换为数值类型、处理中文和英文标点符号的混用、去除首尾空白字符、处理换行符和多余空格、填充或删除缺失值、统一日期格式、纠正OCR识别错误等。建议将清洗规则整理为可配置的清洗管道（pipeline），便于在不同文档之间复用。

import pandas as pd
import re

class PDFDataCleaner:
    """PDF提取数据清洗管道"""

    @staticmethod
    def remove_whitespace(df):
        """去除所有字符串列的前后空白"""
        for col in df.select_dtypes(include=["object"]).columns:
            df[col] = df[col].str.strip().str.replace(r"\s+", " ", regex=True)
        return df

    @staticmethod
    def clean_numeric(df, columns):
        """将带有货币符号和逗号的字符串转换为数值"""
        for col in columns:
            if col in df.columns:
                df[col] = df[col].astype("str")
                df[col] = df[col].str.replace(r"[¥$€,，]", "", regex=True)
                df[col] = df[col].str.replace(r"[^\d.]", "", regex=True)
                df[col] = pd.to_numeric(df[col], errors="coerce")
        return df

    @staticmethod
    def clean_date(df, columns, fmt="%Y-%m-%d"):
        """统一日期格式"""
        for col in columns:
            if col in df.columns:
                # 常见日期格式转换
                df[col] = df[col].astype("str")
                df[col] = df[col].str.replace(r"(\d{4})[年](\d{1,2})[月](\d{1,2})[日]",
                                              r"\1-\2-\3", regex=True)
                df[col] = pd.to_datetime(df[col], errors="coerce")
        return df

    @staticmethod
    def remove_empty_rows(df, thresh=1):
        """删除空值过多的行"""
        return df.dropna(thresh=thresh)

    @classmethod
    def clean(cls, df, numeric_cols=None, date_cols=None):
        """运行完整清洗管道"""
        df = cls.remove_whitespace(df)
        if numeric_cols:
            df = cls.clean_numeric(df, numeric_cols)
        if date_cols:
            df = cls.clean_date(df, date_cols)
        df = cls.remove_empty_rows(df)
        return df

# 使用清洗管道
df = pd.read_csv("raw_data.csv")
cleaner = PDFDataCleaner()
df_cleaned = cleaner.clean(
    df,
    numeric_cols=["金额", "单价", "数量"],
    date_cols=["日期", "到期日"]
)
print(df_cleaned.dtypes)
print(df_cleaned.head())

字段映射与导出

在将 PDF 数据导入到目标系统（如数据库、ERP系统、数据分析平台）之前，通常需要进行字段映射。源数据中的列名可能与目标系统的字段名不一致，需要建立映射关系。此外，还需要进行数据验证（检查必填字段、数据范围、格式正确性）和异常处理。完成清洗和映射后，可以将数据导出为多种格式：Excel（支持多工作表）、CSV（通用交换格式）、直接写入数据库等。

import pandas as pd
from pathlib import Path

def export_pdf_data(pdf_path, output_dir="."):
    """完整的PDF提取、清洗和导出流程"""
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    # 字段映射
    field_mapping = {
        "商品名称": "product_name",
        "规格型号": "specification",
        "单位": "unit",
        "数量": "quantity",
        "含税单价": "unit_price",
        "含税金额": "total_amount",
        "税率": "tax_rate",
        "税额": "tax_amount",
    }

    with pdfplumber.open(pdf_path) as pdf:
        all_data = []
        for page in pdf.pages:
            tables = page.find_tables()
            for table in tables:
                raw = table.extract()
                if len(raw) > 1:
                    # 使用第一行作为列名
                    headers = raw[0]
                    # 映射列名
                    mapped_headers = [field_mapping.get(h, h) for h in headers]
                    df_page = pd.DataFrame(raw[1:], columns=mapped_headers)
                    all_data.append(df_page)

        if not all_data:
            print("未提取到数据")
            return

        # 合并所有页面的数据
        df = pd.concat(all_data, ignore_index=True)

        # 清洗数据
        df = PDFDataCleaner.clean(
            df,
            numeric_cols=["quantity", "unit_price", "total_amount",
                          "tax_rate", "tax_amount"]
        )

        # 导出为Excel（带格式）
        excel_path = output_dir / f"{Path(pdf_path).stem}_output.xlsx"
        with pd.ExcelWriter(excel_path, engine="openpyxl") as writer:
            df.to_excel(writer, sheet_name="PDF提取数据", index=False)
            # 自动调整列宽
            worksheet = writer.sheets["PDF提取数据"]
            for col in worksheet.columns:
                max_len = max(len(str(cell.value or "")) for cell in col)
                worksheet.column_dimensions[col[0].column_letter].width = min(max_len + 2, 40)

        # 同时导出CSV
        csv_path = output_dir / f"{Path(pdf_path).stem}_output.csv"
        df.to_csv(csv_path, index=False, encoding="utf-8-sig")

        print(f"成功导出 {len(df)} 行数据")
        print(f"Excel: {excel_path}")
        print(f"CSV: {csv_path}")

        return df

# 使用示例
df = export_pdf_data("invoice_data.pdf", "output")

七、批量PDF处理

批量提取与文件夹遍历

在真实业务场景中，我们往往需要一次性处理大量 PDF 文件（如批量处理发票、月度报表、合同文档等）。Python 的 pathlib 库提供了便捷的文件系统操作接口，可以遍历指定目录及其子目录下的所有 PDF 文件。结合 pdfplumber 的页面迭代能力，可以实现全自动的批量处理管道。关键在于设计一个鲁棒的文件处理循环，能够优雅地处理单个文件失败的情况（如密码保护的文件、损坏的PDF），而不中断整个批处理流程。

from pathlib import Path
import pdfplumber
import pandas as pd
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("pdf_extract.log", encoding="utf-8"),
        logging.StreamHandler()
    ]
)

def find_all_pdfs(root_dir):
    """递归查找目录下所有PDF文件"""
    return sorted(Path(root_dir).rglob("*.pdf"))

def batch_extract_text(pdf_files, output_dir="extracted_text"):
    """批量提取PDF文本到单独的txt文件"""
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    results = {"success": 0, "failed": 0, "errors": []}

    for pdf_file in pdf_files:
        try:
            logging.info(f"正在处理: {pdf_file.name}")

            with pdfplumber.open(str(pdf_file)) as pdf:
                text_content = []
                for i, page in enumerate(pdf.pages):
                    text = page.extract_text()
                    if text:
                        text_content.append(f"--- 第{i+1}页 ---\n{text}")

            # 保存提取的文本
            output_file = output_path / f"{pdf_file.stem}.txt"
            with open(output_file, "w", encoding="utf-8") as f:
                f.write("\n\n".join(text_content))

            results["success"] += 1
            logging.info(f"  → 已保存: {output_file.name}")

        except Exception as e:
            results["failed"] += 1
            results["errors"].append({"file": str(pdf_file.name), "error": str(e)})
            logging.error(f"  → 失败: {e}")

    # 输出汇总报告
    logging.info("=" * 40)
    logging.info(f"批量处理完成: 成功{results['success']}个, 失败{results['failed']}个")
    if results["errors"]:
        logging.info("失败文件列表:")
        for err in results["errors"]:
            logging.info(f"  - {err['file']}: {err['error']}")

    return results

# 使用示例
# pdf_files = find_all_pdfs("./invoices/")
# results = batch_extract_text(pdf_files, "./extracted_text/")

并发加速处理

当需要处理大量 PDF 文件时（数百甚至数千个），单线程串行处理可能非常耗时。Python 的 concurrent.futures 模块可以方便地实现多线程或多进程并发处理。对于 IO 密集型任务（如读取 PDF 文件），多线程效果显著；对于 CPU 密集型任务（如 OCR 识别），多进程更为适合。需要注意的是，并发处理时需要确保输出文件的命名不会冲突，并且要合理控制并发数量以避免耗尽系统资源。

from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
import pdfplumber
import pandas as pd
import time

def extract_table_from_pdf(pdf_path):
    """从单个PDF中提取表格数据"""
    try:
        with pdfplumber.open(str(pdf_path)) as pdf:
            all_rows = []
            for page in pdf.pages:
                tables = page.find_tables()
                for table in tables:
                    for row in table.extract():
                        all_rows.append(row)
            return {"file": pdf_path.name, "rows": all_rows, "success": True}
    except Exception as e:
        return {"file": pdf_path.name, "error": str(e), "success": False}

def parallel_extract(pdf_dir, max_workers=4):
    """多线程并发提取PDF表格"""
    pdf_files = list(Path(pdf_dir).glob("*.pdf"))
    print(f"找到 {len(pdf_files)} 个PDF文件，使用 {max_workers} 个线程并发处理...")

    start_time = time.time()
    all_data = []
    success_count = 0

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_file = {executor.submit(extract_table_from_pdf, f): f for f in pdf_files}

        # 处理完成的Future
        for i, future in enumerate(as_completed(future_to_file), 1):
            result = future.result()
            if result["success"]:
                all_data.extend(result["rows"])
                success_count += 1
            else:
                print(f"  [{i}/{len(pdf_files)}] 失败: {result['file']} - {result['error']}")

            # 进度提示
            if i % 10 == 0 or i == len(pdf_files):
                elapsed = time.time() - start_time
                print(f"  进度: {i}/{len(pdf_files)} ({i*100//len(pdf_files)}%), 用时{elapsed:.1f}秒")

    elapsed = time.time() - start_time
    print(f"\n并发提取完成!")
    print(f"成功: {success_count}/{len(pdf_files)}, 总行数: {len(all_data)}, 用时: {elapsed:.1f}秒")

    return all_data

# 使用示例
# data = parallel_extract("./invoices/", max_workers=8)

进度监控与错误处理

在批量处理大量文件时，实时的进度反馈和完善的错误处理机制至关重要。除了使用日志记录外，还可以使用 tqdm 库显示进度条、记录处理统计信息（处理总数、成功数、失败数、平均处理时间）、生成处理报告。对于处理失败的文件，应该记录详细的错误信息（文件路径、错误类型、堆栈跟踪），以便后续排查和重新处理。建议将处理失败的文件列表保存到单独的文件中，支持断点续传或重试机制。

from tqdm import tqdm
import json
from datetime import datetime
from pathlib import Path
import traceback

def batch_process_with_report(pdf_dir, output_dir, report_file="batch_report.json"):
    """带进度监控和详细报告的批量处理"""
    pdf_files = list(Path(pdf_dir).glob("*.pdf"))
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    stats = {
        "start_time": datetime.now().isoformat(),
        "total": len(pdf_files),
        "success": 0,
        "failed": 0,
        "skipped": 0,
        "details": []
    }

    # 使用 tqdm 显示进度条
    for pdf_file in tqdm(pdf_files, desc="处理PDF", unit="个"):
        detail = {"file": pdf_file.name, "status": "unknown"}
        try:
            # 跳过空文件
            if pdf_file.stat().st_size == 0:
                detail["status"] = "skipped"
                detail["reason"] = "空文件"
                stats["skipped"] += 1
                stats["details"].append(detail)
                continue

            with pdfplumber.open(str(pdf_file)) as pdf:
                num_pages = len(pdf.pages)
                total_chars = sum(len(page.chars) for page in pdf.pages)
                total_tables = sum(len(page.find_tables()) for page in pdf.pages)

                # 提取文本并保存
                text = ""
                for page in pdf.pages:
                    page_text = page.extract_text()
                    if page_text:
                        text += page_text + "\n\n"

                # 保存提取结果
                txt_path = output_path / f"{pdf_file.stem}.txt"
                with open(txt_path, "w", encoding="utf-8") as f:
                    f.write(text)

            detail["status"] = "success"
            detail["pages"] = num_pages
            detail["chars"] = total_chars
            detail["tables"] = total_tables
            detail["text_length"] = len(text)
            stats["success"] += 1

        except pdfplumber.pdfminer.pdfparser.PDFSyntaxError as e:
            detail["status"] = "failed"
            detail["error_type"] = "PDF语法错误"
            detail["error"] = str(e)
            stats["failed"] += 1

        except Exception as e:
            detail["status"] = "failed"
            detail["error_type"] = type(e).__name__
            detail["error"] = str(e)
            stats["failed"] += 1

        stats["details"].append(detail)

    # 完成统计
    stats["end_time"] = datetime.now().isoformat()
    stats["success_rate"] = round(stats["success"] / stats["total"] * 100, 1) if stats["total"] > 0 else 0

    # 保存报告
    with open(output_path / report_file, "w", encoding="utf-8") as f:
        json.dump(stats, f, ensure_ascii=False, indent=2)

    print("\n处理完成!")
    print(f"总计: {stats['total']}, 成功: {stats['success']}, 失败: {stats['failed']}, 跳过: {stats['skipped']}")
    print(f"成功率: {stats['success_rate']}%")
    print(f"报告已保存: {output_path / report_file}")

    return stats

八、OCR配合使用

pdfplumber + camelot 组合

pdfplumber 在文本提取和表格检测方面表现出色，但在处理扫描件或图像型PDF时，它无法直接提取文本（因为没有可访问的文本层）。此时，可以结合 camelot 库来处理表格数据。camelot 专为表格提取设计，提供了 Lattice（基于线条）和 Stream（基于文本流）两种模式。camelot 的内部使用 Ghostscript 将 PDF 页面转换为图像，然后应用 OpenCV 检测表格线。不过，camelot 对中文表格的支持有限，在处理复杂中文表格时可能需要降级到 pdfplumber 的基于字符的提取策略。

推荐的工作流是：先用 pdfplumber 尝试提取，如果提取结果为空或质量很差，则降级使用 camelot 或 OCR 方案。这种多级降级策略可以最大化提取的成功率和质量。

# pdfplumber + camelot 组合使用示例
import pdfplumber
try:
    import camelot
    HAS_CAMELOT = True
except ImportError:
    HAS_CAMELOT = False

def extract_tables_hybrid(pdf_path):
    """多引擎表格提取：优先pdfplumber，降级到camelot"""
    results = []

    # 尝试 pdfplumber
    try:
        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                tables = page.find_tables()
                for table in tables:
                    data = table.extract()
                    if data and len(data) > 1:
                        results.append({
                            "source": "pdfplumber",
                            "page": page.page_number,
                            "data": data
                        })

        if results:
            print(f"pdfplumber成功提取 {len(results)} 个表格")
            return results
    except Exception as e:
        print(f"pdfplumber提取失败: {e}")

    # 降级到 camelot
    if HAS_CAMELOT:
        try:
            print("降级到 camelot 提取...")
            for flavor in ["lattice", "stream"]:
                try:
                    tables = camelot.read_pdf(pdf_path, flavor=flavor, pages="all")
                    if tables.n > 0:
                        for table in tables:
                            results.append({
                                "source": f"camelot-{flavor}",
                                "page": table.page,
                                "data": table.data
                            })
                        print(f"camelot({flavor})成功提取 {tables.n} 个表格")
                        break
                except Exception as e2:
                    print(f"camelot({flavor})失败: {e2}")
        except Exception as e:
            print(f"camelot整体失败: {e}")

    return results

pytesseract OCR 识别

对于扫描文档或图片型PDF，OCR（光学字符识别）是提取文本的唯一途径。pytesseract 是 Google Tesseract OCR 引擎的 Python 封装，支持包括中文在内的多种语言。配合 pdfplumber 的 to_image() 方法，可以将 PDF 页面渲染为高分辨率图像，然后使用 pytesseract 进行文字识别。需要注意的是，OCR 识别质量受图像质量、分辨率、语言模型、字体类型等因素影响，通常需要一定的预处理（如二值化、降噪、旋转校正）来提高识别准确率。

import pdfplumber
from PIL import Image
import pytesseract

# 配置 tesseract 路径（Windows环境）
# pytesseract.pytesseract.tesseract_cmd = r"C:\Program Files\Tesseract-OCR\tesseract.exe"

def ocr_pdf_page(page, lang="chi_sim+eng", resolution=300):
    """对PDF页面进行OCR识别"""
    # 将页面渲染为高分辨率图像
    im = page.to_image(resolution=resolution)

    # 转换为PIL Image
    pil_image = im.original.convert("RGB")

    # OCR识别
    # lang: chi_sim(简体中文) + eng(英文)
    # config: 自定义配置
    custom_config = "--oem 3 --psm 6"
    text = pytesseract.image_to_string(
        pil_image,
        lang=lang,
        config=custom_config
    )

    return text

def ocr_pdf_document(pdf_path, lang="chi_sim+eng"):
    """对完整的扫描PDF进行OCR识别"""
    all_text = []

    with pdfplumber.open(pdf_path) as pdf:
        print(f"开始OCR识别: {pdf_path}")
        print(f"共 {len(pdf.pages)} 页，语言: {lang}")

        for i, page in enumerate(pdf.pages):
            print(f"  识别第 {i+1} 页...")
            text = ocr_pdf_page(page, lang=lang)

            page_text = f"=== 第{i+1}页 ===\n{text}"
            all_text.append(page_text)

    full_text = "\n\n".join(all_text)

    # 保存识别结果
    output_path = Path(pdf_path).with_suffix(".ocr.txt")
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(full_text)

    print(f"OCR识别完成! 结果已保存到: {output_path}")
    return full_text

# 使用示例
# text = ocr_pdf_document("scanned_doc.pdf", lang="chi_sim+eng")

扫描件处理与中文识别

处理中文扫描件时，Tesseract 的中文识别能力已经相当不错，但仍有局限性。为提高中文OCR的准确率，建议：使用至少300DPI的分辨率渲染页面；在OCR前对图像进行预处理（灰度化、二值化、降噪）；设置正确的语言包（chi_sim 简体中文或 chi_tra 繁体中文）；使用适当的 PSM（页面分割模式，如 --psm 6 表示统一的文本块）。对于包含表格的中文扫描件，可以先OCR识别全文，然后使用正则表达式或规则引擎从识别的文本中重构表格结构。

import pdfplumber
from PIL import Image, ImageFilter, ImageEnhance
import pytesseract
import numpy as np
from pathlib import Path

class ChineseOCRPipeline:
    """中文扫描件OCR处理管道"""

    def __init__(self, lang="chi_sim+eng", resolution=300):
        self.lang = lang
        self.resolution = resolution

    def preprocess_image(self, pil_image):
        """图像预处理：提高OCR准确率"""
        # 转换为灰度
        img = pil_image.convert("L")

        # 增强对比度
        enhancer = ImageEnhance.Contrast(img)
        img = enhancer.enhance(2.0)

        # 二值化（阈值处理）
        img_array = np.array(img)
        threshold = 128
        img_array = np.where(img_array > threshold, 255, 0).astype(np.uint8)
        img = Image.fromarray(img_array)

        # 降噪
        img = img.filter(ImageFilter.MedianFilter(size=3))

        return img

    def ocr_with_position(self, page):
        """带位置信息的OCR识别"""
        im = page.to_image(resolution=self.resolution)
        pil_image = im.original.convert("RGB")
        processed = self.preprocess_image(pil_image)

        # 使用TSD（Tesseract Page Segmentation Mode）获取位置数据
        custom_config = f"--oem 3 --psm 6 -c tessedit_create_tsv=1"
        ocr_data = pytesseract.image_to_data(
            processed,
            lang=self.lang,
            config=custom_config,
            output_type=pytesseract.Output.DICT
        )

        # 组织带位置的结果
        results = []
        for i in range(len(ocr_data["text"])):
            text = ocr_data["text"][i].strip()
            conf = int(ocr_data["conf"][i])
            if text and conf > 30:  # 只保留置信度>30的识别结果
                results.append({
                    "text": text,
                    "x": ocr_data["left"][i],
                    "y": ocr_data["top"][i],
                    "w": ocr_data["width"][i],
                    "h": ocr_data["height"][i],
                    "conf": conf,
                })

        return results

    def process_document(self, pdf_path):
        """完整处理一个扫描文档"""
        print(f"处理文档: {pdf_path}")
        all_results = []

        with pdfplumber.open(pdf_path) as pdf:
            for i, page in enumerate(pdf.pages):
                print(f"  OCR第{i+1}页...")
                page_results = self.ocr_with_position(page)
                all_results.append({"page": i+1, "items": page_results})

        # 保存JSON格式结果
        output_path = Path(pdf_path).with_suffix(".ocr.json")
        import json
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(all_results, f, ensure_ascii=False, indent=2)

        total_items = sum(len(p["items"]) for p in all_results)
        print(f"完成! 共识别 {total_items} 个文本块，已保存到 {output_path}")
        return all_results

# 使用示例
# pipeline = ChineseOCRPipeline(lang="chi_sim+eng", resolution=300)
# results = pipeline.process_document("scanned_chinese_doc.pdf")

九、实战案例

案例一：发票信息提取

电子发票（PDF格式）是日常办公中最常见的PDF类型之一。每张发票包含固定的结构化信息：发票号码、开票日期、购买方信息、销售方信息、商品明细、金额和税额等。利用 pdfplumber 的区域提取能力，可以从发票PDF的固定位置精准提取所需字段。本案例展示了如何从增值税电子普通发票中提取关键字段，并输出为结构化数据。

import pdfplumber
import re
import json

def extract_invoice_info(pdf_path):
    """从PDF发票中提取关键信息"""
    info = {}

    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[0]
        text = page.extract_text()

        # 提取发票号码（通常格式如：12345678）
        invoice_no_match = re.search(r"发票号码[：:]\s*(\d+)", text)
        if invoice_no_match:
            info["发票号码"] = invoice_no_match.group(1)

        # 提取开票日期
        date_match = re.search(r"开票日期[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日)", text)
        if date_match:
            info["开票日期"] = date_match.group(1)

        # 提取购买方信息
        buyer_name_match = re.search(r"购买方[名称]*[：:]\s*([^\n]+)", text)
        if buyer_name_match:
            info["购买方名称"] = buyer_name_match.group(1).strip()

        # 提取金额
        amount_match = re.search(r"价税合计[（(]大写[）)]\s*[^0-9]*?([\d,]+\.\d{2})", text)
        if not amount_match:
            amount_match = re.search(r"价税合计[：:]\s*([\d,]+\.\d{2})", text)
        if amount_match:
            info["价税合计"] = amount_match.group(1)

        # 提取商品明细（表格区域）
        tables = page.find_tables()
        if tables:
            # 通常第一个表格包含商品明细
            table_data = tables[0].extract()
            if len(table_data) > 1:
                items = []
                for row in table_data[1:]:
                    if row[0] and row[0].strip():
                        items.append({
                            "名称": row[0].strip(),
                            "数量": row[2].strip() if len(row) > 2 else "",
                            "金额": row[4].strip() if len(row) > 4 else "",
                        })
                info["商品明细"] = items

    return info

# 使用示例
# invoice = extract_invoice_info("invoice_2025.pdf")
# print(json.dumps(invoice, ensure_ascii=False, indent=2))

案例二：银行对账单解析

银行对账单是另一种常见的结构化PDF文档，通常包含多页的表格数据（交易日期、摘要、交易金额、余额等）。对账单的难点在于：跨页表格的处理（表头在每页重复出现），金额的数字格式（负数和正数的表示方式，千分位分隔符），日期的多种格式。本案例展示了如何构建一个健壮的对账单解析器，将 PDF 格式的银行流水转换为 Excel 格式的结构化数据。

import pdfplumber
import pandas as pd
import re

def parse_bank_statement(pdf_path):
    """解析银行对账单PDF"""
    all_transactions = []
    header = None

    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            tables = page.find_tables({
                "vertical_strategy": "lines",
                "horizontal_strategy": "lines"
            })

            for table in tables:
                data = table.extract()
                if not data or len(data) < 2:
                    continue

                if header is None:
                    header = data[0]
                    transactions = data[1:]
                else:
                    # 检查并跳过重复表头
                    if data[0] == header:
                        transactions = data[1:]
                    else:
                        transactions = data

                for row in transactions:
                    # 清理每行数据
                    cleaned = [cell.strip() if cell else "" for cell in row]
                    # 跳过空行和汇总行
                    if cleaned[0] and not any(kw in cleaned[0]
                       for kw in ["合计", "上期", "本期", "余额"]):
                        all_transactions.append(cleaned)

    # 转换为 DataFrame
    if header and all_transactions:
        df = pd.DataFrame(all_transactions, columns=header)

        # 清理金额列：去除货币符号和逗号
        amount_cols = [col for col in df.columns if "金额" in col or "余额" in col]
        for col in amount_cols:
            df[col] = df[col].str.replace(r"[¥$,\s]", "", regex=True)
            df[col] = pd.to_numeric(df[col], errors="coerce")

        # 日期列转换
        date_cols = [col for col in df.columns if "日期" in col]
        for col in date_cols:
            df[col] = pd.to_datetime(df[col], errors="coerce")

        return df

    return pd.DataFrame()

# 使用示例
# df = parse_bank_statement("bank_statement_2025Q1.pdf")
# df.to_excel("bank_statement_cleaned.xlsx", index=False)

案例三：论文PDF数据提取

学术论文PDF通常具有高度结构化的格式：标题、作者、摘要、关键词、章节标题、正文、参考文献、附录等。利用 pdfplumber 的字体和字号分析功能，可以自动识别论文的各个结构元素。例如，最大字号的文本通常是论文标题，次大字号的文本是章节标题，特定字体的文本可能是代码块或引用。结合位置信息，还可以提取图表标题、表格数据和参考文献列表。这对于文献综述、元分析等研究工作中批量提取论文信息非常有价值。

import pdfplumber
import re

def extract_paper_structure(pdf_path):
    """提取学术论文的结构化信息"""
    paper = {"title": "", "abstract": "", "sections": [], "references": []}

    with pdfplumber.open(pdf_path) as pdf:
        # 分析第一页识别标题和摘要
        first_page = pdf.pages[0]
        chars = first_page.chars

        # 按字号排序，最大的可能是标题
        unique_sizes = sorted(set(c["size"] for c in chars), reverse=True)
        if len(unique_sizes) >= 3:
            title_size = unique_sizes[0]

            # 提取标题（最大字号的连续文本）
            title_chars = [c for c in chars if abs(c["size"] - title_size) < 0.5]
            if title_chars:
                # 按行分组
                lines = {}
                for c in title_chars:
                    y_key = round(c["top"], 0)
                    if y_key not in lines:
                        lines[y_key] = []
                    lines[y_key].append(c)

                title_parts = []
                for y in sorted(lines.keys()):
                    line_chars = sorted(lines[y], key=lambda x: x["x0"])
                    line_text = "".join(c["text"] for c in line_chars)
                    title_parts.append(line_text)
                paper["title"] = " ".join(title_parts)

        # 提取摘要（"Abstract"或"摘要"后面到下一个章节之前的内容）
        text = first_page.extract_text()
        abstract_match = re.search(
            r"(?:Abstract|ABSTRACT|摘要)[：:.\s]*([^.]*(?:\.(?!\s*(?:Introduction|INTRODUCTION|引言|1\.\s))[^.]*)*)",
            text, re.DOTALL
        )
        if abstract_match:
            paper["abstract"] = abstract_match.group(1).strip()

        # 遍历所有页面提取章节
        current_section = ""
        current_content = []
        section_pattern = re.compile(r"^(\d+\.?\s*[A-Za-z一-鿿][^。\n]*)")

        for page in pdf.pages:
            page_text = page.extract_text()
            if not page_text:
                continue

            for line in page_text.split("\n"):
                line = line.strip()
                if not line:
                    continue

                match = section_pattern.match(line)
                if match:
                    if current_section and current_content:
                        paper["sections"].append({
                            "heading": current_section,
                            "content": "\n".join(current_content)
                        })
                    current_section = match.group(1)
                    current_content = [line[len(match.group(1)):].strip()]
                else:
                    current_content.append(line)

        # 不要忘记最后一个章节
        if current_section and current_content:
            paper["sections"].append({
                "heading": current_section,
                "content": "\n".join(current_content)
            })

        # 提取参考文献（通常在最后，以[1]或[1]开头）
        last_page = pdf.pages[-1]
        last_text = last_page.extract_text()
        ref_start = last_text.find("References")
        if ref_start == -1:
            ref_start = last_text.find("参考文献")
        if ref_start != -1:
            ref_text = last_text[ref_start:]
            paper["references"] = [r.strip() for r in ref_text.split("\n")
                                                     if r.strip() and
                                                     re.match(r"^\s*\[?\d+[\]\.,]", r.strip())]

    return paper

# 使用示例
# paper = extract_paper_structure("research_paper.pdf")
# print(f"标题: {paper['title']}")
# print(f"摘要: {paper['abstract'][:200]}...")
# print(f"章节: {[s['heading'] for s in paper['sections']]}")

案例四：年报数据分析

上市公司年报PDF通常篇幅巨大（数百页），包含了大量财务表格和结构化数据。年报的表格形式多样（合并利润表、资产负债表、现金流量表、附注表格等），且同一公司不同年份的表格格式也可能有差异。本案例展示了如何从年报PDF中提取特定的财务表格（如利润表），并进行时间序列对比分析。年报提取的难点在于：超大文件的内存管理、复杂表头的解析、财务数字的精度保留、不同会计准则下的报表差异。

import pdfplumber
import pandas as pd
from pathlib import Path

def extract_financial_table(pdf_path, table_keywords, page_range=None):
    """从年报PDF中提取指定财务表格"""
    target_tables = []

    with pdfplumber.open(pdf_path) as pdf:
        pages_to_scan = pdf.pages[page_range[0]:page_range[1]] if page_range else pdf.pages

        for page in pages_to_scan:
            tables = page.find_tables()

            for table in tables:
                data = table.extract()
                if not data or len(data) < 3:
                    continue

                # 检查表头是否包含目标关键字
                header_text = " ".join(str(cell) for cell in data[0] if cell)
                if any(kw in header_text for kw in table_keywords):
                    print(f"在第{page.page_number}页找到目标表格")
                    target_tables.append({
                        "page": page.page_number,
                        "data": data
                    })

    return target_tables

def clean_financial_data(raw_data):
    """清洗财务表格数据"""
    # 识别表头和单位行
    start_row = 0
    for i, row in enumerate(raw_data):
        row_text = "|".join(str(c) for c in row if c)
        if "单位" in row_text or "项目" in row_text:
            start_row = i
            break

    headers = raw_data[start_row]
    rows = raw_data[start_row + 1:]

    df = pd.DataFrame(rows, columns=headers)

    # 清理数字列：去除空格、逗号，转换为数值
    for col in df.columns[1:]:  # 第一列通常是项目名称
        df[col] = df[col].astype("str")
        df[col] = df[col].str.replace(r"[,\s]", "", regex=True)
        df[col] = pd.to_numeric(df[col], errors="coerce")

    return df

def compare_yearly_reports(pdf_files):
    """对比多个年份的财报数据"""
    all_data = {}

    for pdf_file in pdf_files:
        # 从文件名中提取年份
        year = Path(pdf_file).stem[-4:]
        print(f"处理 {year} 年年报...")

        tables = extract_financial_table(
            pdf_file,
            table_keywords=["利润表", "income", "综合收益"],
            page_range=(30, 60)  # 利润表通常在30-60页之间
        )

        if tables:
            df = clean_financial_data(tables[0]["data"])
            all_data[year] = df

    # 合并对比：关注关键指标
    key_metrics = ["营业收入", "营业利润", "利润总额", "净利润"]

    for year, df in all_data.items():
        print(f"\n=== {year}年关键指标 ===")
        first_col = df.columns[0]
        for metric in key_metrics:
            match = df[df[first_col].astype("str").str.contains(metric, na=False)]
            if not match.empty:
                print(f"  {metric}: {match.iloc[0, 1]}")

    return all_data

# 使用示例
# pdf_files = ["annual_report_2022.pdf", "annual_report_2023.pdf", "annual_report_2024.pdf"]
# results = compare_yearly_reports(pdf_files)