1. 项目背景与需求分析
在现代Web应用中，PDF 文档处理是一个常见但复杂的需求。无论是企业 OA 系统、在线教育平台还是知识管理工具，都需要能够高效解析 PDF 内容。本次作业要求实现一个中间件，专门处理前端上传的PDF文件，具体要求如下：
- 输入格式：接收 Base64 编码的 PDF 数据
- 核心功能：将 Base64 转换为正常 PDF 文件并提取文本内容
- 输出要求：将提取的文本内容完整放入 system_message 中，作为后续对话的上下文
- 扩展目标：最终版本应支持上传或发送包含完整路径信息的任何文件（包括 URL 地址），都能进行合理的解析和处理
2. 技术栈选择
2.1 后端框架
- FastAPI：现代、高性能的 Python Web 框架，支持异步处理
- Pydantic：数据验证和设置管理
2.2 PDF处理库
- PyPDF2 / pdfplumber：PDF文本提取
- pdf2image：PDF转图像（可选，用于OCR场景）
- base64：标准库，用于Base64编解码
2.3 文件处理
- aiofiles：异步文件操作
- httpx：异步 HTTP 客户端（用于 URL 下载）
2.4 环境配置
1# requirements.txt 2fastapi==0.104.1 3uvicorn==0.24.0 4pydantic==2.5.0 5PyPDF2==3.0.1 6pdfplumber==0.10.3 7aiofiles==23.2.1 8httpx==0.25.1 9python-multipart==0.0.6 10
3. 核心中间件实现
3.1 基础模型定义
1from pydantic import BaseModel, Field 2from typing import Optional, Union 3from enum import Enum 4 5class FileSourceType(str, Enum): 6 BASE64 = "base64" 7 URL = "url" 8 LOCAL_PATH = "local_path" 9 UPLOAD = "upload" 10 11class PDFParseRequest(BaseModel): 12 """PDF 解析请求模型""" 13 source_type: FileSourceType = Field(..., description="文件来源类型") 14 content: Optional[str] = Field(None, description="Base64 内容或 URL 路径") 15 file_path: Optional[str] = Field(None, description="本地文件路径") 16 file_upload: Optional[bytes] = Field(None, description="上传的文件二进制数据") 17 include_metadata: bool = Field(True, description="是否包含 PDF 元数据") 18 language: str = Field("chi_sim", description="OCR 语言设置（如需要）") 19 20class PDFParseResponse(BaseModel): 21 """PDF 解析响应模型""" 22 success: bool 23 text_content: str 24 page_count: int 25 metadata: Optional[dict] = None 26 error_message: Optional[str] = None 27
3.2 中间件核心类
1import base64 2import tempfile 3import os 4from pathlib import Path 5import pdfplumber 6import httpx 7from typing import Tuple, Optional 8import logging 9 10logger = logging.getLogger(__name__) 11 12class PDFParserMiddleware: 13 """PDF解析中间件""" 14 15 def __init__(self, temp_dir: str = None): 16 """ 17 初始化PDF解析中间件 18 19 Args: 20 temp_dir: 临时文件目录，默认为系统临时目录 21 """ 22 self.temp_dir = temp_dir or tempfile.gettempdir() 23 Path(self.temp_dir).mkdir(parents=True, exist_ok=True) 24 25 async def parse_pdf(self, request: PDFParseRequest) -> PDFParseResponse: 26 """ 27 解析PDF文件的主入口方法 28 29 Args: 30 request: PDF 解析请求 31 32 Returns: 33 PDFParseResponse: 解析结果 34 """ 35 try: 36 # 根据来源类型获取PDF文件路径 37 pdf_path = await self._get_pdf_file(request) 38 39 # 提取文本内容 40 text_content, page_count, metadata = self._extract_text_from_pdf(pdf_path) 41 42 # 清理临时文件 43 self._cleanup_temp_file(pdf_path, request.source_type) 44 45 return PDFParseResponse( 46 success=True, 47 text_content=text_content, 48 page_count=page_count, 49 metadata=metadata 50 ) 51 52 except Exception as e: 53 logger.error(f"PDF 解析失败: {str(e)}", exc_info=True) 54 return PDFParseResponse( 55 success=False, 56 text_content="", 57 page_count=0, 58 error_message=str(e) 59 ) 60 61 async def _get_pdf_file(self, request: PDFParseRequest) -> str: 62 """ 63 根据来源类型获取PDF文件路径 64 65 Returns: 66 str: 本地PDF文件路径 67 """ 68 if request.source_type == FileSourceType.BASE64: 69 return await self._save_base64_to_file(request.content) 70 elif request.source_type == FileSourceType.URL: 71 return await self._download_from_url(request.content) 72 elif request.source_type == FileSourceType.LOCAL_PATH: 73 return request.file_path 74 elif request.source_type == FileSourceType.UPLOAD: 75 return await self._save_upload_to_file(request.file_upload) 76 else: 77 raise ValueError(f"不支持的来源类型: {request.source_type}") 78 79 async def _save_base64_to_file(self, base64_content: str) -> str: 80 """将Base64内容保存为临时PDF文件""" 81 if not base64_content: 82 raise ValueError("Base64内容不能为空") 83 84 # 移除可能的data URL前缀 85 if "," in base64_content: 86 base64_content = base64_content.split(",")[1] 87 88 # 解码Base64 89 pdf_bytes = base64.b64decode(base64_content) 90 91 # 保存到临时文件 92 temp_file = tempfile.NamedTemporaryFile( 93 suffix=".pdf", 94 dir=self.temp_dir, 95 delete=False 96 ) 97 temp_file.write(pdf_bytes) 98 temp_file.close() 99 100 return temp_file.name 101 102 async def _download_from_url(self, url: str) -> str: 103 """从URL下载PDF文件""" 104 if not url: 105 raise ValueError("URL不能为空") 106 107 async with httpx.AsyncClient() as client: 108 response = await client.get(url) 109 response.raise_for_status() 110 111 # 保存到临时文件 112 temp_file = tempfile.NamedTemporaryFile( 113 suffix=".pdf", 114 dir=self.temp_dir, 115 delete=False 116 ) 117 temp_file.write(response.content) 118 temp_file.close() 119 120 return temp_file.name 121 122 async def _save_upload_to_file(self, file_data: bytes) -> str: 123 """保存上传的文件数据到临时文件""" 124 if not file_data: 125 raise ValueError("上传文件数据不能为空") 126 127 temp_file = tempfile.NamedTemporaryFile( 128 suffix=".pdf", 129 dir=self.temp_dir, 130 delete=False 131 ) 132 temp_file.write(file_data) 133 temp_file.close() 134 135 return temp_file.name 136 137 def _extract_text_from_pdf(self, pdf_path: str) -> Tuple[str, int, dict]: 138 """ 139 从PDF文件中提取文本内容 140 141 Returns: 142 Tuple[str, int, dict]: (文本内容, 页数, 元数据) 143 """ 144 text_parts = [] 145 metadata = {} 146 page_count = 0 147 148 try: 149 with pdfplumber.open(pdf_path) as pdf: 150 page_count = len(pdf.pages) 151 metadata = pdf.metadata 152 153 for page_num, page in enumerate(pdf.pages, 1): 154 page_text = page.extract_text() 155 if page_text: 156 text_parts.append(f"=== 第{page_num}页 ===\n{page_text}\n") 157 else: 158 # 如果无法提取文本，可能是扫描版PDF 159 text_parts.append(f"=== 第{page_num}页 ===\n[扫描图像，需要OCR处理]\n") 160 161 except Exception as e: 162 logger.warning(f"pdfplumber 提取失败，尝试 PyPDF2: {str(e)}") 163 # 回退到PyPDF2 164 import PyPDF2 165 with open(pdf_path, 'rb') as file: 166 pdf_reader = PyPDF2.PdfReader(file) 167 page_count = len(pdf_reader.pages) 168 metadata = pdf_reader.metadata 169 170 for page_num, page in enumerate(pdf_reader.pages, 1): 171 page_text = page.extract_text() 172 if page_text: 173 text_parts.append(f"=== 第{page_num}页 ===\n{page_text}\n") 174 175 full_text = "\n".join(text_parts) 176 return full_text, page_count, metadata 177 178 def _cleanup_temp_file(self, file_path: str, source_type: FileSourceType): 179 """清理临时文件（除本地路径外）""" 180 if source_type != FileSourceType.LOCAL_PATH: 181 try: 182 os.unlink(file_path) 183 except: 184 pass 185
3.3 FastAPI路由集成
1from fastapi import FastAPI, UploadFile, File, HTTPException 2from fastapi.middleware.cors import CORSMiddleware 3from contextlib import asynccontextmanager 4import uvicorn 5 6# 创建应用实例 7app = FastAPI(title="PDF 解析中间件 API") 8 9# 添加CORS中间件 10app.add_middleware( 11 CORSMiddleware, 12 allow_origins=["*"], 13 allow_credentials=True, 14 allow_methods=["*"], 15 allow_headers=["*"], 16) 17 18# 全局PDF解析器实例 19pdf_parser = PDFParserMiddleware() 20 21@asynccontextmanager 22async def lifespan(app: FastAPI): 23 """应用生命周期管理""" 24 # 启动时初始化 25 print("PDF解析中间件服务启动...") 26 yield 27 # 关闭时清理 28 print("PDF解析中间件服务关闭...") 29 30app = FastAPI(lifespan=lifespan) 31 32@app.post("/api/parse/pdf", response_model=PDFParseResponse) 33async def parse_pdf(request: PDFParseRequest): 34 """ 35 解析 PDF 文件接口 36 37 - 支持 Base64、URL、本地路径、文件上传多种方式 38 - 返回提取的文本内容和元数据 39 """ 40 return await pdf_parser.parse_pdf(request) 41 42@app.post("/api/parse/upload", response_model=PDFParseResponse) 43async def parse_upload_pdf( 44 file: UploadFile = File(..., description="上传的PDF文件"), 45 include_metadata: bool = True 46): 47 """ 48 通过文件上传方式解析 PDF 49 50 - 支持 multipart/form-data 上传 51 - 自动检测文件类型 52 """ 53 if not file.filename.lower().endswith('.pdf'): 54 raise HTTPException(400, "只支持PDF文件") 55 56 content = await file.read() 57 58 request = PDFParseRequest( 59 source_type=FileSourceType.UPLOAD, 60 file_upload=content, 61 include_metadata=include_metadata 62 ) 63 64 return await pdf_parser.parse_pdf(request) 65 66@app.get("/api/health") 67async def health_check(): 68 """健康检查接口""" 69 return {"status": "healthy", "service": "pdf-parser-middleware"} 70
4. 系统集成与上下文管理
4.1 集成到对话系统
1class ConversationSystem: 2 """对话系统集成示例""" 3 4 def __init__(self): 5 self.pdf_parser = PDFParserMiddleware() 6 self.system_messages = {} 7 8 async def process_user_query(self, user_id: str, query: str, pdf_request: PDFParseRequest = None): 9 """ 10 处理用户查询，支持PDF上下文 11 12 Args: 13 user_id: 用户ID 14 query: 用户查询文本 15 pdf_request: PDF解析请求（可选） 16 """ 17 system_message = "" 18 19 # 如果有PDF解析请求，先解析PDF 20 if pdf_request: 21 pdf_result = await self.pdf_parser.parse_pdf(pdf_request) 22 23 if pdf_result.success: 24 # 将PDF内容作为系统消息上下文 25 system_message = self._format_pdf_context(pdf_result) 26 self.system_messages[user_id] = system_message 27 else: 28 return { 29 "error": f"PDF解析失败: {pdf_result.error_message}", 30 "query": query 31 } 32 33 # 如果有历史系统消息，合并 34 if user_id in self.system_messages: 35 system_message = self.system_messages[user_id] 36 37 # 构建完整的对话上下文 38 full_context = f""" 39系统上下文（来自PDF文档）： 40{system_message} 41 42用户查询：{query} 43 44请基于以上上下文回答用户问题。 45""" 46 47 # 这里可以调用LLM接口 48 response = await self._call_llm_api(full_context) 49 50 return { 51 "response": response, 52 "has_pdf_context": bool(system_message) 53 } 54 55 def _format_pdf_context(self, pdf_result: PDFParseResponse) -> str: 56 """格式化PDF内容为系统消息""" 57 metadata_str = "" 58 if pdf_result.metadata: 59 metadata_items = [] 60 for key, value in pdf_result.metadata.items(): 61 if value: 62 metadata_items.append(f"{key}: {value}") 63 if metadata_items: 64 metadata_str = f"\n文档元数据：\n" + "\n".join(metadata_items) 65 66 return f""" 67PDF 文档内容（共 {pdf_result.page_count} 页）： 68{pdf_result.text_content} 69{metadata_str} 70""" 71 72 async def _call_llm_api(self, context: str): 73 """调用LLM API（示例）""" 74 # 这里可以集成OpenAI、文心一言、通义千问等 75 # 返回LLM的响应 76 return "这是基于PDF上下文的回答示例" 77
4.2 使用示例
1import asyncio 2import base64 3 4async def demo_usage(): 5 """使用示例""" 6 parser = PDFParserMiddleware() 7 8 # 示例1：Base64 方式 9 with open("sample.pdf", "rb") as f: 10 base64_content = base64.b64encode(f.read()).decode() 11 12 request1 = PDFParseRequest( 13 source_type=FileSourceType.BASE64, 14 content=base64_content 15 ) 16 17 result1 = await parser.parse_pdf(request1) 18 print(f"Base64 解析结果: {result1.success}, 页数: {result1.page_count}") 19 20 # 示例2：URL 方式 21 request2 = PDFParseRequest( 22 source_type=FileSourceType.URL, 23 content="https://example.com/document.pdf" 24 ) 25 26 # 示例3：集成到对话系统 27 conv_system = ConversationSystem() 28 response = await conv_system.process_user_query( 29 user_id="user123", 30 query="请总结文档的主要内容", 31 pdf_request=request1 32 ) 33 print(f"对话响应: {response}") 34 35if __name__ == "__main__": 36 asyncio.run(demo_usage()) 37
5. 高级功能扩展
5.1 OCR 支持（处理扫描版 PDF）
1import pytesseract 2from pdf2image import convert_from_path 3from PIL import Image 4 5class EnhancedPDFParser(PDFParserMiddleware): 6 """增强版 PDF 解析器，支持 OCR""" 7 8 def __init__(self, temp_dir: str = None, tesseract_path: str = None): 9 super().__init__(temp_dir) 10 if tesseract_path: 11 pytesseract.pytesseract.tesseract_cmd = tesseract_path 12 13 def _extract_text_with_ocr(self, pdf_path: str, language: str = "chi_sim") -> str: 14 """使用 OCR 提取扫描版 PDF 文本""" 15 text_parts = [] 16 17 # 将 PDF 转换为图像 18 images = convert_from_path(pdf_path) 19 20 for page_num, image in enumerate(images, 1): 21 # 使用 Tesseract 进行 OCR 22 page_text = pytesseract.image_to_string(image, lang=language) 23 24 if page_text.strip(): 25 text_parts.append(f"=== 第{page_num}页（OCR识别） ===\n{page_text}\n") 26 else: 27 text_parts.append(f"=== 第{page_num}页 ===\n[OCR未识别到文字]\n") 28 29 return "\n".join(text_parts) 30 31 def _extract_text_from_pdf(self, pdf_path: str, language: str = "chi_sim") -> Tuple[str, int, dict]: 32 """重写提取方法，增加OCR支持""" 33 try: 34 # 先尝试普通文本提取 35 text, page_count, metadata = super()._extract_text_from_pdf(pdf_path) 36 37 # 检查文本提取是否成功 38 if text and "需要OCR处理" not in text: 39 return text, page_count, metadata 40 41 # 如果普通提取失败或需要OCR，使用OCR 42 ocr_text = self._extract_text_with_ocr(pdf_path, language) 43 return ocr_text, page_count, metadata 44 45 except Exception as e: 46 logger.error(f"PDF文本提取失败: {str(e)}") 47 raise 48
5.2 批量处理与进度跟踪
1from typing import List 2from concurrent.futures import ThreadPoolExecutor 3import asyncio 4 5class BatchPDFProcessor: 6 """批量 PDF 处理器""" 7 8 def __init__(self, max_workers: int = 4): 9 self.parser = PDFParserMiddleware() 10 self.executor = ThreadPoolExecutor(max_workers=max_workers) 11 12 async def process_batch(self, requests: List[PDFParseRequest]) -> List[PDFParseResponse]: 13 """批量处理 PDF 文件""" 14 tasks = [] 15 for request in requests: 16 task = asyncio.create_task(self._process_single(request)) 17 tasks.append(task) 18 19 results = await asyncio.gather(*tasks, return_exceptions=True) 20 21 # 处理异常结果 22 processed_results = [] 23 for result in results: 24 if isinstance(result, Exception): 25 processed_results.append(PDFParseResponse( 26 success=False, 27 text_content="", 28 page_count=0, 29 error_message=str(r 30
