PDF 文件无处不在，你可能在各种地方都见过它们，比如大学论文、电费账单、办公合同、产品手册等等。它们非常常见，但处理起来却并不像看起来那么简单。假设你想从 PDF 中提取有用信息，比如读取文本、将其拆分为各个部分，或者获取一个快速摘要。这听起来似乎很简单，但当你真正尝试时，就会发现并非如此。
与 Word 或 HTML 文件不同，PDF 文件并不以一种整洁、易于阅读的方式存储内容。它们的设计初衷是为了看起来美观，而不是为了被程序读取。文本可能到处都是，被分割成奇怪的块，散布在页面上，或者与表格和图像混杂在一起。这使得从它们中获取干净、结构化的数据变得非常困难。
接下来，我们将构建一个能够处理这种混乱局面的工具。我们将创建一个自定义的 PDF 解析器，它可以：
-
在页面级别提取并清理 PDF 中的文本，可选地保留布局格式以获得更好的格式化效果 -
处理图像元数据提取 -
通过检测跨页面重复的行来移除不需要的页眉和页脚，从而减少噪声 -
检索详细的文档和页面级别元数据，如作者、标题、创建日期、旋转角度和页面大小 -
将内容拆分为便于进一步进行自然语言处理（NLP）或大型语言模型（LLM）处理的可管理部分
项目结构
在开始之前，最好组织一下项目文件，以便清晰和可扩展。
custom_pdf_parser/
│
├── parser.py
├── langchain_loader.py
├── pipeline.py
├── example.py
├── requirements.txt # 依赖项列表
└── __init__.py # （可选）标记目录为 Python 包
你可以将 __init__.py 文件留空，因为它的主要作用仅仅是表明这个目录应该被视为一个 Python 包。接下来我将逐步解释其余文件的用途。
所需工具（requirements.txt）
所需的库有：
-
PyPDF：一个纯 Python 库，用于读取和写入 PDF 文件。我们将使用它从 PDF 文件中提取文本。 -
LangChain：一个用于构建具有语言模型的情境感知应用程序的框架（我们将用它来处理和链接文档任务）。它将用于正确处理和组织文本。
使用以下命令安装它们：
pip install pypdf langchain
如果你想整齐地管理依赖项，可以创建一个 requirements.txt 文件，内容如下：
pypdf
langchain
requests
然后运行以下命令：
pip install -r requirements.txt
PDF 解析器（parser.py）
核心类 CustomPDFParser 使用 PyPDF 从每页 PDF 中提取文本和元数据。它还包含清理文本、提取图像信息（可选）以及移除每页上经常出现的重复页眉或页脚的方法。
-
它支持保留布局格式 -
它提取诸如页码、旋转角度和媒体框尺寸等元数据 -
它可以过滤掉内容太少的页面 -
文本清理会移除多余的空白，同时保留段落分隔
以下是实现所有这些功能的代码：
import os
import logging
from pathlib import Path
from typing import List, Dict, Any
import pypdf
from pypdf import PdfReader
# 配置日志以显示信息及以上的消息
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CustomPDFParser:
def __init__(
self,extract_images: bool = False,preserve_layout: bool = True,remove_headers_footers: bool = True,min_text_length: int = 10
):
"""
使用选项初始化解析器，以提取图像、保留布局、移除重复的页眉/页脚以及设置页面的最小文本长度。
参数：
extract_images：是否从页面中提取图像信息
preserve_layout：是否在文本提取中保留布局间距
remove_headers_footers：是否检测并移除页眉/页脚
min_text_length：页面被视为有效的最小文本长度
"""
self.extract_images = extract_images
self.preserve_layout = preserve_layout
self.remove_headers_footers = remove_headers_footers
self.min_text_length = min_text_length
def extract_text_from_page(self, page: pypdf.PageObject, page_num: int) -> Dict[str, Any]:
"""
从单个 PDF 页面中提取文本和元数据。
参数：
page：PyPDF 页面对象
page_num：从零开始的页面编号
返回值：
包含以下键的字典：
- 'text'：提取并清理后的文本字符串，
- 'metadata'：页面元数据字典，
- 'word_count'：提取文本中的单词数量
"""
try:
# 提取文本，可选地保留布局以获得更好的格式化效果
if self.preserve_layout:
text = page.extract_text(extraction_mode="layout")
else:
text = page.extract_text()
# 清理文本：移除多余的空白并规范段落
text = self._clean_text(text)
# 收集页面元数据（页面编号、旋转角度、媒体框）
metadata = {
"page_number": page_num + 1, # 从 1 开始编号
"rotation": getattr(page, "rotation", 0),
"mediabox": str(getattr(page, "mediabox", None)),
}
# 如果请求，可选地从页面中提取图像信息
if self.extract_images:
metadata["images"] = self._extract_image_info(page)
# 返回包含该页面文本和元数据的字典
return {
"text": text,
"metadata": metadata,
"word_count": len(text.split()) if text else0
}
except Exception as e:
# 记录错误，并为有问题的页面返回空数据
logger.error(f"提取页面 {page_num} 时出错：{e}")
return {
"text": "",
"metadata": {"page_number": page_num + 1, "error": str(e)},
"word_count": 0
}
def _clean_text(self, text: str) -> str:
"""
清理并规范提取的文本，保留段落分隔。
参数：
text：从 PDF 页面中提取的原始文本
返回值：
清理后的文本字符串
"""
ifnot text:
return""
lines = text.split('\n')
cleaned_lines = []
for line in lines:
line = line.strip() # 移除首尾空白
if line:
# 非空行；保留它
cleaned_lines.append(line)
elif cleaned_lines and cleaned_lines[-1]:
# 通过仅在前一行存在时保留空行来保留段落分隔
cleaned_lines.append("")
cleaned_text = '\n'.join(cleaned_lines)
# 将超过两个连续空行的实例减少到两个
while'\n\n\n'in cleaned_text:
cleaned_text = cleaned_text.replace('\n\n\n', '\n\n')
return cleaned_text.strip()
def _extract_image_info(self, page: pypdf.PageObject) -> List[Dict[str, Any]]:
"""
从页面中提取基本图像元数据（如果可用）。
参数：
page：PyPDF 页面对象
返回值：
包含图像信息（索引、名称、宽度、高度）的字典列表
"""
images = []
try:
# PyPDF 页面可以有一个 'images' 属性，列出嵌入的图像
if hasattr(page, 'images'):
for i, image in enumerate(page.images):
images.append({
"image_index": i,
"name": getattr(image, 'name', f"image_{i}"),
"width": getattr(image, 'width', None),
"height": getattr(image, 'height', None)
})
except Exception as e:
logger.warning(f"图像提取失败：{e}")
return images
def _remove_headers_footers(self, pages_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
移除在许多页面上出现的重复页眉和页脚。
通过识别在超过 50% 的页面文本开头或结尾处出现的行来实现，然后移除这些行。
参数：
pages_data：代表每个页面提取数据的字典列表。
返回值：
移除页眉/页脚后的页面更新列表
"""
# 如果页面数量足够且启用了该选项，则尝试移除
if len(pages_data) < 3ornot self.remove_headers_footers:
return pages_data
# 收集每个页面文本的首行和尾行以供分析
first_lines = [page["text"].split('\n')[0] if page["text"] else""for page in pages_data]
last_lines = [page["text"].split('\n')[-1] if page["text"] else""for page in pages_data]
threshold = len(pages_data) * 0.5# 超过 50% 的页面
# 识别频繁出现的候选页眉和页脚
potential_headers = [line for line in set(first_lines)
if first_lines.count(line) > threshold and line.strip()]
potential_footers = [line for line in set(last_lines)
if last_lines.count(line) > threshold and line.strip()]
# 从每个页面的文本中移除已识别的页眉和页脚
for page_data in pages_data:
lines = page_data["text"].split('\n')
# 如果与频繁出现的页眉匹配，则移除页眉
if lines and potential_headers:
for header in potential_headers:
if lines[0].strip() == header.strip():
lines = lines[1:]
break
# 如果与频繁出现的页脚匹配，则移除页脚
if lines and potential_footers:
for footer in potential_footers:
if lines[-1].strip() == footer.strip():
lines = lines[:-1]
break
page_data["text"] = '\n'.join(lines).strip()
return pages_data
def _extract_document_metadata(self, pdf_reader: PdfReader, pdf_path: str) -> Dict[str, Any]:
"""
从 PDF 文档本身提取元数据。
参数：
pdf_reader：PyPDF PdfReader 实例
pdf_path：PDF 文件路径
返回值：
包含文件信息和 PDF 文档元数据的字典
"""
metadata = {
"file_path": pdf_path,
"file_name": Path(pdf_path).name,
"file_size": os.path.getsize(pdf_path) if os.path.exists(pdf_path) elseNone,
}
try:
if pdf_reader.metadata:
# 如果可用，提取常见的 PDF 元数据键
metadata.update({
"title": pdf_reader.metadata.get('/Title', ''),
"author": pdf_reader.metadata.get('/Author', ''),
"subject": pdf_reader.metadata.get('/Subject', ''),
"creator": pdf_reader.metadata.get('/Creator', ''),
"producer": pdf_reader.metadata.get('/Producer', ''),
"creation_date": str(pdf_reader.metadata.get('/CreationDate', '')),
"modification_date": str(pdf_reader.metadata.get('/ModDate', '')),
})
except Exception as e:
logger.warning(f"元数据提取失败：{e}")
return metadata
def parse_pdf(self, pdf_path: str) -> Dict[str, Any]:
"""
解析整个 PDF 文件。打开文件，逐页提取文本和元数据，如果配置了，则移除页眉/页脚，并汇总结果。
参数：
pdf_path：PDF 文件路径
返回值：
包含以下键的字典：
- 'full_text'：所有页面合并后的文本，
- 'pages'：包含文本和元数据的页面字典列表，
- 'document_metadata'：文件和 PDF 元数据，
- 'total_pages'：PDF 中的总页数，
- 'processed_pages'：过滤后保留的页面数量，
- 'total_words'：解析文本的总单词数
"""
try:
with open(pdf_path, 'rb') as file:
pdf_reader = PdfReader(file)
doc_metadata = self._extract_document_metadata(pdf_reader, pdf_path)
pages_data = []
# 遍历所有页面并提取数据
for i, page in enumerate(pdf_reader.pages):
page_data = self.extract_text_from_page(page, i)
# 只保留文本长度足够的页面
if len(page_data["text"]) >= self.min_text_length:
pages_data.append(page_data)
# 移除重复的页眉和页脚
pages_data = self._remove_headers_footers(pages_data)
# 使用双换行符作为分隔符合并所有页面文本
full_text = '\n\n'.join(page["text"] for page in pages_data if page["text"])
# 返回最终的结构化数据
return {
"full_text": full_text,
"pages": pages_data,
"document_metadata": doc_metadata,
"total_pages": len(pdf_reader.pages),
"processed_pages": len(pages_data),
"total_words": sum(page["word_count"] for page in pages_data)
}
except Exception as e:
logger.error(f"解析 PDF {pdf_path} 失败：{e}")
raise
与 LangChain 集成（langchain_loader.py）
LangChainPDFLoader 类封装了自定义解析器，并将解析后的页面转换为 LangChain 文档对象，这些对象是 LangChain 流水线的基础构建块。
-
它允许使用 LangChain 的 RecursiveCharacterTextSplitter将文档拆分为更小的部分 -
你可以自定义分块大小和重叠部分，以便用于下游的 LLM 输入 -
这个加载器支持将原始 PDF 内容与 LangChain 的文档抽象之间进行干净的集成
其背后的逻辑如下：
from typing import List, Optional, Dict, Any
from langchain.schema import Document
from langchain.document_loaders.base import BaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from parser import CustomPDFParser # 导入上面定义的解析器
class LangChainPDFLoader(BaseLoader):
def __init__(
self,file_path: str,parser_config: Optional[Dict[str, Any]] = None,chunk_size: int = 500, chunk_overlap: int = 50
):
"""
使用 PDF 文件路径、解析器配置和分块参数初始化加载器。
参数：
file_path：PDF 文件路径
parser_config：解析器选项字典
chunk_size：用于拆分长文本的分块大小
chunk_overlap：拆分时的分块重叠部分
"""
self.file_path = file_path
self.parser_config = parser_config or {}
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.parser = CustomPDFParser(**self.parser_config)
def load(self) -> List[Document]:
"""
加载 PDF，解析页面，并将每个页面转换为 LangChain 文档。
返回值：
包含页面文本和合并元数据的文档对象列表。
"""
parsed_data = self.parser.parse_pdf(self.file_path)
documents = []
# 将每个页面字典转换为 LangChain 文档
for page_data in parsed_data["pages"]:
if page_data["text"]:
# 合并文档级别和页面级别的元数据
metadata = {**parsed_data["document_metadata"], **page_data["metadata"]}
doc = Document(page_content=page_data["text"], metadata=metadata)
documents.append(doc)
return documents
def load_and_split(self) -> List[Document]:
"""
加载 PDF 并将大型文档拆分为更小的分块。
返回值：
拆分大型文本后的文档对象列表。
"""
documents = self.load()
# 使用所需的分块大小和重叠部分初始化文本拆分器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=["\n\n", "\n", " ", ""] # 分层拆分
)
# 将文档拆分为更小的分块
split_docs = text_splitter.split_documents(documents)
return split_docs
构建处理流水线（pipeline.py）
PDFProcessingPipeline 类提供了一个更高层次的接口，用于：
-
处理单个 PDF -
选择输出格式（原始字典、LangChain 文档或纯文本） -
启用或禁用分块，并可配置分块大小 -
处理错误和日志记录
这种抽象使得将其轻松集成到更大的应用程序或工作流中成为可能。其背后的逻辑如下：
from typing import List, Optional, Dict, Any
from langchain.schema import Document
from parser import CustomPDFParser
from langchain_loader import LangChainPDFLoader
import logging
logger = logging.getLogger(__name__)
class PDFProcessingPipeline:
def __init__(self, parser_config: Optional[Dict[str, Any]] = None):
"""
参数：
parser_config：传递给 CustomPDFParser 的选项字典
"""
self.parser_config = parser_config or {}
def process_single_pdf(
self,pdf_path: str,output_format: str = "langchain",chunk_documents: bool = True,chunk_size: int = 500,chunk_overlap: int = 50
) -> Any:
"""
参数：
pdf_path：PDF 文件路径
output_format："raw"（字典）、"langchain"（文档）或 "text"（字符串）
chunk_documents：是否拆分 LangChain 文档
chunk_size：拆分时的分块大小
chunk_overlap：拆分时的分块重叠部分
返回值：
按请求格式解析的内容
"""
if output_format == "raw":
# 使用原始 CustomPDFParser 输出
parser = CustomPDFParser(**self.parser_config)
return parser.parse_pdf(pdf_path)
elif output_format == "langchain":
# 使用 LangChain 加载器，可选地进行分块
loader = LangChainPDFLoader(pdf_path, self.parser_config, chunk_size, chunk_overlap)
if chunk_documents:
return loader.load_and_split()
else:
return loader.load()
elif output_format == "text":
# 仅返回合并后的纯文本
parser = CustomPDFParser(**self.parser_config)
parsed_data = parser.parse_pdf(pdf_path)
return parsed_data.get("full_text", "")
else:
raise ValueError(f"未知的 output_format：{output_format}")
测试解析器（example.py）
让我们按照以下方式测试解析器：
import os
from pathlib import Path
def main():
print("欢迎使用自定义 PDF 解析器！")
print("你想做什么？")
print("1. 查看完整的原始解析数据")
print("2. 提取完整的纯文本")
print("3. 获取 LangChain 文档（不分块）")
print("4. 获取 LangChain 文档（分块）")
print("5. 显示文档元数据")
print("6. 显示每页元数据")
print("7. 显示清理后的页面文本（已移除页眉/页脚）")
print("8. 显示提取的图像元数据")
choice = input("输入你的选择编号：").strip()
if choice notin {'1', '2', '3', '4', '5', '6', '7', '8'}:
print("无效选项。")
return
file_path = input("输入你的 PDF 文件路径：").strip()
ifnot Path(file_path).exists():
print("文件未找到。")
return
# 初始化流水线
pipeline = PDFProcessingPipeline({
"preserve_layout": False,
"remove_headers_footers": True,
"extract_images": True,
"min_text_length": 20
})
# 大部分选项需要原始数据
parsed = pipeline.process_single_pdf(file_path, output_format="raw")
if choice == '1':
print("\n完整的原始解析输出：")
for k, v in parsed.items():
print(f"{k}: {str(v)[:300]}...")
elif choice == '2':
print("\n清理后的完整文本（截断预览）：")
print("预览前 1000 个字符：\n"+parsed["full_text"][:1000], "...")
elif choice == '3':
docs = pipeline.process_single_pdf(file_path, output_format="langchain", chunk_documents=False)
print(f"\nLangChain 文档：{len(docs)}")
print("预览前 500 个字符：\n", docs[0].page_content[:500], "...")
elif choice == '4':
docs = pipeline.process_single_pdf(file_path, output_format="langchain", chunk_documents=True)
print(f"\nLangChain 分块：{len(docs)}")
print("样本分块内容（前 500 个字符）：")
print(docs[0].page_content[:500], "...")
elif choice == '5':
print("\n文档元数据：")
for key, value in parsed["document_metadata"].items():
print(f"{key}: {value}")
elif choice == '6':
print("\n每页元数据：")
for i, page in enumerate(parsed["pages"]):
print(f"第 {i+1} 页：{page['metadata']}")
elif choice == '7':
print("\n移除页眉/页脚后的清理文本。")
print("显示前 3 页以及每页的前 500 个字符的文本。")
for i, page in enumerate(parsed["pages"][:3]): # 前 3 页
print(f"\n--- 第 {i+1} 页 ---")
print(page["text"][:500], "...")
elif choice == '8':
print("\n提取的图像元数据（如果可用）：")
found = False
for i, page in enumerate(parsed["pages"]):
images = page["metadata"].get("images", [])
if images:
found = True
print(f"\n--- 第 {i+1} 页 ---")
for img in images:
print(img)
ifnot found:
print("未找到图像元数据。")
if __name__ == "__main__":
main()
运行此代码后，系统会提示你输入选择编号和 PDF 文件路径。输入这些信息后，你将看到相应的输出。我使用的 PDF 是公开可访问的，你可以使用提供的链接下载它。
运行结果
假设你已经按照上述步骤设置了项目，并且有一个名为 articles.pdf 的 PDF 文件。运行 example.py 脚本后，你将看到类似以下的交互界面和结果：
运行脚本
python example.py
交互界面和用户输入
欢迎使用自定义 PDF 解析器！
你想做什么？
1. 查看完整的原始解析数据
2. 提取完整的纯文本
3. 获取 LangChain 文档（不分块）
4. 获取 LangChain 文档（分块）
5. 显示文档元数据
6. 显示每页元数据
7. 显示清理后的页面文本（已移除页眉/页脚）
8. 显示提取的图像元数据
输入你的选择编号：5
输入你的 PDF 文件路径：/content/articles.pdf
运行结果
假设用户选择了 5 来查看文档元数据，以下是可能的输出结果：
文档元数据：
file_path: /content/articles.pdf
file_name: articles.pdf
file_size: 123456
title: Articles (a/an/the)
author: Ben Aldridge
subject: English Grammar
creator: Microsoft Word
producer: Acrobat Distiller
creation_date: D:20140301000000Z
modification_date: D:20140301000000Z
其他选项的运行结果示例
选项 1：查看完整的原始解析数据
完整的原始解析输出：
full_text: San José State University Writing Center...（截断）
pages: [{'text': 'San José State University Writing Center...（截断）', 'metadata': {'page_number': 1, 'rotation': 0, 'mediabox': '[0, 0, 612, 792]'}, 'word_count': 123}, ...]
document_metadata: {'file_path': '/content/articles.pdf', 'file_name': 'articles.pdf', 'file_size': 123456, 'title': 'Articles (a/an/the)', 'author': 'Ben Aldridge', 'subject': 'English Grammar', 'creator': 'Microsoft Word', 'producer': 'Acrobat Distiller', 'creation_date': 'D:20140301000000Z', 'modification_date': 'D:20140301000000Z'}
total_pages: 4
processed_pages: 4
total_words: 456
选项 2：提取完整的纯文本
清理后的完整文本（截断预览）：
San José State University Writing Center
www.sjsu.edu/writingcenter
Written by Ben Aldridge
Articles (a/an/the), Spring 2014. 1 of 4
Articles (a/an/the)
There are three articles in the English language: a, an, and the. They are placed before nouns
and show whether a given noun is general or specific.
Examples of Articles
...
选项 3：获取 LangChain 文档（不分块）
LangChain 文档：4
预览前 500 个字符：
San José State University Writing Center
www.sjsu.edu/writingcenter
Written by Ben Aldridge
Articles (a/an/the), Spring 2014. 1 of 4
Articles (a/an/the)
There are three articles in the English language: a, an, and the. They are placed before nouns
and show whether a given noun is general or specific.
Examples of Articles
...
选项 4：获取 LangChain 文档（分块）
LangChain 分块：16
样本分块内容（前 500 个字符）：
San José State University Writing Center
www.sjsu.edu/writingcenter
Written by Ben Aldridge
Articles (a/an/the), Spring 2014. 1 of 4
Articles (a/an/the)
There are three articles in the English language: a, an, and the. They are placed before nouns
...
选项 6：显示每页元数据
每页元数据：
第 1 页：{'page_number': 1, 'rotation': 0, 'mediabox': '[0, 0, 612, 792]', 'images': [{'image_index': 0, 'name': 'image_0', 'width': 100, 'height': 200}]}
第 2 页：{'page_number': 2, 'rotation': 0, 'mediabox': '[0, 0, 612, 792]'}
第 3 页：{'page_number': 3, 'rotation': 0, 'mediabox': '[0, 0, 612, 792]'}
第 4 页：{'page_number': 4, 'rotation': 0, 'mediabox': '[0, 0, 612, 792]'}
选项 7：显示清理后的页面文本（已移除页眉/页脚）
移除页眉/页脚后的清理文本。
显示前 3 页以及每页的前 500 个字符的文本。
--- 第 1 页 ---
There are three articles in the English language: a, an, and the. They are placed before nouns
and show whether a given noun is general or specific.
Examples of Articles
...
--- 第 2 页 ---
The indefinite articles 'a' and 'an' are used before singular nouns that introduce something for the first time...
...
--- 第 3 页 ---
The definite article 'the' is used before singular and plural nouns when the noun is specific or particular...
...
选项 8：显示提取的图像元数据
提取的图像元数据（如果可用）：
--- 第 1 页 ---
{'image_index': 0, 'name': 'image_0', 'width': 100, 'height': 200}
结论
通过上面的内容，你已经学会了如何使用开源工具构建一个灵活且强大的 PDF 处理流水线。由于它是模块化的，你可以轻松地对其进行扩展，例如使用 Streamlit 添加一个搜索栏，将分块存储在 FAISS 等向量数据库中以便进行更智能的查找，或者甚至将其集成到聊天机器人中。你无需重新构建任何东西，只需连接下一个组件即可。PDF 不再像一个封闭的盒子。通过这种方法，你可以将任何文档转换为你可以按自己意愿阅读、搜索和理解的内容。