PDF解析实现

1. 项目背景与需求分析在现代Web应用中PDF 文档处理是一个常见但复杂的需求。无论是企业 OA 系统、在线教育平台还是知识管理工具都需要能够高效解析 PDF 内容。本次作业要求实现一个中间件专门处理前端上传的PDF文件具体要求如下输入格式接收 Base64 编码的 PDF 数据核心功能将 Base64 转换为正常 PDF 文件并提取文本内容输出要求将提取的文本内容完整放入 system_message 中作为后续对话的上下文扩展目标最终版本应支持上传或发送包含完整路径信息的任何文件包括 URL 地址都能进行合理的解析和处理2. 技术栈选择2.1 后端框架FastAPI现代、高性能的 Python Web 框架支持异步处理Pydantic数据验证和设置管理2.2 PDF处理库PyPDF2/pdfplumberPDF文本提取pdf2imagePDF转图像可选用于OCR场景base64标准库用于Base64编解码2.3 文件处理aiofiles异步文件操作httpx异步 HTTP 客户端用于 URL 下载2.4 环境配置# requirements.txtfastapi0.104.1uvicorn0.24.0pydantic2.5.0PyPDF23.0.1pdfplumber0.10.3aiofiles23.2.1httpx0.25.1python-multipart0.0.63. 核心中间件实现3.1 基础模型定义frompydanticimportBaseModel,FieldfromtypingimportOptional,UnionfromenumimportEnumclassFileSourceType(str,Enum):BASE64base64URLurlLOCAL_PATHlocal_pathUPLOADuploadclassPDFParseRequest(BaseModel):PDF 解析请求模型source_type:FileSourceTypeField(...,description文件来源类型)content:Optional[str]Field(None,descriptionBase64 内容或 URL 路径)file_path:Optional[str]Field(None,description本地文件路径)file_upload:Optional[bytes]Field(None,description上传的文件二进制数据)include_metadata:boolField(True,description是否包含 PDF 元数据)language:strField(chi_sim,descriptionOCR 语言设置如需要)classPDFParseResponse(BaseModel):PDF 解析响应模型success:booltext_content:strpage_count:intmetadata:Optional[dict]Noneerror_message:Optional[str]None3.2 中间件核心类importbase64importtempfileimportosfrompathlibimportPathimportpdfplumberimporthttpxfromtypingimportTuple,Optionalimportlogging loggerlogging.getLogger(__name__)classPDFParserMiddleware:PDF解析中间件def__init__(self,temp_dir:strNone): 初始化PDF解析中间件 Args: temp_dir: 临时文件目录默认为系统临时目录 self.temp_dirtemp_dirortempfile.gettempdir()Path(self.temp_dir).mkdir(parentsTrue,exist_okTrue)asyncdefparse_pdf(self,request:PDFParseRequest)-PDFParseResponse: 解析PDF文件的主入口方法 Args: request: PDF 解析请求 Returns: PDFParseResponse: 解析结果 try:# 根据来源类型获取PDF文件路径pdf_pathawaitself._get_pdf_file(request)# 提取文本内容text_content,page_count,metadataself._extract_text_from_pdf(pdf_path)# 清理临时文件self._cleanup_temp_file(pdf_path,request.source_type)returnPDFParseResponse(successTrue,text_contenttext_content,page_countpage_count,metadatametadata)exceptExceptionase:logger.error(fPDF 解析失败:{str(e)},exc_infoTrue)returnPDFParseResponse(successFalse,text_content,page_count0,error_messagestr(e))asyncdef_get_pdf_file(self,request:PDFParseRequest)-str: 根据来源类型获取PDF文件路径 Returns: str: 本地PDF文件路径 ifrequest.source_typeFileSourceType.BASE64:returnawaitself._save_base64_to_file(request.content)elifrequest.source_typeFileSourceType.URL:returnawaitself._download_from_url(request.content)elifrequest.source_typeFileSourceType.LOCAL_PATH:returnrequest.file_pathelifrequest.source_typeFileSourceType.UPLOAD:returnawaitself._save_upload_to_file(request.file_upload)else:raiseValueError(f不支持的来源类型:{request.source_type})asyncdef_save_base64_to_file(self,base64_content:str)-str:将Base64内容保存为临时PDF文件ifnotbase64_content:raiseValueError(Base64内容不能为空)# 移除可能的data URL前缀if,inbase64_content:base64_contentbase64_content.split(,)[1]# 解码Base64pdf_bytesbase64.b64decode(base64_content)# 保存到临时文件temp_filetempfile.NamedTemporaryFile(suffix.pdf,dirself.temp_dir,deleteFalse)temp_file.write(pdf_bytes)temp_file.close()returntemp_file.nameasyncdef_download_from_url(self,url:str)-str:从URL下载PDF文件ifnoturl:raiseValueError(URL不能为空)asyncwithhttpx.AsyncClient()asclient:responseawaitclient.get(url)response.raise_for_status()# 保存到临时文件temp_filetempfile.NamedTemporaryFile(suffix.pdf,dirself.temp_dir,deleteFalse)temp_file.write(response.content)temp_file.close()returntemp_file.nameasyncdef_save_upload_to_file(self,file_data:bytes)-str:保存上传的文件数据到临时文件ifnotfile_data:raiseValueError(上传文件数据不能为空)temp_filetempfile.NamedTemporaryFile(suffix.pdf,dirself.temp_dir,deleteFalse)temp_file.write(file_data)temp_file.close()returntemp_file.namedef_extract_text_from_pdf(self,pdf_path:str)-Tuple[str,int,dict]: 从PDF文件中提取文本内容 Returns: Tuple[str, int, dict]: (文本内容, 页数, 元数据) text_parts[]metadata{}page_count0try:withpdfplumber.open(pdf_path)aspdf:page_countlen(pdf.pages)metadatapdf.metadataforpage_num,pageinenumerate(pdf.pages,1):page_textpage.extract_text()ifpage_text:text_parts.append(f 第{page_num}页 \n{page_text}\n)else:# 如果无法提取文本可能是扫描版PDFtext_parts.append(f 第{page_num}页 \n[扫描图像需要OCR处理]\n)exceptExceptionase:logger.warning(fpdfplumber 提取失败尝试 PyPDF2:{str(e)})# 回退到PyPDF2importPyPDF2withopen(pdf_path,rb)asfile:pdf_readerPyPDF2.PdfReader(file)page_countlen(pdf_reader.pages)metadatapdf_reader.metadataforpage_num,pageinenumerate(pdf_reader.pages,1):page_textpage.extract_text()ifpage_text:text_parts.append(f 第{page_num}页 \n{page_text}\n)full_text\n.join(text_parts)returnfull_text,page_count,metadatadef_cleanup_temp_file(self,file_path:str,source_type:FileSourceType):清理临时文件除本地路径外ifsource_type!FileSourceType.LOCAL_PATH:try:os.unlink(file_path)except:pass3.3 FastAPI路由集成fromfastapiimportFastAPI,UploadFile,File,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefromcontextlibimportasynccontextmanagerimportuvicorn# 创建应用实例appFastAPI(titlePDF 解析中间件 API)# 添加CORS中间件app.add_middleware(CORSMiddleware,allow_origins[*],allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 全局PDF解析器实例pdf_parserPDFParserMiddleware()asynccontextmanagerasyncdeflifespan(app:FastAPI):应用生命周期管理# 启动时初始化print(PDF解析中间件服务启动...)yield# 关闭时清理print(PDF解析中间件服务关闭...)appFastAPI(lifespanlifespan)app.post(/api/parse/pdf,response_modelPDFParseResponse)asyncdefparse_pdf(request:PDFParseRequest): 解析 PDF 文件接口 - 支持 Base64、URL、本地路径、文件上传多种方式 - 返回提取的文本内容和元数据 returnawaitpdf_parser.parse_pdf(request)app.post(/api/parse/upload,response_modelPDFParseResponse)asyncdefparse_upload_pdf(file:UploadFileFile(...,description上传的PDF文件),include_metadata:boolTrue): 通过文件上传方式解析 PDF - 支持 multipart/form-data 上传 - 自动检测文件类型 ifnotfile.filename.lower().endswith(.pdf):raiseHTTPException(400,只支持PDF文件)contentawaitfile.read()requestPDFParseRequest(source_typeFileSourceType.UPLOAD,file_uploadcontent,include_metadatainclude_metadata)returnawaitpdf_parser.parse_pdf(request)app.get(/api/health)asyncdefhealth_check():健康检查接口return{status:healthy,service:pdf-parser-middleware}4. 系统集成与上下文管理4.1 集成到对话系统classConversationSystem:对话系统集成示例def__init__(self):self.pdf_parserPDFParserMiddleware()self.system_messages{}asyncdefprocess_user_query(self,user_id:str,query:str,pdf_request:PDFParseRequestNone): 处理用户查询支持PDF上下文 Args: user_id: 用户ID query: 用户查询文本 pdf_request: PDF解析请求可选 system_message# 如果有PDF解析请求先解析PDFifpdf_request:pdf_resultawaitself.pdf_parser.parse_pdf(pdf_request)ifpdf_result.success:# 将PDF内容作为系统消息上下文system_messageself._format_pdf_context(pdf_result)self.system_messages[user_id]system_messageelse:return{error:fPDF解析失败:{pdf_result.error_message},query:query}# 如果有历史系统消息合并ifuser_idinself.system_messages:system_messageself.system_messages[user_id]# 构建完整的对话上下文full_contextf 系统上下文来自PDF文档{system_message}用户查询{query}请基于以上上下文回答用户问题。 # 这里可以调用LLM接口responseawaitself._call_llm_api(full_context)return{response:response,has_pdf_context:bool(system_message)}def_format_pdf_context(self,pdf_result:PDFParseResponse)-str:格式化PDF内容为系统消息metadata_strifpdf_result.metadata:metadata_items[]forkey,valueinpdf_result.metadata.items():ifvalue:metadata_items.append(f{key}:{value})ifmetadata_items:metadata_strf\n文档元数据\n\n.join(metadata_items)returnf PDF 文档内容共{pdf_result.page_count}页{pdf_result.text_content}{metadata_str}asyncdef_call_llm_api(self,context:str):调用LLM API示例# 这里可以集成OpenAI、文心一言、通义千问等# 返回LLM的响应return这是基于PDF上下文的回答示例4.2 使用示例importasyncioimportbase64asyncdefdemo_usage():使用示例parserPDFParserMiddleware()# 示例1Base64 方式withopen(sample.pdf,rb)asf:base64_contentbase64.b64encode(f.read()).decode()request1PDFParseRequest(source_typeFileSourceType.BASE64,contentbase64_content)result1awaitparser.parse_pdf(request1)print(fBase64 解析结果:{result1.success}, 页数:{result1.page_count})# 示例2URL 方式request2PDFParseRequest(source_typeFileSourceType.URL,contenthttps://example.com/document.pdf)# 示例3集成到对话系统conv_systemConversationSystem()responseawaitconv_system.process_user_query(user_iduser123,query请总结文档的主要内容,pdf_requestrequest1)print(f对话响应:{response})if__name____main__:asyncio.run(demo_usage())5. 高级功能扩展5.1 OCR 支持处理扫描版 PDFimportpytesseractfrompdf2imageimportconvert_from_pathfromPILimportImageclassEnhancedPDFParser(PDFParserMiddleware):增强版 PDF 解析器支持 OCRdef__init__(self,temp_dir:strNone,tesseract_path:strNone):super().__init__(temp_dir)iftesseract_path:pytesseract.pytesseract.tesseract_cmdtesseract_pathdef_extract_text_with_ocr(self,pdf_path:str,language:strchi_sim)-str:使用 OCR 提取扫描版 PDF 文本text_parts[]# 将 PDF 转换为图像imagesconvert_from_path(pdf_path)forpage_num,imageinenumerate(images,1):# 使用 Tesseract 进行 OCRpage_textpytesseract.image_to_string(image,langlanguage)ifpage_text.strip():text_parts.append(f 第{page_num}页OCR识别 \n{page_text}\n)else:text_parts.append(f 第{page_num}页 \n[OCR未识别到文字]\n)return\n.join(text_parts)def_extract_text_from_pdf(self,pdf_path:str,language:strchi_sim)-Tuple[str,int,dict]:重写提取方法增加OCR支持try:# 先尝试普通文本提取text,page_count,metadatasuper()._extract_text_from_pdf(pdf_path)# 检查文本提取是否成功iftextand需要OCR处理notintext:returntext,page_count,metadata# 如果普通提取失败或需要OCR使用OCRocr_textself._extract_text_with_ocr(pdf_path,language)returnocr_text,page_count,metadataexceptExceptionase:logger.error(fPDF文本提取失败:{str(e)})raise5.2 批量处理与进度跟踪fromtypingimportListfromconcurrent.futuresimportThreadPoolExecutorimportasyncioclassBatchPDFProcessor:批量 PDF 处理器def__init__(self,max_workers:int4):self.parserPDFParserMiddleware()self.executorThreadPoolExecutor(max_workersmax_workers)asyncdefprocess_batch(self,requests:List[PDFParseRequest])-List[PDFParseResponse]:批量处理 PDF 文件tasks[]forrequestinrequests:taskasyncio.create_task(self._process_single(request))tasks.append(task)resultsawaitasyncio.gather(*tasks,return_exceptionsTrue)# 处理异常结果processed_results[]forresultinresults:ifisinstance(result,Exception):processed_results.append(PDFParseResponse(successFalse,text_content,page_count0,error_messagestr(r