Python利用Camelot和Tabula轻松提取复杂表格数据

brew install ghostscript

sudo apt-get install ghostscript

pip install camelot-py[cv]

python -m venv camelot_env
source camelot_env/bin/activate  # Windows用 camelot_env\Scripts\activate
pip install camelot-py[cv]

import camelot
print(camelot.__version__)

pip install opencv-python

pip install tabula-py

pip install tabula-py[jpype]

import camelot
# Lattice模式是默认模式，直接使用即可
tables = camelot.read_pdf('financial_report.pdf')
# 也可以显式指定flavor参数
tables = camelot.read_pdf('financial_report.pdf', flavor='lattice')
# 查看提取结果
print(f"共提取到 {len(tables)} 个表格")
print(tables[0].df)

# 当PDF中存在背景线条可能干扰识别时
tables = camelot.read_pdf(
    'report_with_background.pdf',
    flavor='lattice',
    process_background=True  # 处理背景线条
)

# 当表格边框不完整时，调整line_scale提高线条检测灵敏度
tables = camelot.read_pdf(
    'incomplete_border.pdf',
    flavor='lattice',
    line_scale=60
)

import camelot

# 显式使用Stream模式
tables = camelot.read_pdf(
    'no_border_table.pdf',
    flavor='stream'
)

# Stream模式默认会把整个PDF页面当做一个表格来解析，需要通过参数指定区域
tables = camelot.read_pdf(
    'no_border_table.pdf',
    flavor='stream',
    table_areas=['50,700,500,100'],  # 指定表格区域
    columns=['A', 'B', 'C']  # 手动指定列边界
)

import tabula

# 读取PDF中的所有表格（默认只读取第一页）
tables = tabula.read_pdf("example.pdf")

# 读取所有页面的所有表格
tables = tabula.read_pdf("example.pdf", pages='all', multiple_tables=True)

# 读取指定页面
tables = tabula.read_pdf("example.pdf", pages='1,3,5-7')

# 读取远程PDF
tables = tabula.read_pdf("https://example.com/report.pdf")

# 使用lattice模式（适用于有边框的表格）
tables = tabula.read_pdf(
    "structured_table.pdf",
    pages='all',
    lattice=True,
    multiple_tables=True
)

# 使用stream模式（适用于无边框表格）
tables = tabula.read_pdf(
    "text_table.pdf",
    pages='all',
    lattice=False
)

# 直接转换为CSV文件
tabula.convert_into(
    "example.pdf",
    "output.csv",
    output_format="csv",
    pages='all'
)

# 批量转换目录下所有PDF
tabula.convert_into_by_batch(
    "pdf_directory/",
    output_format='csv',
    pages='all'
)

import camelot

# 提取第1-10页的所有表格
tables = camelot.read_pdf(
    'large_report.pdf',
    pages='1-10',
    flavor='lattice'
)

# 提取指定页面：第1、3、5-7页
tables = camelot.read_pdf(
    'large_report.pdf',
    pages='1,3,5-7'
)

# 提取所有页面
tables = camelot.read_pdf(
    'large_report.pdf',
    pages='all'
)

import camelot
import pandas as pd

# 提取所有页面中的表格
tables = camelot.read_pdf('multi_page_table.pdf', pages='all', flavor='lattice')

# 方法1：检查解析质量后选择性合并
all_data = []
for i, table in enumerate(tables):
    # 获取解析报告，过滤低质量提取结果
    report = table.parsing_report
    if report['accuracy'] > 90:  # 只保留准确率高于90%的表格
        all_data.append(table.df)
        print(f"第{report['page']}页表格，准确率：{report['accuracy']}%")

# 纵向拼接所有表格
if all_data:
    merged_df = pd.concat(all_data, ignore_index=True)
    merged_df.to_excel('merged_output.xlsx', index=False)
else:
    print("未找到有效的表格数据")

# 高级版本：处理表头重复和列对齐
import camelot
import pandas as pd

def merge_multi_page_tables(pdf_path, pages, flavor='lattice'):
    """
    合并跨页表格，自动处理表头和列对齐问题
    """
    tables = camelot.read_pdf(pdf_path, pages=pages, flavor=flavor)
    
    all_data = []
    header = None
    
    for idx, table in enumerate(tables):
        df = table.df
        
        # 第一页：记录表头
        if header is None:
            header = df.iloc[0].tolist()
            all_data.append(df.iloc[1:])  # 跳过表头行
        else:
            # 后续页面：检查并去除可能的表头行
            if df.iloc[0].tolist() == header:
                all_data.append(df.iloc[1:])  # 跳过重复表头
            else:
                all_data.append(df)
    
    # 确保所有子表列数一致
    max_cols = max([df.shape[1] for df in all_data])
    for i, df in enumerate(all_data):
        if df.shape[1] < max_cols:
            for _ in range(max_cols - df.shape[1]):
                df.insert(df.shape[1], f'col_{df.shape[1]}', '')
    
    result = pd.concat(all_data, ignore_index=True)
    result.columns = header
    return result

# 使用示例
merged_df = merge_multi_page_tables(
    'financial_report.pdf',
    pages='5-15',
    flavor='lattice'
)
merged_df.to_excel('merged_financial_data.xlsx', index=False)

import camelot
import pandas as pd
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor

def process_single_pdf(pdf_path):
    """处理单个PDF文件"""
    try:
        tables = camelot.read_pdf(pdf_path, pages='all', flavor='lattice')
        if len(tables) > 0:
            # 合并所有表格
            df = pd.concat([t.df for t in tables], ignore_index=True)
            return pdf_path.stem, df
    except Exception as e:
        print(f"处理 {pdf_path} 时出错: {e}")
        return pdf_path.stem, None

def batch_process_pdfs(folder_path, output_path):
    """批量处理文件夹中的所有PDF"""
    pdf_files = list(Path(folder_path).glob('*.pdf'))
    
    # 使用多进程加速处理
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(process_single_pdf, pdf_files))
    
    # 将结果写入Excel的不同Sheet
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        for filename, df in results:
            if df is not None:
                # Sheet名称不超过31个字符
                sheet_name = filename[:31]
                df.to_excel(writer, sheet_name=sheet_name, index=False)
    
    print(f"批量处理完成，结果保存至 {output_path}")

# 使用示例
batch_process_pdfs('./pdf_reports/', 'batch_output.xlsx')

import tabula
import pandas as pd
from pathlib import Path

# 方式1：批量转换所有PDF为CSV
tabula.convert_into_by_batch(
    "pdf_directory/",
    output_format='csv',
    pages='all'
)

# 方式2：遍历处理多个PDF并合并到单个Excel
def batch_extract_tabula(pdf_folder, output_excel):
    pdf_files = list(Path(pdf_folder).glob('*.pdf'))
    all_data = {}
    
    for pdf_file in pdf_files:
        # 提取所有表格
        tables = tabula.read_pdf(
            str(pdf_file),
            pages='all',
            multiple_tables=True
        )
        
        if tables:
            # 如果单个PDF有多个表格，合并它们
            combined = pd.concat(tables, ignore_index=True)
            all_data[pdf_file.stem] = combined
    
    with pd.ExcelWriter(output_excel) as writer:
        for name, df in all_data.items():
            df.to_excel(writer, sheet_name=name[:31], index=False)

batch_extract_tabula('./pdfs/', 'tabula_batch_output.xlsx')

import camelot
import pandas as pd
import numpy as np

def extract_financial_statement(pdf_path, start_page, end_page):
    """
    提取财务报表数据
    
    参数:
        pdf_path: PDF文件路径
        start_page: 起始页
        end_page: 结束页
    """
    # 使用Lattice模式提取所有页面
    tables = camelot.read_pdf(
        pdf_path,
        pages=f'{start_page}-{end_page}',
        flavor='lattice',
        process_background=True  # 处理可能的背景线条干扰
    )
    
    print(f"共提取到 {len(tables)} 个表格片段")
    
    # 检查每个表格的解析质量
    quality_report = []
    for i, table in enumerate(tables):
        report = table.parsing_report
        quality_report.append({
            'table_index': i,
            'page': report['page'],
            'accuracy': report['accuracy'],
            'whitespace': report['whitespace']
        })
        print(f"表格 {i} - 页码:{report['page']}, 准确率:{report['accuracy']}%")
    
    # 筛选高质量表格（准确率>90%）
    high_quality_tables = [
        table for table in tables 
        if table.parsing_report['accuracy'] > 90
    ]
    
    # 拼接所有表格
    all_dfs = [table.df for table in high_quality_tables]
    
    # 处理合并单元格造成的空值
    for df in all_dfs:
        # 前向填充空值（处理合并单元格）
        df.fillna(method='ffill', axis=0, inplace=True)
    
    # 纵向拼接
    final_df = pd.concat(all_dfs, ignore_index=True)
    
    # 数据清洗：去除全为空的行和列
    final_df.dropna(how='all', axis=0, inplace=True)
    final_df.dropna(how='all', axis=1, inplace=True)
    
    # 设置第一行为列名（通常是表头）
    final_df.columns = final_df.iloc[0]
    final_df = final_df.iloc[1:].reset_index(drop=True)
    
    return final_df, quality_report

# 执行提取
pdf_path = 'annual_report_2024.pdf'
financial_data, report = extract_financial_statement(pdf_path, 5, 15)

# 保存结果
financial_data.to_excel('profit_statement.xlsx', index=False)
print(f"提取完成！共获得 {len(financial_data)} 行数据")

# 数据验证与清洗
def validate_extracted_data(df, expected_columns):
    """验证提取数据的完整性和准确性"""
    
    print("=" * 50)
    print("数据验证报告")
    print("=" * 50)
    
    # 1. 检查列数是否与预期一致
    print(f"实际列数: {len(df.columns)}")
    print(f"预期列数: {len(expected_columns)}")
    
    # 2. 检查关键数据列是否存在
    for col in expected_columns:
        if col in df.columns:
            print(f"✓ 列 '{col}' 存在")
        else:
            print(f"✗ 列 '{col}' 缺失")
    
    # 3. 检查数值列的数据类型
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    print(f"数值列: {len(numeric_cols)} 列")
    
    # 4. 检查空值比例
    null_ratio = df.isnull().sum() / len(df) * 100
    for col, ratio in null_ratio.items():
        if ratio > 10:
            print(f"⚠ 列 '{col}' 空值比例过高: {ratio:.1f}%")
    
    # 5. 数据预览
    print("\n数据预览:")
    print(df.head(10))
    
    return df

expected_columns = ['项目', '2024年度', '2023年度', '增减幅度(%)']
validate_extracted_data(financial_data, expected_columns)

from pdf2image import convert_from_path
import pytesseract
import camelot

def extract_with_ocr(pdf_path, pages):
    """
    处理扫描件PDF：先OCR识别，再提取表格
    """
    # 将PDF页面转换为图像
    images = convert_from_path(pdf_path, dpi=300, first_page=pages[0], last_page=pages[-1])
    
    # 保存为临时PDF文件
    temp_pdf_path = 'temp_ocr_output.pdf'
    images[0].save(temp_pdf_path, save_all=True, append_images=images[1:])
    
    # 对OCR后的PDF进行表格提取
    tables = camelot.read_pdf(temp_pdf_path, pages='all', flavor='lattice')
    
    return tables

# 无边框表格，尝试Stream模式
tables = camelot.read_pdf(
    'no_border_table.pdf',
    flavor='stream',
    table_areas=['50,700,550,150'],  # 指定表格区域
    columns=['A', 'B', 'C', 'D']  # 手动指定列边界
)

# 先获取文本位置信息，确定列边界
from camelot.utils import get_page_layout

layout = get_page_layout('no_border_table.pdf', pages='1')
print(layout)  # 查看文本块的x坐标，推断列边界位置

# 根据分析结果设置列边界
tables = camelot.read_pdf(
    'no_border_table.pdf',
    flavor='stream',
    columns=['100', '250', '400', '550']  # 列边界的x坐标
)

# 对于倾斜表格，使用Lattice模式并调整line_scale参数
tables = camelot.read_pdf(
    'rotated_table.pdf',
    flavor='lattice',
    line_scale=40  # 提高线条检测灵敏度
)

import cv2
import numpy as np
from pdf2image import convert_from_path

def deskew_table(pdf_path, page_num):
    """检测并矫正倾斜的表格"""
    images = convert_from_path(pdf_path, first_page=page_num, last_page=page_num)
    img = np.array(images[0])
    
    # 转换为灰度图
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    
    # 检测边缘
    edges = cv2.Canny(gray, 50, 150, apertureSize=3)
    
    # 霍夫变换检测直线
    lines = cv2.HoughLines(edges, 1, np.pi/180, threshold=100)
    
    if lines is not None:
        # 计算倾斜角度
        angles = []
        for rho, theta in lines[:,0]:
            angle = theta * 180 / np.pi - 90
            if abs(angle) < 45:
                angles.append(angle)
        
        if angles:
            median_angle = np.median(angles)
            # 旋转矫正
            (h, w) = img.shape[:2]
            center = (w // 2, h // 2)
            M = cv2.getRotationMatrix2D(center, median_angle, 1.0)
            rotated = cv2.warpAffine(img, M, (w, h))
            return rotated
    
    return img

# 使用process_background参数过滤背景线条
tables = camelot.read_pdf(
    'table_with_background.pdf',
    flavor='lattice',
    process_background=True  # 区分前景和背景线条
)

def handle_merged_cells(df):
    """
    处理合并单元格造成的空值问题
    """
    # 纵向填充（处理行合并）
    df.fillna(method='ffill', axis=0, inplace=True)
    
    # 横向填充（处理列合并）
    df.fillna(method='ffill', axis=1, inplace=True)
    
    return df

# 使用示例
df_cleaned = handle_merged_cells(extracted_df)

import camelot

# 提取表格
tables = camelot.read_pdf('problematic_table.pdf', flavor='lattice')

# 可视化检测到的表格网格
tables[0].plot(kind='grid')
# 可视化检测到的文本
tables[0].plot(kind='text')
# 可视化表格轮廓
tables[0].plot(kind='contour')

# 保存可视化图像
import matplotlib.pyplot as plt
plt.savefig('debug_grid.png')