python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python PDF文本提取

使用Python打造一个专业的PDF文本提取工具

作者:超级小识

这篇文章主要为大家详细介绍了如何使用Python开发一个专业的PDF文本提取工具,帮助大家从PDF文档中高效提取结构化文本数据,适用于数据分析,内容归档和知识管理等场景

简介

我们将采用PyPDF2、pdfplumber等主流Python库来实现核心功能,并重点解决以下技术难点:

1.文本提取精度优化

2.结构化数据处理

3.性能优化方案

典型应用场景包括:

开发环境要求:

我们将分步骤实现:

最终成果将是一个可复用的PDF处理工具包,支持命令行和API两种调用方式,方便集成到各类数据处理流程中。

环境准备

开发本工具需要以下环境配置:

Python环境:建议Python 3.8或更高版本

必要库

安装命令:

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:
    """专业的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'),  # 限制连续换行
            (r'[^\x00-\x7F]+', ' '),  # 移除非ASCII字符
        ]
    
    def extract_metadata(self, pdf_path: str) -> Dict[str, str]:
        """提取PDF元数据"""
        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:
        """清理和规范化提取的文本"""
        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]]:
        """尝试提取文档目录结构"""
        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"):
        """将提取结果导出为CSV"""
        # 准备元数据表格
        meta_data = [item['metadata'] for item in data if 'metadata' in item]
        meta_df = pd.DataFrame(meta_data)
        
        # 准备页面内容表格
        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"
    ]
    
    # 批量处理并导出
    results = extractor.batch_process(sample_files)
    meta_csv, pages_csv = extractor.export_to_csv(results)
    
    print(f"\n处理完成!\n元数据已保存至: {meta_csv}\n页面内容已保存至: {pages_csv}")

代码深度解析

1. 类设计与初始化

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'),  # 限制连续换行
            (r'[^\x00-\x7F]+', ' '),  # 移除非ASCII字符
        ]

2. PDF元数据提取

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', ''),
            # ...其他元数据字段
        }

3. 文本内容提取与清理

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)

4. 目录结构提取

本工具将实现以下核心功能:

1.PDF文档元数据提取

自动识别并提取文档属性信息,包括但不限于:

2.精准文本内容提取

支持按页面粒度提取文本

智能识别文档分栏排版,保持原始阅读顺序

处理特殊文本元素:

3.格式与结构保留

4.智能目录解析

5.批量处理能力

6.多样化输出选项

结构化数据导出:

自定义输出模板:

尝试提取PDF内置目录结构

处理嵌套目录项(简化版跳过子项)

容错处理确保主流程不受影响

返回标准化的目录条目列表

5. 批量处理与导出

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"):
    # 准备元数据和页面内容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')

高级应用与扩展

1. OCR集成(处理扫描版PDF)

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

2. 表格数据提取

try:
    import camelot
    
    def extract_tables(self, pdf_path: str) -> List[Dict[str, any]]:
        """提取PDF中的表格数据"""
        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

3. 数据库存储支持

import sqlite3

def export_to_sqlite(self, data: List[Dict[str, any]], db_name: str = "pdf_data.db"):
    """将提取结果导出到SQLite数据库"""
    conn = sqlite3.connect(os.path.join(self.output_dir, db_name))
    
    # 创建元数据表
    conn.execute('''
    CREATE TABLE IF NOT EXISTS pdf_metadata (
        file_name TEXT PRIMARY KEY,
        title TEXT,
        author TEXT,
        page_count INTEGER,
        created_date TEXT
    )
    ''')
    
    # 创建页面内容表
    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
    )
    ''')
    
    # 插入数据
    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)
        
        # 创建测试PDF (实际使用中应准备样例文件)
        cls.sample_pdf = cls.test_dir / "sample.pdf"
        # 这里应添加PDF生成代码或使用预准备的测试文件
    
    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()

结语

本文详细讲解了专业PDF文本提取工具的开发过程,涵盖了以下核心技术和实现细节:

1.PDF元数据提取技术

2.文本内容精确提取方法

3.结构化数据导出策略

4.异常处理和性能考量

5.多种扩展可能性

读者可以通过这个基础框架,根据实际需求添加更多高级功能,如:

1.自定义内容过滤规则

2.支持更多输出格式

3.集成到自动化工作流中

4.开发Web服务接口

建议在实际使用前充分测试各种类型的PDF文档,特别是处理以下特殊场景时:

测试时应重点关注:

以上就是使用Python打造一个专业的PDF文本提取工具的详细内容,更多关于Python PDF文本提取的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文