pip install PyPDF2 pdfminer.six pandas

import os
import re
from datetime import datetime
from typing import List, Dict, Optional, Tuple

import pandas as pd
from PyPDF2 import PdfReader
from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextContainer

class PDFTextExtractor:
    """專業(yè)的PDF文本提取工具"""
    
    def __init__(self, output_dir: str = "output"):
        """
        初始化提取工具
        
        :param output_dir: 輸出目錄路徑
        """
        self.output_dir = output_dir
        os.makedirs(self.output_dir, exist_ok=True)
        
        # 文本清理正則表達式
        self.clean_patterns = [
            (r'\s+', ' '),  # 合并多個空白字符
            (r'\n{3,}', '\n\n'),  # 限制連續(xù)換行
            (r'[^\x00-\x7F]+', ' '),  # 移除非ASCII字符
        ]
    
    def extract_metadata(self, pdf_path: str) -> Dict[str, str]:
        """提取PDF元數(shù)據(jù)"""
        with open(pdf_path, 'rb') as file:
            reader = PdfReader(file)
            meta = reader.metadata
            
            return {
                'file_name': os.path.basename(pdf_path),
                'title': meta.get('/Title', ''),
                'author': meta.get('/Author', ''),
                'creator': meta.get('/Creator', ''),
                'producer': meta.get('/Producer', ''),
                'created_date': meta.get('/CreationDate', ''),
                'modified_date': meta.get('/ModDate', ''),
                'page_count': len(reader.pages),
                'extraction_date': datetime.now().isoformat()
            }
    
    def clean_text(self, text: str) -> str:
        """清理和規(guī)范化提取的文本"""
        for pattern, replacement in self.clean_patterns:
            text = re.sub(pattern, replacement, text)
        return text.strip()
    
    def extract_text_from_page(self, page_layout) -> str:
        """從單個頁面布局提取文本"""
        page_text = []
        for element in page_layout:
            if isinstance(element, LTTextContainer):
                text = element.get_text()
                if text.strip():
                    page_text.append(self.clean_text(text))
        return '\n'.join(page_text)
    
    def extract_toc(self, pdf_path: str) -> List[Dict[str, str]]:
        """嘗試提取文檔目錄結(jié)構(gòu)"""
        toc = []
        try:
            with open(pdf_path, 'rb') as file:
                reader = PdfReader(file)
                if reader.outline:
                    for item in reader.outline:
                        if isinstance(item, list):
                            continue  # 跳過子項處理簡化示例
                        toc.append({
                            'title': item.title,
                            'page': reader.get_destination_page_number(item) + 1
                        })
        except Exception:
            pass  # 目錄提取失敗不影響主流程
        return toc
    
    def process_pdf(self, pdf_path: str) -> Dict[str, any]:
        """處理單個PDF文件"""
        if not os.path.isfile(pdf_path):
            raise FileNotFoundError(f"PDF文件不存在: {pdf_path}")
        
        result = {
            'metadata': self.extract_metadata(pdf_path),
            'toc': self.extract_toc(pdf_path),
            'pages': []
        }
        
        # 使用pdfminer逐頁提取文本
        for i, page_layout in enumerate(extract_pages(pdf_path)):
            page_text = self.extract_text_from_page(page_layout)
            if page_text:
                result['pages'].append({
                    'page_number': i + 1,
                    'content': page_text,
                    'char_count': len(page_text),
                    'word_count': len(page_text.split())
                })
        
        return result
    
    def batch_process(self, pdf_files: List[str]) -> List[Dict[str, any]]:
        """批量處理多個PDF文件"""
        results = []
        for pdf_file in pdf_files:
            try:
                print(f"正在處理: {os.path.basename(pdf_file)}...")
                results.append(self.process_pdf(pdf_file))
            except Exception as e:
                print(f"處理 {pdf_file} 時出錯: {str(e)}")
                results.append({
                    'file': pdf_file,
                    'error': str(e)
                })
        return results
    
    def export_to_csv(self, data: List[Dict[str, any]], prefix: str = "pdf_export"):
        """將提取結(jié)果導(dǎo)出為CSV"""
        # 準備元數(shù)據(jù)表格
        meta_data = [item['metadata'] for item in data if 'metadata' in item]
        meta_df = pd.DataFrame(meta_data)
        
        # 準備頁面內(nèi)容表格
        page_data = []
        for doc in data:
            if 'pages' in doc:
                for page in doc['pages']:
                    page_entry = {
                        'file_name': doc['metadata']['file_name'],
                        **page
                    }
                    page_data.append(page_entry)
        pages_df = pd.DataFrame(page_data)
        
        # 生成時間戳文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        meta_file = os.path.join(self.output_dir, f"{prefix}_metadata_{timestamp}.csv")
        pages_file = os.path.join(self.output_dir, f"{prefix}_pages_{timestamp}.csv")
        
        # 保存文件
        meta_df.to_csv(meta_file, index=False, encoding='utf-8-sig')
        pages_df.to_csv(pages_file, index=False, encoding='utf-8-sig')
        
        return meta_file, pages_file

# 使用示例
if __name__ == "__main__":
    # 初始化提取器
    extractor = PDFTextExtractor()
    
    # 示例PDF文件列表（替換為實際路徑）
    sample_files = [
        "documents/sample1.pdf",
        "documents/sample2.pdf"
    ]
    
    # 批量處理并導(dǎo)出
    results = extractor.batch_process(sample_files)
    meta_csv, pages_csv = extractor.export_to_csv(results)
    
    print(f"\n處理完成！\n元數(shù)據(jù)已保存至: {meta_csv}\n頁面內(nèi)容已保存至: {pages_csv}")

class PDFTextExtractor:
    def __init__(self, output_dir: str = "output"):
        self.output_dir = output_dir
        os.makedirs(self.output_dir, exist_ok=True)
        
        # 文本清理正則表達式
        self.clean_patterns = [
            (r'\s+', ' '),  # 合并多個空白字符
            (r'\n{3,}', '\n\n'),  # 限制連續(xù)換行
            (r'[^\x00-\x7F]+', ' '),  # 移除非ASCII字符
        ]

def extract_metadata(self, pdf_path: str) -> Dict[str, str]:
    with open(pdf_path, 'rb') as file:
        reader = PdfReader(file)
        meta = reader.metadata
        
        return {
            'file_name': os.path.basename(pdf_path),
            'title': meta.get('/Title', ''),
            'author': meta.get('/Author', ''),
            # ...其他元數(shù)據(jù)字段
        }

def clean_text(self, text: str) -> str:
    for pattern, replacement in self.clean_patterns:
        text = re.sub(pattern, replacement, text)
    return text.strip()

def extract_text_from_page(self, page_layout) -> str:
    page_text = []
    for element in page_layout:
        if isinstance(element, LTTextContainer):
            text = element.get_text()
            if text.strip():
                page_text.append(self.clean_text(text))
    return '\n'.join(page_text)

def batch_process(self, pdf_files: List[str]) -> List[Dict[str, any]]:
    results = []
    for pdf_file in pdf_files:
        try:
            results.append(self.process_pdf(pdf_file))
        except Exception as e:
            results.append({'file': pdf_file, 'error': str(e)})
    return results
 
def export_to_csv(self, data: List[Dict[str, any]], prefix: str = "pdf_export"):
    # 準備元數(shù)據(jù)和頁面內(nèi)容DataFrame
    meta_df = pd.DataFrame([item['metadata'] for item in data if 'metadata' in item])
    pages_df = pd.DataFrame([
        {'file_name': doc['metadata']['file_name'], **page}
        for doc in data if 'pages' in doc
        for page in doc['pages']
    ])
    
    # 保存CSV文件
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    meta_file = os.path.join(self.output_dir, f"{prefix}_metadata_{timestamp}.csv")
    pages_file = os.path.join(self.output_dir, f"{prefix}_pages_{timestamp}.csv")
    
    meta_df.to_csv(meta_file, index=False, encoding='utf-8-sig')
    pages_df.to_csv(pages_file, index=False, encoding='utf-8-sig')

try:
    import pytesseract
    from pdf2image import convert_from_path
    
    def extract_text_with_ocr(self, pdf_path: str) -> Dict[str, any]:
        """使用OCR處理圖像型PDF"""
        images = convert_from_path(pdf_path)
        ocr_results = []
        
        for i, image in enumerate(images):
            text = pytesseract.image_to_string(image)
            if text.strip():
                ocr_results.append({
                    'page_number': i + 1,
                    'content': self.clean_text(text),
                    'method': 'OCR'
                })
        
        return {
            'metadata': self.extract_metadata(pdf_path),
            'pages': ocr_results
        }
except ImportError:
    pass

try:
    import camelot
    
    def extract_tables(self, pdf_path: str) -> List[Dict[str, any]]:
        """提取PDF中的表格數(shù)據(jù)"""
        tables = camelot.read_pdf(pdf_path, flavor='lattice')
        return [
            {
                'page': table.page,
                'order': table.order,
                'df': table.df.to_dict(),
                'accuracy': table.accuracy
            }
            for table in tables
        ]
except ImportError:
    pass

import sqlite3

def export_to_sqlite(self, data: List[Dict[str, any]], db_name: str = "pdf_data.db"):
    """將提取結(jié)果導(dǎo)出到SQLite數(shù)據(jù)庫"""
    conn = sqlite3.connect(os.path.join(self.output_dir, db_name))
    
    # 創(chuàng)建元數(shù)據(jù)表
    conn.execute('''
    CREATE TABLE IF NOT EXISTS pdf_metadata (
        file_name TEXT PRIMARY KEY,
        title TEXT,
        author TEXT,
        page_count INTEGER,
        created_date TEXT
    )
    ''')
    
    # 創(chuàng)建頁面內(nèi)容表
    conn.execute('''
    CREATE TABLE IF NOT EXISTS pdf_pages (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        file_name TEXT,
        page_number INTEGER,
        content TEXT,
        char_count INTEGER,
        word_count INTEGER
    )
    ''')
    
    # 插入數(shù)據(jù)
    for doc in data:
        if 'metadata' in doc:
            meta = doc['metadata']
            conn.execute(
                'INSERT OR REPLACE INTO pdf_metadata VALUES (?,?,?,?,?)',
                (meta['file_name'], meta['title'], meta['author'], 
                 meta['page_count'], meta['created_date'])
        
        if 'pages' in doc:
            for page in doc['pages']:
                conn.execute(
                    'INSERT INTO pdf_pages VALUES (NULL,?,?,?,?,?)',
                    (doc['metadata']['file_name'], page['page_number'],
                     page['content'], page['char_count'], page['word_count'])
    
    conn.commit()
    conn.close()

from concurrent.futures import ThreadPoolExecutor

def parallel_batch_process(self, pdf_files: List[str], workers: int = 4):
    with ThreadPoolExecutor(max_workers=workers) as executor:
        return list(executor.map(self.process_pdf, pdf_files))

import unittest
import shutil
from pathlib import Path

class TestPDFTextExtractor(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.test_dir = Path("test_output")
        cls.test_dir.mkdir(exist_ok=True)
        
        # 創(chuàng)建測試PDF (實際使用中應(yīng)準備樣例文件)
        cls.sample_pdf = cls.test_dir / "sample.pdf"
        # 這里應(yīng)添加PDF生成代碼或使用預(yù)準備的測試文件
    
    def test_metadata_extraction(self):
        extractor = PDFTextExtractor(self.test_dir)
        result = extractor.process_pdf(self.sample_pdf)
        self.assertIn('metadata', result)
        self.assertGreater(result['metadata']['page_count'], 0)
    
    def test_text_extraction(self):
        extractor = PDFTextExtractor(self.test_dir)
        result = extractor.process_pdf(self.sample_pdf)
        self.assertIn('pages', result)
        self.assertGreater(len(result['pages']), 0)
        self.assertGreater(result['pages'][0]['word_count'], 0)
    
    @classmethod
    def tearDownClass(cls):
        shutil.rmtree(cls.test_dir)

if __name__ == '__main__':
    unittest.main()