当前位置：首页 > news >正文

新手向:使用Python从PDF中高效提取结构化文本

news 2025/7/14 7:46:30

本文将详细讲解如何使用Python开发一个专业的PDF文本提取工具，帮助您从PDF文档中高效提取结构化文本数据，适用于数据分析、内容归档和知识管理等场景。

环境准备

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

Python环境：建议Python 3.8或更高版本
必要库：
- PyPDF2（基础PDF操作）
- pdfminer.six（高级文本提取）
- pandas（数据导出）

安装命令：

pip install PyPDF2 pdfminer.six pandas

工具功能概述

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

提取PDF文档元数据（作者、标题等）
按页面提取文本内容
保留文本基本格式和结构
识别文档目录结构
支持批量处理多个PDF文件
导出为结构化格式（CSV/Excel）

完整代码实现

import os
import re
from datetime import datetime
from typing import List, Dict, Optional, Tupleimport pandas as pd
from PyPDF2 import PdfReader
from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextContainerclass PDFTextExtractor:"""专业的PDF文本提取工具"""def __init__(self, output_dir: str = "output"):"""初始化提取工具:param output_dir: 输出目录路径"""self.output_dir = output_diros.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.metadatareturn {'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 tocdef 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 resultdef 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 resultsdef 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_diros.makedirs(self.output_dir, exist_ok=True)# 文本清理正则表达式self.clean_patterns = [(r'\s+', ' '),  # 合并多个空白字符(r'\n{3,}', '\n\n'),  # 限制连续换行(r'[^\x00-\x7F]+', ' '),  # 移除非ASCII字符]

默认输出目录为"output"，自动创建目录
预定义文本清理规则，确保提取文本质量
使用exist_ok=True避免目录已存在错误

2. PDF元数据提取

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

使用PyPDF2读取PDF基础信息
提取标准文档属性（标题、作者等）
包含文件基本信息（名称、页数等）
记录提取时间戳便于追踪

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)

使用pdfminer的布局分析功能
精确识别文本容器元素
应用多级文本清理规则
保留合理的文本结构（段落分隔）

4. 目录结构提取

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):continuetoc.append({'title': item.title,'page': reader.get_destination_page_number(item) + 1})except Exception:passreturn toc

尝试提取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 resultsdef export_to_csv(self, data: List[Dict[str, any]], prefix: str = "pdf_export"):# 准备元数据和页面内容DataFramemeta_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 docfor 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')

支持批量处理多个PDF文件
每个文件独立错误处理不影响整体
使用pandas构建结构化数据
自动生成时间戳文件名避免覆盖
UTF-8编码确保特殊字符正确保存

高级应用与扩展

1. OCR集成（处理扫描版PDF）

try:import pytesseractfrom pdf2image import convert_from_pathdef 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 camelotdef 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 sqlite3def 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 ThreadPoolExecutordef 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))

增量处理：
- 记录已处理文件避免重复工作
- 支持断点续处理
内存优化：
- 流式处理大文件
- 限制同时打开的文件数

安全注意事项

文件验证：
- 检查文件确实是PDF格式
- 验证文件完整性
敏感数据处理：
- 可选擦除敏感内容
- 提供内容过滤选项
权限控制：
- 检查文件读写权限
- 安全处理临时文件

单元测试建议

import unittest
import shutil
from pathlib import Pathclass TestPDFTextExtractor(unittest.TestCase):@classmethoddef 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)@classmethoddef tearDownClass(cls):shutil.rmtree(cls.test_dir)if __name__ == '__main__':unittest.main()