# 基本安裝
pip install camelot-py[cv]
# 如果需要PDF轉(zhuǎn)換功能
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"檢測(cè)到 {len(tables)} 個(gè)表格")
    # 表格基本信息
    for i, table in enumerate(tables):
        print(f"\n表格 #{i+1}:")
        print(f"頁(yè)碼: {table.page}")
        print(f"表格區(qū)域: {table.area}")
        print(f"維度: {table.shape}")
        print(f"準(zhǔn)確度分?jǐn)?shù): {table.accuracy}")
        print(f"空白率: {table.whitespace}")
        # 顯示表格前幾行
        print("\n表格預(yù)覽:")
        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')
    # 比較結(jié)果
    print(f"Stream方法: 檢測(cè)到 {len(stream_tables)} 個(gè)表格")
    print(f"Lattice方法: 檢測(cè)到 {len(lattice_tables)} 個(gè)表格")
    # 如果檢測(cè)到表格，比較第一個(gè)表格
    if len(stream_tables) > 0 and len(lattice_tables) > 0:
        # 獲取第一個(gè)表格
        stream_table = stream_tables[0]
        lattice_table = lattice_tables[0]
        # 比較準(zhǔn)確度和空白率
        print("\n準(zhǔn)確度和空白率比較:")
        print(f"Stream - 準(zhǔn)確度: {stream_table.accuracy}, 空白率: {stream_table.whitespace}")
        print(f"Lattice - 準(zhǔn)確度: {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):
    """使用精確區(qū)域坐標(biāo)提取表格"""
    if table_area is None:
        # 默認(rèn)值覆蓋整個(gè)頁(yè)面
        table_area = [0, 0, 100, 100]  # [x1, y1, x2, y2] 以百分比表示
    # 使用Stream方法提取指定區(qū)域的表格
    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"在指定區(qū)域檢測(cè)到 {len(tables)} 個(gè)表格")
    # 顯示第一個(gè)表格
    if len(tables) > 0:
        print("\n表格預(yù)覽:")
        print(tables[0].df.head())
    return tables
# 使用示例 - 提取頁(yè)面中間大約位置的表格
tables = extract_table_with_area("financial_report.pdf", table_area=[10, 30, 90, 70])

def extract_complex_tables(pdf_path, page='1'):
    """處理復(fù)雜表格的高級(jí)配置"""
    # 使用Lattice方法處理有邊框的復(fù)雜表格
    lattice_tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='lattice',
        line_scale=40,  # 調(diào)整線條檢測(cè)靈敏度
        process_background=True,  # 處理背景
        line_margin=2  # 線條間隔容忍度
    )
    # 使用Stream方法處理無(wú)邊框的復(fù)雜表格
    stream_tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='stream',
        edge_tol=500,  # 邊緣容忍度
        row_tol=10,    # 行容忍度
        column_tol=10  # 列容忍度
    )
    print(f"Lattice方法: 檢測(cè)到 {len(lattice_tables)} 個(gè)表格")
    print(f"Stream方法: 檢測(cè)到 {len(stream_tables)} 個(gè)表格")
    # 選擇最佳結(jié)果
    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'):
    """可視化表格提取過(guò)程，幫助調(diào)試和優(yōu)化"""
    # 提取表格
    tables = camelot.read_pdf(pdf_path, pages=page)
    # 檢查是否成功提取表格
    if len(tables) == 0:
        print("未檢測(cè)到表格")
        return
    # 獲取第一個(gè)表格
    table = tables[0]
    # 顯示表格
    print(f"表格形狀: {table.shape}")
    print(f"準(zhǔn)確度: {table.accuracy}")
    # 繪制表格結(jié)構(gòu)
    plot = table.plot(kind='grid')
    plt.title(f"表格網(wǎng)格結(jié)構(gòu) - 準(zhǔn)確度: {table.accuracy}")
    plt.tight_layout()
    plt.savefig('table_grid.png')
    plt.close()
    # 繪制表格單元格
    plot = table.plot(kind='contour')
    plt.title(f"表格單元格結(jié)構(gòu) - 空白率: {table.whitespace}")
    plt.tight_layout()
    plt.savefig('table_contour.png')
    plt.close()
    # 繪制表格線條（僅適用于lattice方法）
    if table.flavor == 'lattice':
        plot = table.plot(kind='line')
        plt.title("表格線條檢測(cè)")
        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提取的表格數(shù)據(jù)"""
    # 獲取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. 處理合并單元格的問(wèn)題（向下填充）
    df = df.fillna(method='ffill')
    # 4. 檢測(cè)并移除頁(yè)眉或頁(yè)腳（通常出現(xiàn)在第一行或最后一行）
    if df.shape[0] > 2:
        # 檢查第一行是否為頁(yè)眉
        if df.iloc[0].astype(str).str.contains('Page|頁(yè)碼|日期').any():
            df = df.iloc[1:]
        # 檢查最后一行是否為頁(yè)腳
        if df.iloc[-1].astype(str).str.contains('總計(jì)|合計(jì)|Total').any():
            df = df.iloc[:-1]
    # 5. 重置索引
    df = df.reset_index(drop=True)
    # 6. 設(shè)置第一行為列名（可選）
    # 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'):
    """合并多個(gè)表格"""
    if not tables or len(tables) == 0:
        return None
    dfs = [table.df for table in tables]
    if merge_method == 'vertical':
        # 垂直合并（適用于跨頁(yè)表格）
        merged_df = pd.concat(dfs, ignore_index=True)
    elif merge_method == 'horizontal':
        # 水平合并（適用于分列表格）
        merged_df = pd.concat(dfs, axis=1)
    else:
        raise ValueError("合并方法必須是 'vertical' 或 'horizontal'")
    # 清洗合并后的數(shù)據(jù)
    # 刪除完全相同的重復(fù)行（可能來(lái)自表格頁(yè)眉）
    merged_df = merged_df.drop_duplicates()
    return merged_df
# 使用示例 - 合并跨頁(yè)表格
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):
    """將表格數(shù)據(jù)轉(zhuǎn)換為適當(dāng)?shù)臄?shù)據(jù)類型"""
    # 創(chuàng)建DataFrame副本
    df = df.copy()
    for col in df.columns:
        # 嘗試將列轉(zhuǎn)換為數(shù)值型
        try:
            # 檢查列是否包含數(shù)字（帶有貨幣符號(hào)或千位分隔符）
            if df[col].str.contains(r'[$￥€￡]|\d,\d').any():
                # 移除貨幣符號(hào)和千位分隔符
                df[col] = df[col].replace(r'[$￥€￡,]', '', regex=True)
            # 嘗試轉(zhuǎn)換為數(shù)值型
            df[col] = pd.to_numeric(df[col])
            print(f"列 '{col}' 已轉(zhuǎn)換為數(shù)值型")
        except (ValueError, AttributeError):
            # 嘗試轉(zhuǎn)換為日期型
            try:
                df[col] = pd.to_datetime(df[col])
                print(f"列 '{col}' 已轉(zhuǎn)換為日期型")
            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'):
    """從年度報(bào)告中提取財(cái)務(wù)報(bào)表"""
    # 提取所有表格
    tables = camelot.read_pdf(
        pdf_path,
        pages=pages,
        flavor='stream',
        edge_tol=500,
        row_tol=10
    )
    print(f"共提取了 {len(tables)} 個(gè)表格")
    # 查找財(cái)務(wù)報(bào)表（通過(guò)關(guān)鍵詞）
    balance_sheet = None
    income_statement = None
    cash_flow = None
    for table in tables:
        df = table.df
        # 檢查表格是否包含特定關(guān)鍵詞
        text = ' '.join([' '.join(row) for row in df.values.tolist()])
        if any(term in text for term in ['資產(chǎn)負(fù)債表', 'Balance Sheet', '財(cái)務(wù)狀況表']):
            balance_sheet = clean_table_data(table)
            print("找到資產(chǎn)負(fù)債表")
        elif any(term in text for term in ['利潤(rùn)表', 'Income Statement', '損益表']):
            income_statement = clean_table_data(table)
            print("找到利潤(rùn)表")
        elif any(term in text for term in ['現(xiàn)金流量表', 'Cash Flow']):
            cash_flow = clean_table_data(table)
            print("找到現(xiàn)金流量表")
    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'):
    """批量處理多個(gè)PDF文件，提取所有表格"""
    import os
    from pathlib import Path
    # 創(chuàng)建輸出文件夾
    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}")
        # 創(chuàng)建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)} 個(gè)表格")
            # 保存每個(gè)表格為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')
            # 記錄結(jié)果
            results[pdf_file] = {
                'status': 'success',
                'tables_count': len(tables),
                'output_folder': pdf_output_folder
            }
        except Exception as e:
            print(f"處理 {pdf_file} 時(shí)出錯(cuò): {str(e)}")
            results[pdf_file] = {
                'status': 'error',
                'error_message': str(e)
            }
    # 匯總報(bào)告
    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'):
    """從提取的表格創(chuàng)建簡(jiǎn)單的交互式儀表板"""
    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("沒(méi)有表格數(shù)據(jù)可用于創(chuàng)建儀表板")
        return
    # 為了簡(jiǎn)單起見(jiàn)，使用第一個(gè)表格
    df = clean_table_data(tables[0])
    # 如果所有列都是字符串，嘗試將其中一些轉(zhuǎn)換為數(shù)值
    df = convert_table_datatypes(df)
    # 創(chuàng)建儀表板 HTML
    with open(output_html, 'w', encoding='utf-8') as f:
        f.write("<html><head>")
        f.write("<title>PDF表格數(shù)據(jù)儀表板</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表格數(shù)據(jù)儀表板</h1>")
        # 添加表格
        f.write("<div class='chart'>")
        f.write("<h2>提取的表格數(shù)據(jù)</h2>")
        f.write(df.to_html(classes='dataframe', index=False))
        f.write("</div>")
        # 如果有數(shù)值型列，創(chuàng)建圖表
        numeric_cols = df.select_dtypes(include=['number']).columns
        if len(numeric_cols) > 0:
            # 選擇第一個(gè)數(shù)值列創(chuàng)建圖表
            value_col = numeric_cols[0]
            # 尋找一個(gè)可能的類別列
            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:
                # 創(chuàng)建條形圖
                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>")
                # 創(chuàng)建餅圖
                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"儀表板已創(chuàng)建: {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'):
    """優(yōu)化表格檢測(cè)參數(shù)，嘗試不同設(shè)置并評(píng)估結(jié)果"""
    # 定義不同的參數(shù)組合
    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 = []
    # 測(cè)試Stream方法的不同配置
    print("測(cè)試Stream方法...")
    for config in stream_configs:
        try:
            tables = camelot.read_pdf(
                pdf_path,
                pages=page,
                flavor='stream',
                **config
            )
            # 評(píng)估結(jié)果
            if len(tables) > 0:
                accuracy = tables[0].accuracy
                whitespace = tables[0].whitespace
                print(f"配置 {config}: 準(zhǔn)確度={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} 出錯(cuò): {str(e)}")
    # 測(cè)試Lattice方法的不同配置
    print("\n測(cè)試Lattice方法...")
    for config in lattice_configs:
        try:
            tables = camelot.read_pdf(
                pdf_path,
                pages=page,
                flavor='lattice',
                **config
            )
            # 評(píng)估結(jié)果
            if len(tables) > 0:
                accuracy = tables[0].accuracy
                whitespace = tables[0].whitespace
                print(f"配置 {config}: 準(zhǔn)確度={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} 出錯(cuò): {str(e)}")
    # 找出最佳配置
    if results:
        # 按準(zhǔn)確度排序
        best_result = sorted(results, key=lambda x: x['accuracy'], reverse=True)[0]
        print(f"\n最佳配置: {best_result['flavor']} 方法, 參數(shù): {best_result['config']}")
        print(f"準(zhǔn)確度: {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中提取表格（需要預(yù)處理）"""
    import cv2
    import numpy as np
    import tempfile
    from pdf2image import convert_from_path
    # 轉(zhuǎn)換PDF頁(yè)面為圖像
    images = convert_from_path(pdf_path, first_page=int(page), last_page=int(page))
    if not images:
        print("無(wú)法轉(zhuǎn)換PDF頁(yè)面為圖像")
        return None
    # 獲取第一個(gè)頁(yè)面圖像
    image = np.array(images[0])
    # 圖像預(yù)處理
    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"掃描頁(yè)面已預(yù)處理并保存為臨時(shí)圖像: {temp_image_path}")
    # 從圖像中提取表格
    tables = camelot.read_pdf(
        pdf_path,
        pages=page,
        flavor='lattice',
        process_background=True,
        line_scale=150
    )
    print(f"從掃描PDF提取了 {len(tables)} 個(gè)表格")
    # 可選: 移除臨時(shí)文件
    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()
    # 檢測(cè)并處理垂直合并的單元格
    for col in df.columns:
        # 在連續(xù)的空單元格上向下填充值
        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))
            # 填充檢測(cè)到的合并單元格
            for idx, value in fill_values:
                df.at[idx, col] = value
    # 檢測(cè)并處理水平合并的單元格
    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] != '':
                    # 可能是水平合并，從左側(cè)單元格填充
                    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分析提取的表格數(shù)據(jù)"""
    # 清潔數(shù)據(jù)
    df = clean_table_data(table)
    # 轉(zhuǎn)換數(shù)據(jù)類型
    df = convert_table_datatypes(df)
    # 基本統(tǒng)計(jì)分析
    print("\n==== 基本統(tǒng)計(jì)分析 ====")
    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]:  # 只顯示前三個(gè)類別列
        value_counts = df[col].value_counts()
        print(f"\n{col}:")
        print(value_counts.head())
    # 相關(guān)性分析
    if len(numeric_cols) >= 2:
        print("\n==== 相關(guā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可視化表格數(shù)據(jù)"""
    import matplotlib.pyplot as plt
    import seaborn as sns
    # 設(shè)置樣式
    sns.set(style="whitegrid")
    # 清潔并轉(zhuǎn)換數(shù)據(jù)
    df = clean_table_data(table)
    df = convert_table_datatypes(df)
    # 獲取數(shù)值列
    numeric_cols = df.select_dtypes(include=['number']).columns
    if len(numeric_cols) == 0:
        print("沒(méi)有數(shù)值列可供可視化")
        return
    # 1. 熱圖 - 相關(guān)性
    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('相關(guān)性熱圖', fontsize=15)
        plt.tight_layout()
        plt.savefig(f'{output_prefix}_correlation.png')
        plt.close()
    # 2. 條形圖 - 數(shù)值分布
    for col in numeric_cols[:3]:  # 只處理前三個(gè)數(shù)值列
        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('數(shù)值列箱線圖', fontsize=15)
    plt.tight_layout()
    plt.savefig(f'{output_prefix}_boxplot.png')
    plt.close()
    # 4. 散點(diǎn)圖矩陣 - 變量關(guān)系
    if len(numeric_cols) >= 2 and len(df) > 5:
        sns.pairplot(df[numeric_cols])
        plt.suptitle('散點(diǎn)圖矩陣', 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'):
    """將提取的表格導(dǎo)出到Excel工作簿的不同工作表"""
    import pandas as pd
    # 檢查是否有表格
    if not tables or len(tables) == 0:
        print("沒(méi)有表格可導(dǎo)出")
        return None
    # 創(chuàng)建Excel Writer對(duì)象
    with pd.ExcelWriter(output_path, engine='xlsxwriter') as writer:
        workbook = writer.book
        # 創(chuàng)建表格樣式
        header_format = workbook.add_format({
            'bold': True,
            'text_wrap': True,
            'valign': 'top',
            'fg_color': '#D7E4BC',
            'border': 1
        })
        cell_format = workbook.add_format({
            'border': 1
        })
        # 為每個(gè)表格創(chuàng)建工作表
        for i, table in enumerate(tables):
            # 清潔數(shù)據(jù)
            df = clean_table_data(table)
            # 寫(xiě)入數(shù)據(jù)
            sheet_name = f'Table_{i+1}'
            df.to_excel(writer, sheet_name=sheet_name, index=False)
            # 獲取工作表對(duì)象
            worksheet = writer.sheets[sheet_name]
            # 設(shè)置列寬
            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)
            # 設(shè)置表格格式
            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
            })
            # 添加表格元數(shù)據(jù)
            worksheet.write(df.shape[0] + 2, 0, f"頁(yè)碼: {table.page}")
            worksheet.write(df.shape[0] + 3, 0, f"表格區(qū)域: {table.area}")
            worksheet.write(df.shape[0] + 4, 0, f"準(zhǔn)確度分?jǐn)?shù): {table.accuracy}")
    print(f"表格已導(dǎo)出至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文檔以節(jié)省內(nèi)存"""
    import os
    from pathlib import Path
    # 創(chuàng)建輸出文件夾
    Path(output_folder).mkdir(exist_ok=True)
    # 首先獲取PDF頁(yè)數(shù)
    with pdfplumber.open(pdf_path) as pdf:
        total_pages = len(pdf.pages)
    print(f"PDF共有 {total_pages} 頁(yè)")
    # 分批處理頁(yè)面
    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"處理頁(yè)面范圍: {page_range}")
        try:
            # 提取當(dāng)前批次的表格
            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"從頁(yè)面 {page_range} 提取了 {batch_tables_count} 個(gè)表格")
            # 保存這一批次的表格
            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')
            # 顯式釋放內(nèi)存
            tables = None
            import gc
            gc.collect()
        except Exception as e:
            print(f"處理頁(yè)面 {page_range} 時(shí)出錯(cuò): {str(e)}")
    print(f"處理完成，共提取了 {all_tables_count} 個(gè)表格，保存到 {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'):
    """調(diào)優(yōu)Camelot的性能參數(shù)"""
    import time
    import psutil
    import os
    def measure_performance(func, *args, **kwargs):
        """測(cè)量函數(shù)的執(zhí)行時(shí)間和內(nèi)存使用情況"""
        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
    # 測(cè)試不同的參數(shù)組合
    configs = [
        {
            'name': '默認(rèn)配置',
            'params': {}
        },
        {
            'name': '啟用后臺(tái)處理',
            'params': {'process_background': True}
        },
        {
            'name': '禁用線條檢測(cè)',
            'params': {'line_scale': 0}
        },
        {
            'name': '提高線條檢測(cè)靈敏度',
            'params': {'line_scale': 80}
        }
    ]
    results = []
    for config in configs:
        print(f"\n測(cè)試配置: {config['name']}")
        # 測(cè)試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 - 時(shí)間: {lattice_time:.2f}秒, 內(nèi)存: {lattice_mem:.2f}MB, 準(zhǔn)確度: {lattice_tables[0].accuracy if len(lattice_tables) > 0 else 0}")
        except Exception as e:
            print(f"  Lattice方法出錯(cuò): {str(e)}")
        # 測(cè)試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 - 時(shí)間: {stream_time:.2f}秒, 內(nèi)存: {stream_mem:.2f}MB, 準(zhǔn)確度: {stream_tables[0].accuracy if len(stream_tables) > 0 else 0}")
        except Exception as e:
            print(f"  Stream方法出錯(cuò): {str(e)}")
    # 查找最佳性能配置
    if results:
        # 按準(zhǔn)確度排序
        accuracy_best = sorted(results, key=lambda x: x['accuracy'], reverse=True)[0]
        print(f"\n最高準(zhǔn)確度配置: {accuracy_best['config_name']} / {accuracy_best['method']}")
        print(f"  準(zhǔn)確度: {accuracy_best['accuracy']:.2f}, 耗時(shí): {accuracy_best['time']:.2f}秒")
        # 按時(shí)間排序
        time_best = sorted(results, key=lambda x: x['time'])[0]
        print(f"\n最快配置: {time_best['config_name']} / {time_best['method']}")
        print(f"  耗時(shí): {time_best['time']:.2f}秒, 準(zhǔn)確度: {time_best['accuracy']:.2f}")
        # 按內(nèi)存使用排序
        memory_best = sorted(results, key=lambda x: x['memory'])[0]
        print(f"\n最低內(nèi)存配置: {memory_best['config_name']} / {memory_best['method']}")
        print(f"  內(nèi)存: {memory_best['memory']:.2f}MB, 準(zhǔn)確度: {memory_best['accuracy']:.2f}")
        # 綜合考慮速度和準(zhǔn)確度的最佳配置
        balanced = sorted(results, key=lambda x: (1/x['accuracy']) * x['time'])[0]
        print(f"\n平衡配置: {balanced['config_name']} / {balanced['method']}")
        print(f"  準(zhǔn)確度: {balanced['accuracy']:.2f}, 耗時(shí): {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提取結(jié)果 =====")
    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無(wú)法識(shí)別表格結(jié)構(gòu)，只能提取純文本")
            else:
                print(f"頁(yè)碼 {page} 超出范圍")
    except Exception as e:
        print(f"PyPDF2提取出錯(cuò): {str(e)}")
    print("\n===== Camelot提取結(jié)果 =====")
    try:
        # 使用Camelot提取表格
        tables = camelot.read_pdf(pdf_path, pages=str(page+1))  # Camelot頁(yè)碼從1開(kāi)始
        print(f"檢測(cè)到 {len(tables)} 個(gè)表格")
        if len(tables) > 0:
            table = tables[0]
            print(f"表格維度: {table.shape}")
            print(f"準(zhǔn)確度: {table.accuracy}")
            print("\n表格預(yù)覽:")
            print(table.df.head().to_string())
            print("\nCamelot可以識(shí)別表格結(jié)構(gòu)，保留行列關(guān)系")
    except Exception as e:
        print(f"Camelot提取出錯(cuò): {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("請(qǐng)安裝tabula-py: pip install tabula-py")
        return
    print("\n===== Tabula提取結(jié)果 =====")
    try:
        # 使用Tabula提取表格
        tabula_tables = tabula.read_pdf(pdf_path, pages=page)
        print(f"檢測(cè)到 {len(tabula_tables)} 個(gè)表格")
        if len(tabula_tables) > 0:
            tabula_df = tabula_tables[0]
            print(f"表格維度: {tabula_df.shape}")
            print("\n表格預(yù)覽:")
            print(tabula_df.head().to_string())
    except Exception as e:
        print(f"Tabula提取出錯(cuò): {str(e)}")
    print("\n===== Camelot提取結(jié)果 =====")
    try:
        # 使用Camelot提取表格
        camelot_tables = camelot.read_pdf(pdf_path, pages=page)
        print(f"檢測(cè)到 {len(camelot_tables)} 個(gè)表格")
        if len(camelot_tables) > 0:
            camelot_df = camelot_tables[0].df
            print(f"表格維度: {camelot_df.shape}")
            print(f"準(zhǔn)確度: {camelot_tables[0].accuracy}")
            print("\n表格預(yù)覽:")
            print(camelot_df.head().to_string())
    except Exception as e:
        print(f"Camelot提取出錯(cuò): {str(e)}")
    # 比較結(jié)果
    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===== 比較結(jié)果 =====")
            print(f"Tabula表格大小: {tabula_df.shape}")
            print(f"Camelot表格大小: {camelot_df.shape}")
            # 檢查是否提取了相同的列數(shù)
            if tabula_df.shape[1] != camelot_df.shape[1]:
                print(f"列數(shù)不同: Tabula={tabula_df.shape[1]}, Camelot={camelot_df.shape[1]}")
                print("這可能表明其中一個(gè)工具更好地識(shí)別了表格結(jié)構(gòu)")
            # 檢查是否提取了相同的行數(shù)
            if tabula_df.shape[0] != camelot_df.shape[0]:
                print(f"行數(shù)不同: Tabula={tabula_df.shape[0]}, Camelot={camelot_df.shape[0]}")
                print("這可能表明其中一個(gè)工具更好地識(shí)別了表格邊界")
    return None
# 使用示例
compare_with_tabula("complex_table.pdf")

def compare_with_pdfplumber(pdf_path, page=0):
    """比較Camelot與pdfplumber的表格提取能力"""
    try:
        import pdfplumber
    except ImportError:
        print("請(qǐng)安裝pdfplumber: pip install pdfplumber")
        return
    print("\n===== pdfplumber提取結(jié)果 =====")
    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"檢測(cè)到 {len(plumber_tables)} 個(gè)表格")
                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表格預(yù)覽:")
                    print(plumber_df.head().to_string())
            else:
                print(f"頁(yè)碼 {page} 超出范圍")
    except Exception as e:
        print(f"pdfplumber提取出錯(cuò): {str(e)}")
    print("\n===== Camelot提取結(jié)果 =====")
    try:
        # 使用Camelot提取表格
        camelot_tables = camelot.read_pdf(pdf_path, pages=str(page+1))  # Camelot頁(yè)碼從1開(kāi)始
        print(f"檢測(cè)到 {len(camelot_tables)} 個(gè)表格")
        if len(camelot_tables) > 0:
            camelot_df = camelot_tables[0].df
            print(f"表格維度: {camelot_df.shape}")
            print(f"準(zhǔn)確度: {camelot_tables[0].accuracy}")
            print("\n表格預(yù)覽:")
            print(camelot_df.head().to_string())
    except Exception as e:
        print(f"Camelot提取出錯(cuò): {str(e)}")
    return None
# 使用示例
compare_with_pdfplumber("annual_report.pdf")

def diagnose_extraction_issues(pdf_path, page='1'):
    """診斷和解決表格提取問(wèn)題"""
    # 檢查PDF是否可訪問(wèn)
    try:
        with open(pdf_path, 'rb') as f:
            pass
    except Exception as e:
        print(f"無(wú)法訪問(wèn)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("檢測(cè)到可能是掃描PDF或圖像PDF")
            print("建議: 使用OCR軟件先將PDF轉(zhuǎn)換為可搜索的PDF")
        # 檢查頁(yè)面旋轉(zhuǎn)
        rotation = page_obj.rotation
        if rotation != 0:
            print(f"頁(yè)面旋轉(zhuǎn)了 {rotation} 度")
            print("建議: 使用PyMuPDF或其他工具先將PDF頁(yè)面旋轉(zhuǎn)到正常方向")
    except Exception as e:
        print(f"檢查PDF格式時(shí)出錯(cuò): {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)} 個(gè)表格")
            print(f"準(zhǔn)確度: {lattice_tables[0].accuracy}")
        else:
            print("未檢測(cè)到表格")
            print("建議: 嘗試調(diào)整line_scale參數(shù)和表格區(qū)域")
    except Exception as e:
        print(f"Lattice方法出錯(cuò): {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)} 個(gè)表格")
            print(f"準(zhǔn)確度: {stream_tables[0].accuracy}")
        else:
            print("未檢測(cè)到表格")
            print("建議: 嘗試指定表格區(qū)域")
    except Exception as e:
        print(f"Stream方法出錯(cuò): {str(e)}")
    # 建議
    print("\n==== 一般建議 ====")
    print("1. 如果兩種方法都失敗，嘗試指定表格區(qū)域")
    print("2. 對(duì)于有明顯表格線的PDF，優(yōu)先使用Lattice方法并調(diào)整line_scale")
    print("3. 對(duì)于無(wú)表格線的PDF，優(yōu)先使用Stream方法并調(diào)整邊緣容忍度")
    print("4. 嘗試將PDF頁(yè)面轉(zhuǎn)換為圖像，然后使用OpenCV預(yù)處理后再提取")
    print("5. 如果是掃描PDF，考慮先使用OCR軟件進(jìn)行處理")
    return None
# 使用示例
diagnose_extraction_issues("problematic_report.pdf")

def common_errors_guide():
    """提供Camelot常見(jiàn)錯(cuò)誤的解決指南"""
    errors = {
        "ImportError: No module named 'cv2'": {
            "原因": "缺少OpenCV依賴",
            "解決方案": "運(yùn)行 pip install opencv-python"
        },
        "File does not exist": {
            "原因": "文件路徑錯(cuò)誤",
            "解決方案": "檢查文件路徑是否正確，包括大小寫(xiě)和空格"
        },
        "OCR engine not reachable": {
            "原因": "嘗試使用OCR但未安裝Tesseract",
            "解決方案": "安裝Tesseract OCR并確保它在系統(tǒng)路徑中"
        },
        "Invalid page range specified": {
            "原因": "指定的頁(yè)碼超出了PDF范圍",
            "解決方案": "確保頁(yè)碼在文檔頁(yè)數(shù)范圍內(nèi)，Camelot的頁(yè)碼從1開(kāi)始"
        },
        "Unable to process background": {
            "原因": "在處理背景時(shí)遇到問(wèn)題，通常與GhostScript有關(guān)",
            "解決方案": "檢查GhostScript是否正確安裝，或嘗試禁用背景處理 (process_background=False)"
        },
        "No tables found on page": {
            "原因": "Camelot無(wú)法在指定頁(yè)面檢測(cè)到表格",
            "解決方案": [
                "1. 嘗試另一種提取方法 (lattice 或 stream)",
                "2. 手動(dòng)指定表格區(qū)域",
                "3. 調(diào)整檢測(cè)參數(shù) (line_scale, edge_tol等)",
                "4. 檢查PDF是否為掃描版，如果是請(qǐng)先使用OCR處理"
            ]
        }
    }
    print("==== Camelot常見(jiàn)錯(cuò)誤及解決方案 ====\n")
    for error, info in errors.items():
        print(f"錯(cuò)誤: {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. 對(duì)于復(fù)雜表格，嘗試分析表格結(jié)構(gòu)后手動(dòng)指定區(qū)域")
    print("3. 使用可視化工具驗(yàn)證表格邊界檢測(cè)")
    print("4. 對(duì)于大型PDF，考慮按批次處理頁(yè)面")
    print("5. 如果一種提取方法失敗，嘗試另一種方法")
    return None
# 使用示例
common_errors_guide()

# Lattice方法參數(shù)參考
lattice_params = {
    'line_scale': 15,       # 線條檢測(cè)靈敏度，值越高檢測(cè)越少的線
    'copy_text': [],        # 要從PDF復(fù)制的文本區(qū)域
    'shift_text': [],       # 要移動(dòng)的文本區(qū)域
    'line_margin': 2,       # 線條檢測(cè)間隔容忍度
    'joint_tol': 2,         # 連接點(diǎn)容忍度
    'threshold_blocksize': 15, # 自適應(yīng)閾值的塊大小
    'threshold_constant': -2,  # 自適應(yīng)閾值的常數(shù)
    'iterations': 0,        # 形態(tài)學(xué)操作的迭代次數(shù)
    'resolution': 300,      # PDF-to-PNG轉(zhuǎn)換的DPI
    'process_background': False, # 是否處理背景
    'table_areas': [],      # 表格區(qū)域列表，格式為[x1,y1,x2,y2]
    'table_regions': []     # 表格區(qū)域名稱
}
# Stream方法參數(shù)參考
stream_params = {
    'table_areas': [],      # 表格區(qū)域列表
    'columns': [],          # 列坐標(biāo)
    'row_tol': 2,           # 行容忍度
    'column_tol': 0,        # 列容忍度
    'edge_tol': 50,         # 邊緣容忍度
    'split_text': False,    # 是否拆分文本，實(shí)驗(yàn)性功能
    'flag_size': False,     # 是否標(biāo)記文本大小
    'strip_text': '',       # 要從文本中刪除的字符
    'edge_segment_counts': 50, # 用于檢測(cè)表格邊緣的線段數(shù)
    'min_columns': 1,       # 最小列數(shù)
    'max_columns': 0,       # 最大列數(shù)，0表示無(wú)限制
    'split_columns': False, # 是否拆分列，實(shí)驗(yàn)性功能
    'process_background': False, # 是否處理背景
    'line_margin': 2,       # 線條檢測(cè)間隔容忍度
    'joint_tol': 2,         # 連接點(diǎn)容忍度
    'threshold_blocksize': 15, # 自適應(yīng)閾值的塊大小
    'threshold_constant': -2,  # 自適應(yīng)閾值的常數(shù)
    'iterations': 0,        # 形態(tài)學(xué)操作的迭代次數(shù)
    'resolution': 300       # PDF-to-PNG轉(zhuǎn)換的DPI
}