Python使用Camelot从PDF中精准获取表格数据

# 基本安装
pip install camelot-py[cv]
# 如果需要pdf转换功能
pip install ghostscript

import camelot
import pandas as pd
import matplotlib.pyplot as plt
import cv2

def extract_basic_tables(pdf_path, pages='1'):
    """从pdf中提取基本表格"""
    # 使用stream模式提取表格
    tables = camelot.read_pdf(pdf_path, pages=pages, flavor='stream')
    print(f"检测到 {len(tables)} 个表格")
    # 表格基本信息
    for i, table in enumerate(tables):
        print(f"\n表格 #{i+1}:")
        print(f"页码: {table.page}")
        print(f"表格区域: {table.area}")
        print(f"维度: {table.shape}")
        print(f"准确度分数: {table.accuracy}")
        print(f"空白率: {table.whitespace}")
        # 显示表格前几行
        print("\n表格预览:")
        print(table.df.head())
    return tables
# 使用示例
tables = extract_basic_tables("financial_report.pdf", pages='1-3')

def compare_extraction_methods(pdf_path, page='1'):
    """比较stream和lattice两种提取方法"""
    # 使用stream方法
    stream_tables = camelot.read_pdf(pdf_path, pages=page, flavor='stream')
    # 使用lattice方法
    lattice_tables = camelot.read_pdf(pdf_path, pages=page, flavor='lattice')
    # 比较结果
    print(f"stream方法: 检测到 {len(stream_tables)} 个表格")
    print(f"lattice方法: 检测到 {len(lattice_tables)} 个表格")
    # 如果检测到表格，比较第一个表格
    if len(stream_tables) > 0 and len(lattice_tables) > 0:
        # 获取第一个表格
        stream_table = stream_tables[0]
        lattice_table = lattice_tables[0]
        # 比较准确度和空白率
        print("\n准确度和空白率比较:")
        print(f"stream - 准确度: {stream_table.accuracy}, 空白率: {stream_table.whitespace}")
        print(f"lattice - 准确度: {lattice_table.accuracy}, 空白率: {lattice_table.whitespace}")
        # 比较表格形状
        print("\n表格维度比较:")
        print(f"stream: {stream_table.shape}")
        print(f"lattice: {lattice_table.shape}")
        # 返回两种方法的表格
        return stream_tables, lattice_tables
    return none, none
# 使用示例
stream_tables, lattice_tables = compare_extraction_methods("report_with_tables.pdf")

def extract_table_with_area(pdf_path, page='1', table_area=none):
    """使用精确区域坐标提取表格"""
    if table_area is none:
        # 默认值覆盖整个页面
        table_area = [0, 0, 100, 100]  # [x1, y1, x2, y2] 以百分比表示
    # 使用stream方法提取指定区域的表格
    tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='stream',
        table_areas=[f"{table_area[0]},{table_area[1]},{table_area[2]},{table_area[3]}"]
    )
    print(f"在指定区域检测到 {len(tables)} 个表格")
    # 显示第一个表格
    if len(tables) > 0:
        print("\n表格预览:")
        print(tables[0].df.head())
    return tables
# 使用示例 - 提取页面中间大约位置的表格
tables = extract_table_with_area("financial_report.pdf", table_area=[10, 30, 90, 70])

def extract_complex_tables(pdf_path, page='1'):
    """处理复杂表格的高级配置"""
    # 使用lattice方法处理有边框的复杂表格
    lattice_tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='lattice',
        line_scale=40,  # 调整线条检测灵敏度
        process_background=true,  # 处理背景
        line_margin=2  # 线条间隔容忍度
    )
    # 使用stream方法处理无边框的复杂表格
    stream_tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='stream',
        edge_tol=500,  # 边缘容忍度
        row_tol=10,    # 行容忍度
        column_tol=10  # 列容忍度
    )
    print(f"lattice方法: 检测到 {len(lattice_tables)} 个表格")
    print(f"stream方法: 检测到 {len(stream_tables)} 个表格")
    # 选择最佳结果
    best_tables = lattice_tables if lattice_tables[0].accuracy > stream_tables[0].accuracy else stream_tables
    return best_tables
# 使用示例
complex_tables = extract_complex_tables("complex_financial_report.pdf")

def visualize_table_extraction(pdf_path, page='1'):
    """可视化表格提取过程，帮助调试和优化"""
    # 提取表格
    tables = camelot.read_pdf(pdf_path, pages=page)
    # 检查是否成功提取表格
    if len(tables) == 0:
        print("未检测到表格")
        return
    # 获取第一个表格
    table = tables[0]
    # 显示表格
    print(f"表格形状: {table.shape}")
    print(f"准确度: {table.accuracy}")
    # 绘制表格结构
    plot = table.plot(kind='grid')
    plt.title(f"表格网格结构 - 准确度: {table.accuracy}")
    plt.tight_layout()
    plt.savefig('table_grid.png')
    plt.close()
    # 绘制表格单元格
    plot = table.plot(kind='contour')
    plt.title(f"表格单元格结构 - 空白率: {table.whitespace}")
    plt.tight_layout()
    plt.savefig('table_contour.png')
    plt.close()
    # 绘制表格线条（仅适用于lattice方法）
    if table.flavor == 'lattice':
        plot = table.plot(kind='line')
        plt.title("表格线条检测")
        plt.tight_layout()
        plt.savefig('table_lines.png')
        plt.close()
    print("可视化图形已保存")
    return tables
# 使用示例
visualized_tables = visualize_table_extraction("quarterly_report.pdf")

def clean_table_data(table):
    """清洗从pdf提取的表格数据"""
    # 获取dataframe
    df = table.df.copy()
    # 1. 替换空白单元格
    df = df.replace('', pd.na)
    # 2. 清理多余空格
    for col in df.columns:
        if df[col].dtype == object:  # 仅处理字符串列
            df[col] = df[col].str.strip() if df[col].notna().any() else df[col]
    # 3. 处理合并单元格的问题（向下填充）
    df = df.fillna(method='ffill')
    # 4. 检测并移除页眉或页脚（通常出现在第一行或最后一行）
    if df.shape[0] > 2:
        # 检查第一行是否为页眉
        if df.iloc[0].astype(str).str.contains('page|页码|日期').any():
            df = df.iloc[1:]
        # 检查最后一行是否为页脚
        if df.iloc[-1].astype(str).str.contains('总计|合计|total').any():
            df = df.iloc[:-1]
    # 5. 重置索引
    df = df.reset_index(drop=true)
    # 6. 设置第一行为列名（可选）
    # df.columns = df.iloc[0]
    # df = df.iloc[1:].reset_index(drop=true)
    return df
# 使用示例
tables = camelot.read_pdf("financial_data.pdf")
if tables:
    cleaned_df = clean_table_data(tables[0])
    print(cleaned_df.head())

def merge_tables(tables, merge_method='vertical'):
    """合并多个表格"""
    if not tables or len(tables) == 0:
        return none
    dfs = [table.df for table in tables]
    if merge_method == 'vertical':
        # 垂直合并（适用于跨页表格）
        merged_df = pd.concat(dfs, ignore_index=true)
    elif merge_method == 'horizontal':
        # 水平合并（适用于分列表格）
        merged_df = pd.concat(dfs, axis=1)
    else:
        raise valueerror("合并方法必须是 'vertical' 或 'horizontal'")
    # 清洗合并后的数据
    # 删除完全相同的重复行（可能来自表格页眉）
    merged_df = merged_df.drop_duplicates()
    return merged_df
# 使用示例 - 合并跨页表格
tables = camelot.read_pdf("multipage_report.pdf", pages='1-3')
if tables:
    merged_table = merge_tables(tables, merge_method='vertical')
    print(f"合并后表格大小: {merged_table.shape}")
    print(merged_table.head())

def convert_table_datatypes(df):
    """将表格数据转换为适当的数据类型"""
    # 创建dataframe副本
    df = df.copy()
    for col in df.columns:
        # 尝试将列转换为数值型
        try:
            # 检查列是否包含数字（带有货币符号或千位分隔符）
            if df[col].str.contains(r'[$¥€£]|\d,\d').any():
                # 移除货币符号和千位分隔符
                df[col] = df[col].replace(r'[$¥€£,]', '', regex=true)
            # 尝试转换为数值型
            df[col] = pd.to_numeric(df[col])
            print(f"列 '{col}' 已转换为数值型")
        except (valueerror, attributeerror):
            # 尝试转换为日期型
            try:
                df[col] = pd.to_datetime(df[col])
                print(f"列 '{col}' 已转换为日期型")
            except (valueerror, attributeerror):
                # 保持为字符串型
                pass
    return df
# 使用示例
tables = camelot.read_pdf("sales_report.pdf")
if tables:
    df = clean_table_data(tables[0])
    typed_df = convert_table_datatypes(df)
    print(typed_df.dtypes)

def extract_financial_statements(pdf_path, pages='all'):
    """从年度报告中提取财务报表"""
    # 提取所有表格
    tables = camelot.read_pdf(
        pdf_path,
        pages=pages,
        flavor='stream',
        edge_tol=500,
        row_tol=10
    )
    print(f"共提取了 {len(tables)} 个表格")
    # 查找财务报表（通过关键词）
    balance_sheet = none
    income_statement = none
    cash_flow = none
    for table in tables:
        df = table.df
        # 检查表格是否包含特定关键词
        text = ' '.join([' '.join(row) for row in df.values.tolist()])
        if any(term in text for term in ['资产负债表', 'balance sheet', '财务状况表']):
            balance_sheet = clean_table_data(table)
            print("找到资产负债表")
        elif any(term in text for term in ['利润表', 'income statement', '损益表']):
            income_statement = clean_table_data(table)
            print("找到利润表")
        elif any(term in text for term in ['现金流量表', 'cash flow']):
            cash_flow = clean_table_data(table)
            print("找到现金流量表")
    return {
        'balance_sheet': balance_sheet,
        'income_statement': income_statement,
        'cash_flow': cash_flow
    }
# 使用示例
financial_data = extract_financial_statements("annual_report_2022.pdf", pages='10-30')
for statement_name, df in financial_data.items():
    if df is not none:
        print(f"\n{statement_name}:")
        print(df.head())

def batch_process_pdfs(pdf_folder, output_folder='extracted_tables'):
    """批量处理多个pdf文件，提取所有表格"""
    import os
    from pathlib import path
    # 创建输出文件夹
    path(output_folder).mkdir(exist_ok=true)
    # 获取所有pdf文件
    pdf_files = [f for f in os.listdir(pdf_folder) if f.lower().endswith('.pdf')]
    results = {}
    for pdf_file in pdf_files:
        pdf_path = os.path.join(pdf_folder, pdf_file)
        pdf_name = os.path.splitext(pdf_file)[0]
        print(f"\n处理: {pdf_file}")
        # 创建pdf专属输出文件夹
        pdf_output_folder = os.path.join(output_folder, pdf_name)
        path(pdf_output_folder).mkdir(exist_ok=true)
        try:
            # 提取表格
            tables = camelot.read_pdf(pdf_path, pages='all')
            print(f"从 {pdf_file} 提取了 {len(tables)} 个表格")
            # 保存每个表格为csv文件
            for i, table in enumerate(tables):
                df = clean_table_data(table)
                output_path = os.path.join(pdf_output_folder, f"table_{i+1}.csv")
                df.to_csv(output_path, index=false, encoding='utf-8-sig')
            # 记录结果
            results[pdf_file] = {
                'status': 'success',
                'tables_count': len(tables),
                'output_folder': pdf_output_folder
            }
        except exception as e:
            print(f"处理 {pdf_file} 时出错: {str(e)}")
            results[pdf_file] = {
                'status': 'error',
                'error_message': str(e)
            }
    # 汇总报告
    success_count = sum(1 for result in results.values() if result['status'] == 'success')
    print(f"\n批处理完成。成功: {success_count}/{len(pdf_files)}")
    return results
# 使用示例
batch_results = batch_process_pdfs("reports_folder", "extracted_data")

def create_dashboard_from_tables(tables, output_html='table_dashboard.html'):
    """从提取的表格创建简单的交互式仪表板"""
    import plotly.express as px
    import plotly.graph_objects as go
    from plotly.subplots import make_subplots
    import pandas as pd
    # 确保我们有表格
    if not tables or len(tables) == 0:
        print("没有表格数据可用于创建仪表板")
        return
    # 为了简单起见，使用第一个表格
    df = clean_table_data(tables[0])
    # 如果所有列都是字符串，尝试将其中一些转换为数值
    df = convert_table_datatypes(df)
    # 创建仪表板 html
    with open(output_html, 'w', encoding='utf-8') as f:
        f.write("<html><head>")
        f.write("<title>pdf表格数据仪表板</title>")
        f.write("<style>body {font-family: arial; margin: 20px;} .chart {margin: 20px 0; padding: 20px; border: 1px solid #ddd;}</style>")
        f.write("</head><body>")
        f.write("<h1>pdf表格数据仪表板</h1>")
        # 添加表格
        f.write("<div class='chart'>")
        f.write("<h2>提取的表格数据</h2>")
        f.write(df.to_html(classes='dataframe', index=false))
        f.write("</div>")
        # 如果有数值型列，创建图表
        numeric_cols = df.select_dtypes(include=['number']).columns
        if len(numeric_cols) > 0:
            # 选择第一个数值列创建图表
            value_col = numeric_cols[0]
            # 寻找一个可能的类别列
            category_col = none
            for col in df.columns:
                if col != value_col and df[col].dtype == object and df[col].nunique() < len(df) * 0.5:
                    category_col = col
                    break
            if category_col:
                # 创建条形图
                fig = px.bar(df, x=category_col, y=value_col, title=f"{category_col} vs {value_col}")
                f.write("<div class='chart'>")
                f.write(f"<h2>{category_col} vs {value_col}</h2>")
                f.write(fig.to_html(full_html=false))
                f.write("</div>")
                # 创建饼图
                fig = px.pie(df, names=category_col, values=value_col, title=f"{value_col} by {category_col}")
                f.write("<div class='chart'>")
                f.write(f"<h2>{value_col} by {category_col} (饼图)</h2>")
                f.write(fig.to_html(full_html=false))
                f.write("</div>")
        f.write("</body></html>")
    print(f"仪表板已创建: {output_html}")
    return output_html
# 使用示例
tables = camelot.read_pdf("sales_by_region.pdf")
if tables:
    dashboard_path = create_dashboard_from_tables(tables)

def optimize_table_detection(pdf_path, page='1'):
    """优化表格检测参数，尝试不同设置并评估结果"""
    # 定义不同的参数组合
    stream_configs = [
        {'edge_tol': 50, 'row_tol': 5, 'column_tol': 5},
        {'edge_tol': 100, 'row_tol': 10, 'column_tol': 10},
        {'edge_tol': 500, 'row_tol': 15, 'column_tol': 15}
    ]
    lattice_configs = [
        {'process_background': true, 'line_scale': 15},
        {'process_background': true, 'line_scale': 40},
        {'process_background': true, 'line_scale': 60, 'iterations': 1}
    ]
    results = []
    # 测试stream方法的不同配置
    print("测试stream方法...")
    for config in stream_configs:
        try:
            tables = camelot.read_pdf(
                pdf_path,
                pages=page,
                flavor='stream',
                **config
            )
            # 评估结果
            if len(tables) > 0:
                accuracy = tables[0].accuracy
                whitespace = tables[0].whitespace
                print(f"配置 {config}: 准确度={accuracy:.2f}, 空白率={whitespace:.2f}")
                results.append({
                    'flavor': 'stream',
                    'config': config,
                    'tables_found': len(tables),
                    'accuracy': accuracy,
                    'whitespace': whitespace,
                    'tables': tables
                })
        except exception as e:
            print(f"配置 {config} 出错: {str(e)}")
    # 测试lattice方法的不同配置
    print("\n测试lattice方法...")
    for config in lattice_configs:
        try:
            tables = camelot.read_pdf(
                pdf_path,
                pages=page,
                flavor='lattice',
                **config
            )
            # 评估结果
            if len(tables) > 0:
                accuracy = tables[0].accuracy
                whitespace = tables[0].whitespace
                print(f"配置 {config}: 准确度={accuracy:.2f}, 空白率={whitespace:.2f}")
                results.append({
                    'flavor': 'lattice',
                    'config': config,
                    'tables_found': len(tables),
                    'accuracy': accuracy,
                    'whitespace': whitespace,
                    'tables': tables
                })
        except exception as e:
            print(f"配置 {config} 出错: {str(e)}")
    # 找出最佳配置
    if results:
        # 按准确度排序
        best_result = sorted(results, key=lambda x: x['accuracy'], reverse=true)[0]
        print(f"\n最佳配置: {best_result['flavor']} 方法, 参数: {best_result['config']}")
        print(f"准确度: {best_result['accuracy']:.2f}, 空白率: {best_result['whitespace']:.2f}")
        return best_result['tables']
    return none
# 使用示例
optimized_tables = optimize_table_detection("complex_report.pdf")

def extract_tables_from_scanned_pdf(pdf_path, page='1'):
    """从扫描pdf中提取表格（需要预处理）"""
    import cv2
    import numpy as np
    import tempfile
    from pdf2image import convert_from_path
    # 转换pdf页面为图像
    images = convert_from_path(pdf_path, first_page=int(page), last_page=int(page))
    if not images:
        print("无法转换pdf页面为图像")
        return none
    # 获取第一个页面图像
    image = np.array(images[0])
    # 图像预处理
    gray = cv2.cvtcolor(image, cv2.color_bgr2gray)
    thresh = cv2.threshold(gray, 150, 255, cv2.thresh_binary)[1]
    # 保存处理后的图像
    with tempfile.namedtemporaryfile(suffix='.png', delete=false) as tmp:
        temp_image_path = tmp.name
        cv2.imwrite(temp_image_path, thresh)
    print(f"扫描页面已预处理并保存为临时图像: {temp_image_path}")
    # 从图像中提取表格
    tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='lattice',
        process_background=true,
        line_scale=150
    )
    print(f"从扫描pdf提取了 {len(tables)} 个表格")
    # 可选: 移除临时文件
    import os
    os.unlink(temp_image_path)
    return tables
# 使用示例
scanned_tables = extract_tables_from_scanned_pdf("scanned_report.pdf")

def handle_merged_cells(table):
    """处理表格中的合并单元格"""
    # 获取dataframe
    df = table.df.copy()
    # 检测并处理垂直合并的单元格
    for col in df.columns:
        # 在连续的空单元格上向下填充值
        mask = df[col].eq('')
        if mask.any():
            prev_value = none
            fill_values = []
            for idx, is_empty in enumerate(mask):
                if not is_empty:
                    prev_value = df.at[idx, col]
                elif prev_value is not none:
                    fill_values.append((idx, prev_value))
            # 填充检测到的合并单元格
            for idx, value in fill_values:
                df.at[idx, col] = value
    # 检测并处理水平合并的单元格
    for idx, row in df.iterrows():
        empty_cols = row.index[row == ''].tolist()
        if empty_cols and idx > 0:
            # 检查这一行是否有空单元格后跟非空单元格
            for i, col in enumerate(empty_cols):
                if i + 1 < len(row) and row.iloc[i + 1] != '':
                    # 可能是水平合并，从左侧单元格填充
                    left_col_idx = row.index.get_loc(col) - 1
                    if left_col_idx >= 0 and row.iloc[left_col_idx] != '':
                        df.at[idx, col] = row.iloc[left_col_idx]
    return df
# 使用示例
tables = camelot.read_pdf("report_with_merged_cells.pdf")
if tables:
    cleaned_df = handle_merged_cells(tables[0])
    print(cleaned_df.head())

def analyze_extracted_table(table):
    """使用pandas分析提取的表格数据"""
    # 清洁数据
    df = clean_table_data(table)
    # 转换数据类型
    df = convert_table_datatypes(df)
    # 基本统计分析
    print("\n==== 基本统计分析 ====")
    numeric_cols = df.select_dtypes(include=['number']).columns
    if len(numeric_cols) > 0:
        print(df[numeric_cols].describe())
    # 检查缺失值
    print("\n==== 缺失值分析 ====")
    missing = df.isnull().sum()
    print(missing[missing > 0])
    # 类别变量分析
    print("\n==== 类别变量分析 ====")
    categorical_cols = df.select_dtypes(include=['object']).columns
    for col in categorical_cols[:3]:  # 只显示前三个类别列
        value_counts = df[col].value_counts()
        print(f"\n{col}:")
        print(value_counts.head())
    # 相关性分析
    if len(numeric_cols) >= 2:
        print("\n==== 相关性分析 ====")
        correlation = df[numeric_cols].corr()
        print(correlation)
    return df
# 使用示例
tables = camelot.read_pdf("sales_data.pdf")
if tables:
    analyzed_df = analyze_extracted_table(tables[0])

def visualize_table_data(table, output_prefix='table_viz'):
    """使用matplotlib和seaborn可视化表格数据"""
    import matplotlib.pyplot as plt
    import seaborn as sns
    # 设置样式
    sns.set(style="whitegrid")
    # 清洁并转换数据
    df = clean_table_data(table)
    df = convert_table_datatypes(df)
    # 获取数值列
    numeric_cols = df.select_dtypes(include=['number']).columns
    if len(numeric_cols) == 0:
        print("没有数值列可供可视化")
        return
    # 1. 热图 - 相关性
    if len(numeric_cols) >= 2:
        plt.figure(figsize=(10, 8))
        corr = df[numeric_cols].corr()
        mask = np.triu(np.ones_like(corr, dtype=bool))
        sns.heatmap(corr, mask=mask, annot=true, fmt=".2f", cmap='coolwarm', 
                   square=true, linewidths=.5)
        plt.title('相关性热图', fontsize=15)
        plt.tight_layout()
        plt.savefig(f'{output_prefix}_correlation.png')
        plt.close()
    # 2. 条形图 - 数值分布
    for col in numeric_cols[:3]:  # 只处理前三个数值列
        plt.figure(figsize=(12, 6))
        sns.barplot(x=df.index, y=df[col])
        plt.title(f'{col} 分布', fontsize=15)
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig(f'{output_prefix}_{col}_barplot.png')
        plt.close()
    # 3. 箱线图 - 查找异常值
    plt.figure(figsize=(12, 6))
    sns.boxplot(data=df[numeric_cols])
    plt.title('数值列箱线图', fontsize=15)
    plt.tight_layout()
    plt.savefig(f'{output_prefix}_boxplot.png')
    plt.close()
    # 4. 散点图矩阵 - 变量关系
    if len(numeric_cols) >= 2 and len(df) > 5:
        sns.pairplot(df[numeric_cols])
        plt.suptitle('散点图矩阵', y=1.02, fontsize=15)
        plt.tight_layout()
        plt.savefig(f'{output_prefix}_pairplot.png')
        plt.close()
    print(f"可视化图表已保存，前缀: {output_prefix}")
    return df
# 使用示例
tables = camelot.read_pdf("product_sales.pdf")
if tables:
    df = visualize_table_data(tables[0], output_prefix='sales_viz')

def export_tables_to_excel(tables, output_path='extracted_tables.xlsx'):
    """将提取的表格导出到excel工作簿的不同工作表"""
    import pandas as pd
    # 检查是否有表格
    if not tables or len(tables) == 0:
        print("没有表格可导出")
        return none
    # 创建excel writer对象
    with pd.excelwriter(output_path, engine='xlsxwriter') as writer:
        workbook = writer.book
        # 创建表格样式
        header_format = workbook.add_format({
            'bold': true,
            'text_wrap': true,
            'valign': 'top',
            'fg_color': '#d7e4bc',
            'border': 1
        })
        cell_format = workbook.add_format({
            'border': 1
        })
        # 为每个表格创建工作表
        for i, table in enumerate(tables):
            # 清洁数据
            df = clean_table_data(table)
            # 写入数据
            sheet_name = f'table_{i+1}'
            df.to_excel(writer, sheet_name=sheet_name, index=false)
            # 获取工作表对象
            worksheet = writer.sheets[sheet_name]
            # 设置列宽
            for j, col in enumerate(df.columns):
                column_width = max(
                    df[col].astype(str).map(len).max(),
                    len(col)
                ) + 2
                worksheet.set_column(j, j, column_width)
            # 设置表格格式
            worksheet.add_table(0, 0, df.shape[0], df.shape[1] - 1, {
                'columns': [{'header': col} for col in df.columns],
                'style': 'table style medium 9',
                'header_row': true
            })
            # 添加表格元数据
            worksheet.write(df.shape[0] + 2, 0, f"页码: {table.page}")
            worksheet.write(df.shape[0] + 3, 0, f"表格区域: {table.area}")
            worksheet.write(df.shape[0] + 4, 0, f"准确度分数: {table.accuracy}")
    print(f"表格已导出至excel: {output_path}")
    return output_path
# 使用示例
tables = camelot.read_pdf("quarterly_report.pdf", pages='all')
if tables:
    excel_path = export_tables_to_excel(tables, "quarterly_report_tables.xlsx")

def process_large_pdf(pdf_path, batch_size=5, output_folder='large_pdf_tables'):
    """分批处理大型pdf文档以节省内存"""
    import os
    from pathlib import path
    # 创建输出文件夹
    path(output_folder).mkdir(exist_ok=true)
    # 首先获取pdf页数
    with pdfplumber.open(pdf_path) as pdf:
        total_pages = len(pdf.pages)
    print(f"pdf共有 {total_pages} 页")
    # 分批处理页面
    all_tables_count = 0
    for start_page in range(1, total_pages + 1, batch_size):
        end_page = min(start_page + batch_size - 1, total_pages)
        page_range = f"{start_page}-{end_page}"
        print(f"处理页面范围: {page_range}")
        try:
            # 提取当前批次的表格
            tables = camelot.read_pdf(
                pdf_path,
                pages=page_range,
                flavor='stream',
                edge_tol=500
            )
            batch_tables_count = len(tables)
            all_tables_count += batch_tables_count
            print(f"从页面 {page_range} 提取了 {batch_tables_count} 个表格")
            # 保存这一批次的表格
            for i, table in enumerate(tables):
                table_index = all_tables_count - batch_tables_count + i + 1
                df = clean_table_data(table)
                output_path = os.path.join(output_folder, f"table_{table_index}_p{table.page}.csv")
                df.to_csv(output_path, index=false, encoding='utf-8-sig')
            # 显式释放内存
            tables = none
            import gc
            gc.collect()
        except exception as e:
            print(f"处理页面 {page_range} 时出错: {str(e)}")
    print(f"处理完成，共提取了 {all_tables_count} 个表格，保存到 {output_folder}")
    return all_tables_count
# 使用示例
table_count = process_large_pdf("very_large_report.pdf", batch_size=10)

def optimize_camelot_performance(pdf_path, page='1'):
    """调优camelot的性能参数"""
    import time
    import psutil
    import os
    def measure_performance(func, *args, **kwargs):
        """测量函数的执行时间和内存使用情况"""
        process = psutil.process(os.getpid())
        mem_before = process.memory_info().rss / 1024 / 1024  # mb
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        mem_after = process.memory_info().rss / 1024 / 1024  # mb
        execution_time = end_time - start_time
        memory_used = mem_after - mem_before
        return result, execution_time, memory_used
    # 测试不同的参数组合
    configs = [
        {
            'name': '默认配置',
            'params': {}
        },
        {
            'name': '启用后台处理',
            'params': {'process_background': true}
        },
        {
            'name': '禁用线条检测',
            'params': {'line_scale': 0}
        },
        {
            'name': '提高线条检测灵敏度',
            'params': {'line_scale': 80}
        }
    ]
    results = []
    for config in configs:
        print(f"\n测试配置: {config['name']}")
        # 测试lattice方法
        try:
            lattice_func = lambda: camelot.read_pdf(
                pdf_path, 
                pages=page, 
                flavor='lattice',
                **config['params']
            )
            lattice_tables, lattice_time, lattice_mem = measure_performance(lattice_func)
            results.append({
                'config_name': config['name'],
                'method': 'lattice',
                'time': lattice_time,
                'memory': lattice_mem,
                'tables_count': len(lattice_tables),
                'accuracy': lattice_tables[0].accuracy if len(lattice_tables) > 0 else 0
            })
            print(f"  lattice - 时间: {lattice_time:.2f}秒, 内存: {lattice_mem:.2f}mb, 准确度: {lattice_tables[0].accuracy if len(lattice_tables) > 0 else 0}")
        except exception as e:
            print(f"  lattice方法出错: {str(e)}")
        # 测试stream方法
        try:
            stream_func = lambda: camelot.read_pdf(
                pdf_path, 
                pages=page, 
                flavor='stream',
                **config['params']
            )
            stream_tables, stream_time, stream_mem = measure_performance(stream_func)
            results.append({
                'config_name': config['name'],
                'method': 'stream',
                'time': stream_time,
                'memory': stream_mem,
                'tables_count': len(stream_tables),
                'accuracy': stream_tables[0].accuracy if len(stream_tables) > 0 else 0
            })
            print(f"  stream - 时间: {stream_time:.2f}秒, 内存: {stream_mem:.2f}mb, 准确度: {stream_tables[0].accuracy if len(stream_tables) > 0 else 0}")
        except exception as e:
            print(f"  stream方法出错: {str(e)}")
    # 查找最佳性能配置
    if results:
        # 按准确度排序
        accuracy_best = sorted(results, key=lambda x: x['accuracy'], reverse=true)[0]
        print(f"\n最高准确度配置: {accuracy_best['config_name']} / {accuracy_best['method']}")
        print(f"  准确度: {accuracy_best['accuracy']:.2f}, 耗时: {accuracy_best['time']:.2f}秒")
        # 按时间排序
        time_best = sorted(results, key=lambda x: x['time'])[0]
        print(f"\n最快配置: {time_best['config_name']} / {time_best['method']}")
        print(f"  耗时: {time_best['time']:.2f}秒, 准确度: {time_best['accuracy']:.2f}")
        # 按内存使用排序
        memory_best = sorted(results, key=lambda x: x['memory'])[0]
        print(f"\n最低内存配置: {memory_best['config_name']} / {memory_best['method']}")
        print(f"  内存: {memory_best['memory']:.2f}mb, 准确度: {memory_best['accuracy']:.2f}")
        # 综合考虑速度和准确度的最佳配置
        balanced = sorted(results, key=lambda x: (1/x['accuracy']) * x['time'])[0]
        print(f"\n平衡配置: {balanced['config_name']} / {balanced['method']}")
        print(f"  准确度: {balanced['accuracy']:.2f}, 耗时: {balanced['time']:.2f}秒")
        return balanced
    return none
# 使用示例
best_config = optimize_camelot_performance("sample_report.pdf")

def compare_with_pypdf(pdf_path, page=0):
    """比较camelot与pypdf2的提取能力"""
    import pypdf2
    print("\n===== pypdf2提取结果 =====")
    try:
        # 使用pypdf2提取文本
        with open(pdf_path, 'rb') as file:
            reader = pypdf2.pdfreader(file)
            if page < len(reader.pages):
                text = reader.pages[page].extract_text()
                print(f"提取的文本 ({len(text)} 字符):")
                print(text[:500] + "..." if len(text) > 500 else text)
                print("\npypdf2无法识别表格结构，只能提取纯文本")
            else:
                print(f"页码 {page} 超出范围")
    except exception as e:
        print(f"pypdf2提取出错: {str(e)}")
    print("\n===== camelot提取结果 =====")
    try:
        # 使用camelot提取表格
        tables = camelot.read_pdf(pdf_path, pages=str(page+1))  # camelot页码从1开始
        print(f"检测到 {len(tables)} 个表格")
        if len(tables) > 0:
            table = tables[0]
            print(f"表格维度: {table.shape}")
            print(f"准确度: {table.accuracy}")
            print("\n表格预览:")
            print(table.df.head().to_string())
            print("\ncamelot可以识别表格结构，保留行列关系")
    except exception as e:
        print(f"camelot提取出错: {str(e)}")
    return none
# 使用示例
compare_with_pypdf("financial_data.pdf")

def compare_with_tabula(pdf_path, page='1'):
    """比较camelot与tabula的表格提取能力"""
    try:
        import tabula
    except importerror:
        print("请安装tabula-py: pip install tabula-py")
        return
    print("\n===== tabula提取结果 =====")
    try:
        # 使用tabula提取表格
        tabula_tables = tabula.read_pdf(pdf_path, pages=page)
        print(f"检测到 {len(tabula_tables)} 个表格")
        if len(tabula_tables) > 0:
            tabula_df = tabula_tables[0]
            print(f"表格维度: {tabula_df.shape}")
            print("\n表格预览:")
            print(tabula_df.head().to_string())
    except exception as e:
        print(f"tabula提取出错: {str(e)}")
    print("\n===== camelot提取结果 =====")
    try:
        # 使用camelot提取表格
        camelot_tables = camelot.read_pdf(pdf_path, pages=page)
        print(f"检测到 {len(camelot_tables)} 个表格")
        if len(camelot_tables) > 0:
            camelot_df = camelot_tables[0].df
            print(f"表格维度: {camelot_df.shape}")
            print(f"准确度: {camelot_tables[0].accuracy}")
            print("\n表格预览:")
            print(camelot_df.head().to_string())
    except exception as e:
        print(f"camelot提取出错: {str(e)}")
    # 比较结果
    if 'tabula_tables' in locals() and 'camelot_tables' in locals():
        if len(tabula_tables) > 0 and len(camelot_tables) > 0:
            tabula_df = tabula_tables[0]
            camelot_df = camelot_tables[0].df
            print("\n===== 比较结果 =====")
            print(f"tabula表格大小: {tabula_df.shape}")
            print(f"camelot表格大小: {camelot_df.shape}")
            # 检查是否提取了相同的列数
            if tabula_df.shape[1] != camelot_df.shape[1]:
                print(f"列数不同: tabula={tabula_df.shape[1]}, camelot={camelot_df.shape[1]}")
                print("这可能表明其中一个工具更好地识别了表格结构")
            # 检查是否提取了相同的行数
            if tabula_df.shape[0] != camelot_df.shape[0]:
                print(f"行数不同: tabula={tabula_df.shape[0]}, camelot={camelot_df.shape[0]}")
                print("这可能表明其中一个工具更好地识别了表格边界")
    return none
# 使用示例
compare_with_tabula("complex_table.pdf")

def compare_with_pdfplumber(pdf_path, page=0):
    """比较camelot与pdfplumber的表格提取能力"""
    try:
        import pdfplumber
    except importerror:
        print("请安装pdfplumber: pip install pdfplumber")
        return
    print("\n===== pdfplumber提取结果 =====")
    try:
        # 使用pdfplumber提取表格
        with pdfplumber.open(pdf_path) as pdf:
            if page < len(pdf.pages):
                plumber_page = pdf.pages[page]
                plumber_tables = plumber_page.extract_tables()
                print(f"检测到 {len(plumber_tables)} 个表格")
                if len(plumber_tables) > 0:
                    plumber_table = plumber_tables[0]
                    plumber_df = pd.dataframe(plumber_table[1:], columns=plumber_table[0])
                    print(f"表格维度: {plumber_df.shape}")
                    print("\n表格预览:")
                    print(plumber_df.head().to_string())
            else:
                print(f"页码 {page} 超出范围")
    except exception as e:
        print(f"pdfplumber提取出错: {str(e)}")
    print("\n===== camelot提取结果 =====")
    try:
        # 使用camelot提取表格
        camelot_tables = camelot.read_pdf(pdf_path, pages=str(page+1))  # camelot页码从1开始
        print(f"检测到 {len(camelot_tables)} 个表格")
        if len(camelot_tables) > 0:
            camelot_df = camelot_tables[0].df
            print(f"表格维度: {camelot_df.shape}")
            print(f"准确度: {camelot_tables[0].accuracy}")
            print("\n表格预览:")
            print(camelot_df.head().to_string())
    except exception as e:
        print(f"camelot提取出错: {str(e)}")
    return none
# 使用示例
compare_with_pdfplumber("annual_report.pdf")

def diagnose_extraction_issues(pdf_path, page='1'):
    """诊断和解决表格提取问题"""
    # 检查pdf是否可访问
    try:
        with open(pdf_path, 'rb') as f:
            pass
    except exception as e:
        print(f"无法访问pdf文件: {str(e)}")
        return
    # 检查是否为扫描pdf
    import fitz  # pymupdf
    try:
        doc = fitz.open(pdf_path)
        page_obj = doc[int(page) - 1]
        text = page_obj.get_text()
        if len(text.strip()) < 50:
            print("检测到可能是扫描pdf或图像pdf")
            print("建议: 使用ocr软件先将pdf转换为可搜索的pdf")
        # 检查页面旋转
        rotation = page_obj.rotation
        if rotation != 0:
            print(f"页面旋转了 {rotation} 度")
            print("建议: 使用pymupdf或其他工具先将pdf页面旋转到正常方向")
    except exception as e:
        print(f"检查pdf格式时出错: {str(e)}")
    # 尝试使用不同的提取方法
    print("\n尝试使用不同的camelot配置...")
    # 尝试lattice方法
    try:
        print("\n使用lattice方法:")
        lattice_tables = camelot.read_pdf(
            pdf_path,
            pages=page,
            flavor='lattice'
        )
        if len(lattice_tables) > 0:
            print(f"成功提取 {len(lattice_tables)} 个表格")
            print(f"准确度: {lattice_tables[0].accuracy}")
        else:
            print("未检测到表格")
            print("建议: 尝试调整line_scale参数和表格区域")
    except exception as e:
        print(f"lattice方法出错: {str(e)}")
    # 尝试stream方法
    try:
        print("\n使用stream方法:")
        stream_tables = camelot.read_pdf(
            pdf_path,
            pages=page,
            flavor='stream'
        )
        if len(stream_tables) > 0:
            print(f"成功提取 {len(stream_tables)} 个表格")
            print(f"准确度: {stream_tables[0].accuracy}")
        else:
            print("未检测到表格")
            print("建议: 尝试指定表格区域")
    except exception as e:
        print(f"stream方法出错: {str(e)}")
    # 建议
    print("\n==== 一般建议 ====")
    print("1. 如果两种方法都失败，尝试指定表格区域")
    print("2. 对于有明显表格线的pdf，优先使用lattice方法并调整line_scale")
    print("3. 对于无表格线的pdf，优先使用stream方法并调整边缘容忍度")
    print("4. 尝试将pdf页面转换为图像，然后使用opencv预处理后再提取")
    print("5. 如果是扫描pdf，考虑先使用ocr软件进行处理")
    return none
# 使用示例
diagnose_extraction_issues("problematic_report.pdf")

def common_errors_guide():
    """提供camelot常见错误的解决指南"""
    errors = {
        "importerror: no module named 'cv2'": {
            "原因": "缺少opencv依赖",
            "解决方案": "运行 pip install opencv-python"
        },
        "file does not exist": {
            "原因": "文件路径错误",
            "解决方案": "检查文件路径是否正确，包括大小写和空格"
        },
        "ocr engine not reachable": {
            "原因": "尝试使用ocr但未安装tesseract",
            "解决方案": "安装tesseract ocr并确保它在系统路径中"
        },
        "invalid page range specified": {
            "原因": "指定的页码超出了pdf范围",
            "解决方案": "确保页码在文档页数范围内，camelot的页码从1开始"
        },
        "unable to process background": {
            "原因": "在处理背景时遇到问题，通常与ghostscript有关",
            "解决方案": "检查ghostscript是否正确安装，或尝试禁用背景处理 (process_background=false)"
        },
        "no tables found on page": {
            "原因": "camelot无法在指定页面检测到表格",
            "解决方案": [
                "1. 尝试另一种提取方法 (lattice 或 stream)",
                "2. 手动指定表格区域",
                "3. 调整检测参数 (line_scale, edge_tol等)",
                "4. 检查pdf是否为扫描版，如果是请先使用ocr处理"
            ]
        }
    }
    print("==== camelot常见错误及解决方案 ====\n")
    for error, info in errors.items():
        print(f"错误: {error}")
        print(f"原因: {info['原因']}")
        if isinstance(info['解决方案'], list):
            print("解决方案:")
            for solution in info['解决方案']:
                print(f"  {solution}")
        else:
            print(f"解决方案: {info['解决方案']}")
        print()
    print("==== 一般性建议 ====")
    print("1. 始终使用最新版本的camelot和其依赖")
    print("2. 对于复杂表格，尝试分析表格结构后手动指定区域")
    print("3. 使用可视化工具验证表格边界检测")
    print("4. 对于大型pdf，考虑按批次处理页面")
    print("5. 如果一种提取方法失败，尝试另一种方法")
    return none
# 使用示例
common_errors_guide()

# lattice方法参数参考
lattice_params = {
    'line_scale': 15,       # 线条检测灵敏度，值越高检测越少的线
    'copy_text': [],        # 要从pdf复制的文本区域
    'shift_text': [],       # 要移动的文本区域
    'line_margin': 2,       # 线条检测间隔容忍度
    'joint_tol': 2,         # 连接点容忍度
    'threshold_blocksize': 15, # 自适应阈值的块大小
    'threshold_constant': -2,  # 自适应阈值的常数
    'iterations': 0,        # 形态学操作的迭代次数
    'resolution': 300,      # pdf-to-png转换的dpi
    'process_background': false, # 是否处理背景
    'table_areas': [],      # 表格区域列表，格式为[x1,y1,x2,y2]
    'table_regions': []     # 表格区域名称
}
# stream方法参数参考
stream_params = {
    'table_areas': [],      # 表格区域列表
    'columns': [],          # 列坐标
    'row_tol': 2,           # 行容忍度
    'column_tol': 0,        # 列容忍度
    'edge_tol': 50,         # 边缘容忍度
    'split_text': false,    # 是否拆分文本，实验性功能
    'flag_size': false,     # 是否标记文本大小
    'strip_text': '',       # 要从文本中删除的字符
    'edge_segment_counts': 50, # 用于检测表格边缘的线段数
    'min_columns': 1,       # 最小列数
    'max_columns': 0,       # 最大列数，0表示无限制
    'split_columns': false, # 是否拆分列，实验性功能
    'process_background': false, # 是否处理背景
    'line_margin': 2,       # 线条检测间隔容忍度
    'joint_tol': 2,         # 连接点容忍度
    'threshold_blocksize': 15, # 自适应阈值的块大小
    'threshold_constant': -2,  # 自适应阈值的常数
    'iterations': 0,        # 形态学操作的迭代次数
    'resolution': 300       # pdf-to-png转换的dpi
}