零基础上手使用Python pdfplumber提取原生PDF表格

import PyPDF2
reader = PyPDF2.PdfReader('invoice.pdf')
text = reader.pages[0].extract_text()
print(text)

import pdfplumber
with pdfplumber.open('invoice.pdf') as pdf:
    table = pdf.pages[0].extract_table()
    for row in table:
        print(row)

pip install pdfplumber

python -c "import pdfplumber; print(pdfplumber.__version__)"

# 数据处理
pip install pandas
# Excel导出
pip install openpyxl
# 可视化（Jupyter环境）
pip install matplotlib

# 创建虚拟环境
python -m venv pdf_env
# 激活（Windows）
pdf_env\Scripts\activate
# 激活（Mac/Linux）
source pdf_env/bin/activate
# 安装依赖
pip install pdfplumber pandas openpyxl
# 导出依赖清单
pip freeze > requirements.txt

import pdfplumber

# 使用with语句自动管理资源
with pdfplumber.open("example.pdf") as pdf:
    print(f"总页数: {len(pdf.pages)}")
    
    # 获取第一页
    first_page = pdf.pages[0]
    print(f"页面尺寸: {first_page.width} x {first_page.height}")

import pdfplumber

with pdfplumber.open("report.pdf") as pdf:
    # 提取第一页的表格
    page = pdf.pages[0]
    table = page.extract_table()
    
    if table:
        # 打印表头
        print("表头:", table[0])
        # 打印数据行
        for row in table[1:]:
            print(row)

import pdfplumber

all_tables = []

with pdfplumber.open("multi_page_report.pdf") as pdf:
    for page_num, page in enumerate(pdf.pages, 1):
        tables = page.extract_tables()
        if tables:
            print(f"第{page_num}页发现 {len(tables)} 个表格")
            all_tables.extend(tables)

print(f"共提取 {len(all_tables)} 个表格")

import csv
import pdfplumber

def table_to_csv(pdf_path, output_csv, page_num=0):
    with pdfplumber.open(pdf_path) as pdf:
        table = pdf.pages[page_num].extract_table()
        
        if table:
            with open(output_csv, 'w', newline='', encoding='utf-8') as f:
                writer = csv.writer(f)
                writer.writerows(table)
            print(f"已保存至: {output_csv}")
        else:
            print("未检测到表格")

# 使用
table_to_csv("report.pdf", "output.csv")

import pdfplumber
from openpyxl import Workbook

def table_to_excel(pdf_path, excel_path, page_num=0):
    wb = Workbook()
    ws = wb.active
    ws.title = "PDF表格数据"
    
    with pdfplumber.open(pdf_path) as pdf:
        tables = pdf.pages[page_num].extract_tables()
        
        row_offset = 0
        for table_idx, table in enumerate(tables):
            if table_idx > 0:
                # 表格之间加空行
                row_offset += 1
            
            for i, row in enumerate(table):
                for j, cell in enumerate(row):
                    ws.cell(row=row_offset + i + 1, column=j + 1, value=cell)
            
            row_offset += len(table)
        
        wb.save(excel_path)
        print(f"已保存至: {excel_path}")

table_to_excel("report.pdf", "output.xlsx")

import pandas as pd
import pdfplumber

def table_to_dataframe(pdf_path, page_num=0):
    with pdfplumber.open(pdf_path) as pdf:
        table = pdf.pages[page_num].extract_table()
        
        if table:
            # 第一行作为列名
            df = pd.DataFrame(table[1:], columns=table[0])
            return df
        else:
            return pd.DataFrame()

# 使用
df = table_to_dataframe("report.pdf")
print(df.head())
print(f"数据维度: {df.shape}")

# 导出为Excel
df.to_excel("output.xlsx", index=False)

import os
import pandas as pd
import pdfplumber
from pathlib import Path

def batch_extract_tables(input_folder, output_folder):
    """批量提取文件夹内所有PDF的表格，每个PDF单独保存为一个Excel"""
    Path(output_folder).mkdir(parents=True, exist_ok=True)
    
    pdf_files = list(Path(input_folder).glob("*.pdf")) + list(Path(input_folder).glob("*.PDF"))
    
    for pdf_file in pdf_files:
        print(f"处理: {pdf_file.name}")
        all_tables = []
        
        with pdfplumber.open(pdf_file) as pdf:
            for page_num, page in enumerate(pdf.pages, 1):
                tables = page.extract_tables()
                for table_idx, table in enumerate(tables):
                    if table and len(table) > 1:  # 至少包含表头+一行数据
                        df = pd.DataFrame(table[1:], columns=table[0])
                        all_tables.append((page_num, table_idx + 1, df))
        
        # 保存到Excel，每个表格一个sheet
        if all_tables:
            output_path = Path(output_folder) / f"{pdf_file.stem}_tables.xlsx"
            with pd.ExcelWriter(output_path) as writer:
                for page_num, table_idx, df in all_tables:
                    sheet_name = f"P{page_num}_T{table_idx}"
                    # Excel sheet名称不能超过31个字符
                    df.to_excel(writer, sheet_name=sheet_name[:31], index=False)
            print(f"  ✓ 已保存 {len(all_tables)} 个表格至 {output_path.name}")
        else:
            print(f"  ✗ 未检测到有效表格")

batch_extract_tables("./pdfs", "./output")

import pdfplumber

def extract_borderless_table(pdf_path, page_num=0):
    """提取无边框表格"""
    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[page_num]
        
        # 关键：将vertical_strategy设为"text"
        table = page.extract_table({
            "vertical_strategy": "text",    # 根据文本对齐推断竖线
            "horizontal_strategy": "text",  # 根据文本对齐推断横线
            "snap_tolerance": 3,            # 线条对齐容差
        })
        
        return table

# 使用
table = extract_borderless_table("borderless.pdf")
if table:
    for row in table:
        print(row)

def extract_borderless_table_advanced(pdf_path, bbox=None):
    """高级无边框表格提取：先裁剪区域，再使用text策略"""
    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[0]
        
        # 如果指定了边界框，先裁剪
        if bbox:
            page = page.crop(bbox)  # bbox = (x0, y0, x1, y1)
        
        table = page.extract_table({
            "vertical_strategy": "text",
            "horizontal_strategy": "text",
            "snap_tolerance": 5,
            "intersection_tolerance": 5,
        })
        
        return table

# 手动指定表格区域
# 可以通过page.to_image()可视化找到坐标
table = extract_borderless_table_advanced("report.pdf", bbox=(50, 200, 550, 800))

import pandas as pd
import pdfplumber

def extract_multi_page_table(pdf_path, header_rows=1):
    """
    提取跨页表格，自动合并多页数据
    header_rows: 每页开头的表头行数（通常为1）
    """
    all_data = []
    header = None
    
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            table = page.extract_table()
            
            if not table:
                continue
            
            if page_num == 0:
                # 第一页：保存表头
                header = table[:header_rows]
                # 数据行从表头之后开始
                data_rows = table[header_rows:]
            else:
                # 后续页：跳过本页的表头
                data_rows = table[header_rows:] if len(table) > header_rows else []
            
            all_data.extend(data_rows)
    
    # 构建DataFrame
    if header and all_data:
        # 将表头列表展平为列名
        columns = [cell for row in header for cell in row if cell]
        df = pd.DataFrame(all_data, columns=columns[:len(all_data[0])])
        return df
    return pd.DataFrame()

# 使用
df = extract_multi_page_table("long_report.pdf")
df.to_excel("merged_table.xlsx", index=False)

def extract_cross_page_table_advanced(pdf_path):
    """
    高级跨页表格处理：处理行在页面中间被截断的情况
    """
    all_rows = []
    partial_row = None
    
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            table = page.extract_table()
            
            if not table:
                continue
            
            if page_num == 0:
                # 第一页：完整表格
                all_rows.extend(table)
            else:
                # 后续页：需要判断第一行是否为上一页的续行
                # 简化处理：直接追加，实际项目中需根据业务逻辑判断
                first_row = table[0] if table else []
                if partial_row and first_row:
                    # 合并续行（假设续行第一列为空）
                    merged_row = [partial_row[i] + first_row[i] if first_row[i] else partial_row[i] 
                                  for i in range(len(partial_row))]
                    all_rows.append(merged_row)
                    all_rows.extend(table[1:])
                else:
                    all_rows.extend(table)
            
            # 暂存最后一行，供下一页判断
            partial_row = table[-1] if table else None
    
    return all_rows

def clean_table_data(table):
    """
    过滤空行和无效数据
    """
    if not table:
        return []
    
    cleaned = []
    for row in table:
        # 检查行是否全为空
        if not row or all(cell is None or str(cell).strip() == '' for cell in row):
            continue
        
        # 清洗每个单元格
        cleaned_row = []
        for cell in row:
            if cell is None:
                cleaned_row.append('')
            else:
                # 去除首尾空白，替换换行符
                cell_clean = str(cell).strip().replace('\n', ' ')
                cleaned_row.append(cell_clean)
        
        cleaned.append(cleaned_row)
    
    return cleaned

# 使用示例
table = page.extract_table()
clean_table = clean_table_data(table)

import pdfplumber
import pandas as pd
from pathlib import Path

def extract_text_and_tables(pdf_path):
    """
    同时提取PDF中的文本和表格，返回结构化结果
    """
    result = {
        "metadata": {},
        "pages": []
    }
    
    with pdfplumber.open(pdf_path) as pdf:
        # 提取元信息
        result["metadata"] = {
            "total_pages": len(pdf.pages),
            "page_width": pdf.pages[0].width if pdf.pages else 0,
            "page_height": pdf.pages[0].height if pdf.pages else 0,
        }
        
        for page_num, page in enumerate(pdf.pages, 1):
            page_data = {
                "page_number": page_num,
                "text": page.extract_text() or "",
                "tables": []
            }
            
            # 提取表格
            tables = page.extract_tables()
            for table_idx, table in enumerate(tables):
                if table and len(table) > 1:
                    # 转换为DataFrame便于后续处理
                    df = pd.DataFrame(table[1:], columns=table[0])
                    page_data["tables"].append({
                        "table_index": table_idx + 1,
                        "dataframe": df,
                        "raw_table": table
                    })
            
            result["pages"].append(page_data)
    
    return result

# 使用示例
result = extract_text_and_tables("annual_report.pdf")
print(f"总页数: {result['metadata']['total_pages']}")

for page in result["pages"]:
    print(f"\n第{page['page_number']}页:")
    print(f"  文本长度: {len(page['text'])} 字符")
    print(f"  表格数量: {len(page['tables'])}")

def extract_pure_text_excluding_tables(pdf_path, page_num=0):
    """
    提取纯文本：排除表格区域内的内容
    """
    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[page_num]
        
        # 第一步：提取所有表格的边界框
        table_bboxes = []
        tables = page.find_tables()
        for table in tables:
            # 每个Table对象都有.bbox属性：(x0, y0, x1, y1)
            table_bboxes.append(table.bbox)
        
        # 第二步：获取所有文本对象（带坐标）
        words = page.extract_words()
        
        # 第三步：过滤掉位于表格区域内的文本
        pure_text_words = []
        for word in words:
            x0, top, x1, bottom = word['x0'], word['top'], word['x1'], word['bottom']
            
            is_in_table = False
            for bbox in table_bboxes:
                if (x0 >= bbox[0] and x1 <= bbox[2] and 
                    top >= bbox[1] and bottom <= bbox[3]):
                    is_in_table = True
                    break
            
            if not is_in_table:
                pure_text_words.append(word['text'])
        
        return ' '.join(pure_text_words)

pure_text = extract_pure_text_excluding_tables("report.pdf")
print(pure_text)

# 方案1：使用可视化调试找到问题
with pdfplumber.open("problem.pdf") as pdf:
    page = pdf.pages[0]
    im = page.to_image()
    im.draw_rects(page.extract_words())  # 绘制文本块
    im.save("debug.png")  # 查看图片，确认文本分布
    # 也可以使用debug_tablefinder查看表格检测结果
    page.debug_tablefinder().show()  # 在Jupyter中运行

# 方案2：调整表格检测参数
table = page.extract_table({
    "vertical_strategy": "lines",
    "horizontal_strategy": "lines",
    "snap_tolerance": 5,      # 线条对齐容差
    "join_tolerance": 3,      # 线段连接容差
    "edge_min_length": 3,     # 最小边缘长度
})

# 方案3：针对线条缺失的情况，使用lattice模式
table = page.extract_table({
    "horizontal_strategy": "lines",
    "vertical_strategy": "lines",
    "min_horizontal_line_length": 50,  # 最小横线长度
    "min_vertical_line_length": 20,    # 最小竖线长度
})

# 方案1：检查字体信息
with pdfplumber.open("chinese.pdf") as pdf:
    page = pdf.pages[0]
    # 查看页面的字体列表
    if hasattr(page, 'fonts'):
        for font in page.fonts:
            print(f"字体名: {font.get('name')}, 编码: {font.get('encoding')}")

# 方案2：尝试不同的文本提取参数
text = page.extract_text(laparams={
    "detect_vertical": True,  # 检测竖排文本
})

# 方案3：如果pdfplumber乱码严重，可考虑OCR方案
# 将PDF页面转为图像后用PaddleOCR识别
from pdf2image import convert_from_path
from paddleocr import PaddleOCR

ocr = PaddleOCR(use_angle_cls=True, lang='ch')
images = convert_from_path('chinese.pdf')
for img in images:
    result = ocr.ocr(img, cls=True)
    for line in result:
        print(line[1][0])

def process_large_pdf_in_batches(pdf_path, batch_size=50):
    """
    分批处理大PDF文件，控制内存占用
    """
    with pdfplumber.open(pdf_path) as pdf:
        total_pages = len(pdf.pages)
        all_tables = []
        
        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_num in range(batch_start, batch_end):
                page = pdf.pages[page_num]
                tables = page.extract_tables()
                if tables:
                    all_tables.extend(tables)
            
            # 批次处理后可以添加内存清理逻辑
            # 注意：pdf.pages是惰性加载的，每次访问才加载
            # 实际项目中可考虑将中间结果写入文件而非全量保存
        
        return all_tables

# 使用
all_tables = process_large_pdf_in_batches("large.pdf", batch_size=50)

import pdfplumber
import pandas as pd
import re
from pathlib import Path
from typing import Dict, List, Optional

def extract_invoice_info(pdf_path: str) -> Dict:
    """
    从PDF电子发票中提取关键信息
    
    支持全电发票和传统增值税电子发票
    """
    result = {
        "file_name": Path(pdf_path).name,
        "invoice_code": "",
        "invoice_number": "",
        "invoice_date": "",
        "buyer_name": "",
        "seller_name": "",
        "total_amount": "",
        "total_tax": "",
        "total_amount_with_tax": "",
        "items": [],  # 商品明细列表
        "success": False
    }
    
    try:
        with pdfplumber.open(pdf_path) as pdf:
            page = pdf.pages[0]
            text = page.extract_text() or ""
            
            # 1. 提取发票基本信息（使用正则表达式）
            # 发票代码：通常为10-12位数字
            code_match = re.search(r'发票代码[：:]\s*(\d{10,12})', text)
            if code_match:
                result["invoice_code"] = code_match.group(1)
            
            # 发票号码：通常为8位数字
            number_match = re.search(r'发票号码[：:]\s*(\d{8})', text)
            if number_match:
                result["invoice_number"] = number_match.group(1)
            
            # 开票日期：多种格式
            date_match = re.search(r'开票日期[：:]\s*(\d{4}[年/-]\d{1,2}[月/-]\d{1,2}日?)', text)
            if date_match:
                result["invoice_date"] = date_match.group(1)
            
            # 购买方名称
            buyer_match = re.search(r'购买方[：:]?\s*名称[：:]\s*(.+?)(?:\n|单位|纳税人)', text)
            if not buyer_match:
                buyer_match = re.search(r'购方名称[：:]\s*(.+?)(?:\n|纳税人)', text)
            if buyer_match:
                result["buyer_name"] = buyer_match.group(1).strip()
            
            # 销售方名称
            seller_match = re.search(r'销售方[：:]?\s*名称[：:]\s*(.+?)(?:\n|纳税人)', text)
            if not seller_match:
                seller_match = re.search(r'销方名称[：:]\s*(.+?)(?:\n|纳税人)', text)
            if seller_match:
                result["seller_name"] = seller_match.group(1).strip()
            
            # 金额汇总
            amount_match = re.search(r'价税合计[（(]小写[）)]\s*[￥¥]?\s*(\d+\.?\d*)', text)
            if amount_match:
                result["total_amount_with_tax"] = amount_match.group(1)
            
            # 2. 提取货物清单表格
            tables = page.extract_tables()
            for table in tables:
                if not table or len(table) < 2:
                    continue
                
                # 判断是否为货物清单表格：检查表头是否包含"货物名称"、"金额"等关键词
                header = [str(cell).lower() if cell else "" for cell in table[0]]
                header_str = " ".join(header)
                
                if any(keyword in header_str for keyword in ["货物", "名称", "商品", "服务", "金额", "税率", "税额"]):
                    # 找到表格列索引
                    col_indices = {}
                    for idx, col in enumerate(header):
                        col_lower = col.lower() if col else ""
                        if "货物" in col_lower or "名称" in col_lower or "商品" in col_lower or "服务" in col_lower:
                            col_indices["name"] = idx
                        elif "金额" in col_lower and "税额" not in col_lower:
                            col_indices["amount"] = idx
                        elif "税率" in col_lower:
                            col_indices["tax_rate"] = idx
                        elif "税额" in col_lower:
                            col_indices["tax_amount"] = idx
                        elif "数量" in col_lower:
                            col_indices["quantity"] = idx
                        elif "单价" in col_lower:
                            col_indices["unit_price"] = idx
                    
                    # 提取数据行
                    for row in table[1:]:  # 跳过表头
                        if not row or all(cell is None or str(cell).strip() == "" for cell in row):
                            continue
                        
                        item = {}
                        if "name" in col_indices:
                            item["name"] = str(row[col_indices["name"]]).strip() if row[col_indices["name"]] else ""
                        if "amount" in col_indices:
                            item["amount"] = str(row[col_indices["amount"]]).strip() if row[col_indices["amount"]] else ""
                        if "tax_rate" in col_indices:
                            item["tax_rate"] = str(row[col_indices["tax_rate"]]).strip() if row[col_indices["tax_rate"]] else ""
                        if "tax_amount" in col_indices:
                            item["tax_amount"] = str(row[col_indices["tax_amount"]]).strip() if row[col_indices["tax_amount"]] else ""
                        if "quantity" in col_indices:
                            item["quantity"] = str(row[col_indices["quantity"]]).strip() if row[col_indices["quantity"]] else ""
                        if "unit_price" in col_indices:
                            item["unit_price"] = str(row[col_indices["unit_price"]]).strip() if row[col_indices["unit_price"]] else ""
                        
                        if item.get("name"):
                            result["items"].append(item)
            
            result["success"] = True
            
    except Exception as e:
        result["error"] = str(e)
        print(f"处理失败: {result['file_name']}, 错误: {e}")
    
    return result

def batch_extract_invoices(input_folder: str, output_excel: str):
    """
    批量处理文件夹内的所有发票PDF，汇总输出到Excel
    """
    input_path = Path(input_folder)
    pdf_files = list(input_path.glob("*.pdf")) + list(input_path.glob("*.PDF"))
    
    if not pdf_files:
        print(f"未找到PDF文件: {input_folder}")
        return
    
    all_results = []
    
    for pdf_file in pdf_files:
        print(f"处理: {pdf_file.name}")
        info = extract_invoice_info(str(pdf_file))
        all_results.append(info)
    
    # 转换为DataFrame
    df_main = pd.DataFrame([{
        "文件名": r["file_name"],
        "发票代码": r["invoice_code"],
        "发票号码": r["invoice_number"],
        "开票日期": r["invoice_date"],
        "购买方": r["buyer_name"],
        "销售方": r["seller_name"],
        "价税合计": r["total_amount_with_tax"],
        "是否成功": r["success"],
        "商品数量": len(r.get("items", []))
    } for r in all_results])
    
    # 商品明细DataFrame
    items_list = []
    for r in all_results:
        for item in r.get("items", []):
            items_list.append({
                "发票号码": r["invoice_number"],
                "商品名称": item.get("name", ""),
                "金额": item.get("amount", ""),
                "税率": item.get("tax_rate", ""),
                "税额": item.get("tax_amount", ""),
                "数量": item.get("quantity", ""),
                "单价": item.get("unit_price", "")
            })
    
    df_items = pd.DataFrame(items_list)
    
    # 保存到Excel（多Sheet）
    with pd.ExcelWriter(output_excel) as writer:
        df_main.to_excel(writer, sheet_name="发票汇总", index=False)
        if not df_items.empty:
            df_items.to_excel(writer, sheet_name="商品明细", index=False)
    
    print(f"\n处理完成！共 {len(all_results)} 张发票")
    print(f"成功: {df_main['是否成功'].sum()} 张")
    print(f"输出文件: {output_excel}")

# 使用示例
if __name__ == "__main__":
    # 批量处理发票
    batch_extract_invoices("./invoices", "./invoice_summary.xlsx")
    
    # 单张发票测试
    # info = extract_invoice_info("./invoices/single_invoice.pdf")
    # print(info)