企业官网建设流程全解析

1. 项目背景与核心问题定位

这个标题指向的是一个Python脚本文件process_pdf.py的修改需求，结合相关热搜词可以判断这是一个涉及PDF处理、PostgreSQL数据库和FastAPI框架的技术项目。从标题编号"1902"和"0121-3"来看，这很可能是一个企业内部或团队协作项目中的代码文件，需要针对特定问题进行修改。

在实际开发中，PDF处理脚本的修改通常涉及以下几个典型场景：

• PDF内容提取逻辑变更（如文本、图片或表格的提取方式）
• 数据库交互层调整（PostgreSQL连接或查询优化）
• API接口规范更新（FastAPI路由或响应格式变更）
• 性能优化需求（大文件处理或并发处理改进）

提示：修改已有PDF处理脚本时，务必先通过git或svn确认文件历史修改记录，避免重复劳动或引入冲突。

2. 必须修改的代码模块分析

2.1 PDF解析功能改造

从热词"python提取pdf中的图片"和"pdf图片中文设置"可以推测，该脚本可能涉及PDF内容提取功能。常见需要修改的部分包括：

# 原版可能使用的PyPDF2基础代码 from PyPDF2 import PdfFileReader def extract_text(pdf_path): with open(pdf_path, 'rb') as f: reader = PdfFileReader(f) text = "" for page in range(reader.numPages): text += reader.getPage(page).extractText() return text

需要升级为更强大的pdfplumber库处理复杂PDF：

import pdfplumber def extract_text_enhanced(pdf_path): text = "" with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: text += page.extract_text(x_tolerance=1, y_tolerance=1) return text

修改要点：

• 增加对中文编码的支持（指定encoding参数）
• 处理扫描件PDF时添加OCR集成
• 改进表格提取逻辑（使用pdfplumber的extract_table()）

2.2 PostgreSQL交互层优化

根据热词"postgresql安装"和"datax同步oracle到postgresql"，数据库操作可能是另一个修改重点。原始代码可能使用基础psycopg2：

import psycopg2 conn = psycopg2.connect( host="localhost", database="mydb", user="postgres", password="password" )

应升级为连接池管理和异步操作：

from psycopg2.pool import ThreadedConnectionPool from contextlib import contextmanager pool = ThreadedConnectionPool( minconn=3, maxconn=10, host="localhost", database="mydb", user="postgres", password="password" ) @contextmanager def get_db_connection(): conn = pool.getconn() try: yield conn finally: pool.putconn(conn)

关键修改：

• 增加连接重试机制
• 添加事务管理装饰器
• 优化批量插入性能（使用copy_from）

3. FastAPI集成改造

热词"FastAPI"表明该脚本可能作为后台服务的一部分。原始实现可能是简单的函数调用：

def process_pdf(file_path): # PDF处理逻辑 return result

需要改造为标准的FastAPI路由：

from fastapi import FastAPI, UploadFile from fastapi.responses import JSONResponse app = FastAPI() @app.post("/process-pdf") async def process_pdf(file: UploadFile): try: contents = await file.read() # 临时保存文件 with open(f"/tmp/{file.filename}", "wb") as f: f.write(contents) # 处理逻辑 result = process_pdf_file(f"/tmp/{file.filename}") return JSONResponse({ "status": "success", "data": result }) except Exception as e: return JSONResponse( {"status": "error", "message": str(e)}, status_code=500 )

必须修改的部分包括：

• 增加文件上传大小限制配置
• 添加异步处理支持
• 完善错误处理机制
• 增加请求验证中间件

4. 性能优化关键修改点

4.1 内存管理改进

处理大PDF文件时常见的内存泄漏问题修改：

# 修改前 def process_large_pdf(path): with open(path, 'rb') as f: reader = PdfFileReader(f) # 一次性加载所有页面 pages = [reader.getPage(i) for i in range(reader.numPages)] # ...处理逻辑 # 修改后 def process_large_pdf(path): with open(path, 'rb') as f: reader = PdfFileReader(f) for i in range(reader.numPages): page = reader.getPage(i) # 逐页处理 # ...处理逻辑 del page # 显式释放内存

4.2 并发处理改造

原始串行处理代码：

def batch_process(files): results = [] for file in files: results.append(process_pdf(file)) return results

应改为多进程池处理：

from multiprocessing import Pool def batch_process(files, workers=4): with Pool(workers) as p: return p.map(process_pdf, files)

注意事项：

• Windows平台需使用ifname== 'main'保护
• 限制最大并发数避免OOM
• 添加任务超时控制

5. 测试与验证方案修改

5.1 单元测试增强

原始可能缺少测试或只有基础测试：

def test_extract_text(): text = extract_text("test.pdf") assert "sample" in text

应扩展为全面的测试套件：

import pytest from unittest.mock import patch @pytest.mark.parametrize("pdf_file,expected", [ ("normal.pdf", {"pages": 3}), ("empty.pdf", {"pages": 0}), ("corrupted.pdf", {"error": True}) ]) def test_pdf_processing(pdf_file, expected): if "error" in expected: with pytest.raises(PDFProcessingError): process_pdf(pdf_file) else: result = process_pdf(pdf_file) assert result["page_count"] == expected["pages"] @patch("psycopg2.connect") def test_db_connection(mock_connect): mock_connect.return_value.cursor.return_value.fetchall.return_value = [("test",)] result = query_db("SELECT * FROM test") assert result == [("test",)]

5.2 集成测试方案

添加使用Docker的测试环境配置：

# test.Dockerfile FROM python:3.9 RUN apt-get update && apt-get install -y \ poppler-utils \ tesseract-ocr COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["pytest", "-v"]

配套的CI配置示例：

# .github/workflows/test.yml jobs: test: runs-on: ubuntu-latest services: postgres: image: postgres:13 env: POSTGRES_PASSWORD: postgres ports: - 5432:5432 steps: - uses: actions/checkout@v2 - run: docker build -f test.Dockerfile -t pdf-processor . - run: docker run --network host pdf-processor

6. 部署配置的必要修改

6.1 依赖管理升级

原始requirements.txt可能简单列出依赖：

PyPDF2==1.26.0 psycopg2==2.8.6

应细分为不同环境需求：

# requirements-core.txt pdfplumber==0.7.4 python-multipart==0.0.5 psycopg2-binary==2.9.3 # requirements-dev.txt -r requirements-core.txt pytest==7.1.2 pytest-cov==3.0.0 # requirements-prod.txt -r requirements-core.txt gunicorn==20.1.0 uvicorn==0.18.2

6.2 配置文件改造

从硬编码配置改为环境变量：

# 修改前 DB_HOST = "localhost" DB_PORT = 5432 # 修改后 import os from pydantic import BaseSettings class Settings(BaseSettings): db_host: str = os.getenv("DB_HOST", "localhost") db_port: int = os.getenv("DB_PORT", 5432) pdf_worker_count: int = os.getenv("PDF_WORKERS", 4) settings = Settings()

配套添加.env文件模板：

# .env.example DB_HOST=your_postgres_host DB_PORT=5432 PDF_WORKERS=4 MAX_FILE_SIZE_MB=50

7. 监控与日志改进

7.1 日志格式标准化

原始可能使用简单print：

print(f"Processing {filename}...")

应改为结构化日志：

import logging from pythonjsonlogger import jsonlogger logger = logging.getLogger("pdf_processor") handler = logging.StreamHandler() formatter = jsonlogger.JsonFormatter( '%(asctime)s %(levelname)s %(name)s %(message)s' ) handler.setFormatter(formatter) logger.addHandler(handler) logger.setLevel(logging.INFO) # 使用示例 logger.info("Processing file", extra={ "filename": filename, "size": os.path.getsize(filename) })

7.2 性能监控集成

添加Prometheus监控端点：

from prometheus_client import start_http_server, Counter, Histogram PDF_PROCESSED = Counter( 'pdf_processed_total', 'Total processed PDF files' ) PROCESS_TIME = Histogram( 'pdf_process_time_seconds', 'Time spent processing PDFs' ) @app.post("/process-pdf") @PROCESS_TIME.time() async def process_pdf(file: UploadFile): PDF_PROCESSED.inc() # ...处理逻辑

配套的Prometheus配置示例：

# prometheus.yml scrape_configs: - job_name: 'pdf_processor' static_configs: - targets: ['localhost:8000']

8. 安全加固必须修改项

8.1 文件上传安全

原始代码可能直接处理上传文件：

@app.post("/upload") async def upload(file: UploadFile): contents = await file.read() # 直接处理

应添加安全检查：

import magic from fastapi import HTTPException ALLOWED_MIME_TYPES = { 'application/pdf': '.pdf', 'application/x-pdf': '.pdf' } @app.post("/upload") async def upload(file: UploadFile): # 检查文件类型 contents = await file.read() mime = magic.from_buffer(contents, mime=True) if mime not in ALLOWED_MIME_TYPES: raise HTTPException(400, "Invalid file type") # 检查文件大小 max_size = 50 * 1024 * 1024 # 50MB if len(contents) > max_size: raise HTTPException(400, "File too large") # 安全保存 safe_name = secure_filename(file.filename) save_path = os.path.join("/secure/uploads", safe_name) with open(save_path, "wb") as f: f.write(contents)

8.2 数据库访问安全

改进SQL注入防护：

# 不安全的方式 cursor.execute(f"SELECT * FROM users WHERE id = {user_id}") # 安全的方式 cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

添加敏感数据加密：

from cryptography.fernet import Fernet key = Fernet.generate_key() cipher_suite = Fernet(key) # 加密 encrypted_text = cipher_suite.encrypt(b"Sensitive data") # 解密 decrypted_text = cipher_suite.decrypt(encrypted_text)

9. 异常处理与容错改进

9.1 自定义异常体系

原始可能使用基础异常：

try: process_pdf(file) except Exception as e: print(f"Error: {e}")

应建立完整的异常处理体系：

class PDFProcessorError(Exception): """Base exception class""" pass class PDFParseError(PDFProcessorError): """PDF解析错误""" pass class DBConnectionError(PDFProcessorError): """数据库连接错误""" pass # 使用示例 try: process_pdf(file) except PDFParseError as e: logger.error(f"PDF解析失败: {e}") raise HTTPException(400, "Invalid PDF format") except DBConnectionError as e: logger.critical("数据库连接失败") raise HTTPException(503, "Service unavailable")

9.2 重试机制实现

添加自动重试装饰器：

from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), retry=retry_if_exception_type(DBConnectionError) ) def query_database(sql, params=None): # 数据库查询逻辑 pass

10. 文档与维护性改进

10.1 代码文档标准化

原始可能缺少文档：

def process(file): # 处理文件 pass

应添加类型提示和docstring：

from typing import Dict, Union from pathlib import Path def process_pdf_file( file_path: Union[str, Path], options: Dict[str, any] = None ) -> Dict[str, any]: """处理PDF文件并提取结构化数据 Args: file_path: PDF文件路径 options: 处理选项字典，包含: - extract_text: bool 是否提取文本 - extract_images: bool 是否提取图片 - ocr: bool 是否启用OCR Returns: 包含提取数据的字典，结构为: { "text": str, "images": List[bytes], "metadata": Dict[str, str] } Raises: PDFParseError: 当PDF解析失败时抛出 FileNotFoundError: 当文件不存在时抛出 """ # 实现逻辑

10.2 变更日志维护

添加规范的CHANGELOG.md：

# Change Log ## [1.1.0] - 2023-06-15 ### Added - 新增PDF表格提取功能 - 添加PostgreSQL连接池支持 ### Changed - 升级pdfplumber替代PyPDF2 - 优化大文件处理内存占用 ### Fixed - 修复中文编码识别问题 - 解决并发写入冲突

配套的版本管理建议：

• 使用semantic versioning (MAJOR.MINOR.PATCH)
• 每个PR必须关联对应的changelog条目
• 重大变更添加迁移指南

在实际修改process_pdf.py时，我通常会先创建一个功能分支，然后通过以下步骤系统性地实施修改：

1. 添加新测试用例覆盖修改需求
2. 进行最小化修改使测试通过
3. 运行完整测试套件
4. 更新相关文档
5. 提交包含详细说明的PR

这种工作流程可以确保修改不会破坏现有功能，同时保持代码库的可维护性。对于特别复杂的修改，我会使用git bisect等工具帮助定位可能引入问题的提交。