pdf-inspector 高级进阶实战:从单库调用到生产级智能解析管线

PDF输入    │    ▼┌─────────────────────────────────────────────────────┐│ 阶段1：分类（pdf-inspector）                          ││ → 分析PDF内部结构（字体编码、文本操作符、图像覆盖率）     ││ → 毫秒级判断每页类型：TextBased / Scanned / Mixed     │└─────────────────────────────────────────────────────┘    │    ├──► TextBased页 ──► 原生提取（跳过GPU，~150ms）    │    └──► Scanned/Mixed页 ──► 阶段2：渲染 → 阶段3：布局检测 → 阶段4：GPU处理 → 阶段5：后处理

import { classifyPdf, extractTextInRegions, PageRegions } from '@firecrawl/pdf-inspector';import { readFileSync } from 'fs';/** * 智能区域处理器：仅对需要OCR的区域调用服务 * 适用于Mixed类型PDF（如扫描文档中夹杂少量文字区域） */async function smartRegionProcessor(pdfPath: string, layoutModelOutput: any[]) {    const pdfBuffer = readFileSync(pdfPath);    // 步骤1：全文档分类（确定整体策略）    const classification = classifyPdf(pdfBuffer);    console.log(`PDF类型: ${classification.pdfType}`);    console.log(`需要OCR的页面: ${classification.pagesNeedingOcr}`);    // 步骤2：如果全部文字页，直接用全页提取（不走混合管线）    if (classification.pdfType === 'TextBased') {        const fullPageRegions: PageRegions[] = [];        for (let i = 0; i < classification.pageCount; i++) {            fullPageRegions.push({                page: i,                regions: [[0, 0, 612, 792]]  // 标准Letter/A4尺寸            });        }        const results = extractTextInRegions(pdfBuffer, fullPageRegions);        // 将全部结果拼接后返回        return results.flatMap(p => p.regions.map(r => r.text)).join('\n');    }    // 步骤3：混合型PDF — 配合布局模型结果做区域处理    // layoutModelOutput 是外部布局检测模型的输出，格式类似：    // [{ page: 0, bbox: [x1, y1, x2, y2], type: 'text' }, ...]    const regionRequests: PageRegions[] = [];    for (const pageLayout of layoutModelOutput) {        const textRegions = pageLayout.regions.filter(r => r.type === 'text');        if (textRegions.length > 0) {            regionRequests.push({                page: pageLayout.page,                regions: textRegions.map(r => r.bbox)            });        }    }    // 执行区域提取    const regionResults = extractTextInRegions(pdfBuffer, regionRequests);    // 步骤4：路由决策 — 仅将needsOcr的区域送到OCR    const ocrQueue = [];    const extractedTexts = [];    for (const pageResult of regionResults) {        for (const region of pageResult.regions) {            if (region.needsOcr) {                ocrQueue.push({                    page: pageResult.page,                    region: region,  // 坐标信息可传递给OCR服务                });                console.log(`页面 ${pageResult.page} 区域 ${JSON.stringify(region)} 需要OCR`);            } else {                extractedTexts.push(region.text);            }        }    }    // 步骤5：并行处理OCR队列（使用外部OCR服务）    // const ocrResults = await processOCRQueue(ocrQueue);    return {        extractedText: extractedTexts.join('\n'),        ocrRequiredRegions: ocrQueue,        stats: {            totalRegions: regionRequests.length,            ocrFallbackCount: ocrQueue.length        }    };}

import pdf_inspectorfrom typing import List, Tupledef extract_with_layout_awareness(    pdf_path: str,    layout_regions: List[dict]  # 外部布局模型输出的区域列表) -> dict:    """    配合布局检测模型进行智能区域提取    Args:        pdf_path: PDF文件路径        layout_regions: 布局模型输出，格式为            [{"page": 0, "type": "text", "bbox": [x1, y1, x2, y2]}, ...]    """    # 读取PDF    with open(pdf_path, "rb") as f:        pdf_bytes = f.read()    # 先做分类决策    detection = pdf_inspector.detect_pdf_bytes(pdf_bytes)    # 纯文本PDF直接用全量处理    if detection.pdf_type == "text_based":        result = pdf_inspector.process_pdf_bytes(pdf_bytes)        return {            "markdown": result.markdown,            "ocr_needed": False,            "pages_needing_ocr": []        }    # 混合/扫描型PDF：按区域提取    # 按页面分组布局区域    page_to_regions = {}    for region in layout_regions:        page = region["page"]        bbox = region["bbox"]  # [x1, y1, x2, y2]        if page not in page_to_regions:            page_to_regions[page] = []        page_to_regions[page].append(bbox)    # 执行区域提取（需要配合PDF解析库，目前pdf_inspector Python绑定暂时不直接暴露region API，    # 可以结合detect_pdf的pages_needing_ocr结果做粗略分页处理）    #    # 核心策略：对detect_pdf检测出的pages_needing_ocr页面，送入OCR；    # 对可信文字页，用extract_text直接提取。    need_ocr_pages = set(detection.pages_needing_ocr)    results = []    for page_num in range(detection.page_count):        if page_num in need_ocr_pages:            # 该页需要OCR处理（整页或更精细区域可后续接入布局模型）            results.append({                "page": page_num,                "method": "ocr",                "content": None,  # OCR结果由外部服务填充                "needs_ocr": True            })        else:            # 该页为可信文字页，直接提取            # 提取第page_num页的文本（过程需要调整，简化示例）            text = pdf_inspector.extract_text_bytes(pdf_bytes)            results.append({                "page": page_num,                "method": "direct",                "content": text,                "needs_ocr": False            })    return {        "pages": results,        "ocr_pages_list": detection.pages_needing_ocr,        "confidence": detection.confidence    }

def smart_routing_with_confidence(pdf_path: str, ocr_service, ocr_threshold: float = 0.7):    """基于置信度的三级路由策略"""    detection = pdf_inspector.detect_pdf(pdf_path)    # 策略矩阵    if detection.pdf_type == "text_based":        if detection.confidence > 0.95:            # 高置信度纯文本：直接走本地提取，相信结果质量            return pdf_inspector.process_pdf(pdf_path)        elif detection.confidence > ocr_threshold:            # 中等置信度：提取后做质量校验            result = pdf_inspector.process_pdf(pdf_path)            if result.has_encoding_issues or len(result.markdown or "") < 100:                # 疑似有问题，降级到OCR                return ocr_service.extract(pdf_path)            return result        else:            # 低置信度但被标记为text_based：可能有字体编码问题            return ocr_service.extract(pdf_path)    elif detection.pdf_type == "mixed":        # 混合型PDF：pages_needing_ocr 给出精确路由        if len(detection.pages_needing_ocr) / detection.page_count < 0.3:            # 仅少数页需要OCR，分页处理            return process_mixed_with_ocr_fallback(pdf_path, detection.pages_needing_ocr)        else:            # 大量页需要OCR，直接整文档送入OCR            return ocr_service.extract(pdf_path)    else:  # scanned / image_based        return ocr_service.extract(pdf_path)

use pdf_inspector::process_pdf;use rayon::prelude::*;use std::path::PathBuf;use std::sync::Arc;#[derive(Debug)]struct ProcessedDocument {    path: String,    pdf_type: String,    markdown: Option<String>,    processing_time_ms: u64,    error: Option<String>,}fnbatch_process_parallel(pdf_paths: Vec<PathBuf>) -> Vec<ProcessedDocument> {    // 使用Rayon自动并行处理（利用所有CPU核心）    pdf_paths        .par_iter()        .map(|path| {            let start = std::time::Instant::now();            match process_pdf(path.to_str().unwrap()) {                Ok(result) => ProcessedDocument {                    path: path.display().to_string(),                    pdf_type: format!("{:?}", result.pdf_type),                    markdown: result.markdown,                    processing_time_ms: start.elapsed().as_millis() as u64,                    error: None,                },                Err(e) => ProcessedDocument {                    path: path.display().to_string(),                    pdf_type: "error".to_string(),                    markdown: None,                    processing_time_ms: start.elapsed().as_millis() as u64,                    error: Some(e.to_string()),                },            }        })        .collect()}

use tokio::fs::File;use tokio::io::AsyncReadExt;use pdf_inspector::process_pdf_bytes;async fn async_process_pdf(path: &str) -> Result<(), Box<dyn std::error::Error>> {    let mut file = File::open(path).await?;    let mut buffer = Vec::new();    file.read_to_end(&mut buffer).await?;    let result = tokio::task::spawn_blocking(move || {        // process_pdf_bytes会解析PDF，是CPU密集型操作        // 使用spawn_blocking将其移出异步运行时        process_pdf_bytes(&buffer)    }).await??;    println!("PDF类型: {:?}", result.pdf_type);    Ok(())}

use lopdf::Document;use std::collections::HashMap;use std::sync::RwLock;struct DocumentCache {    cache: RwLock<HashMap<String, Document>>,}impl DocumentCache {    fn get_or_parse(&self, path: &str) -> Result<Document, Box<dyn std::error::Error>> {        {            let read_lock = self.cache.read().unwrap();            if let Some(doc) = read_lock.get(path) {                return Ok(doc.clone());            }        }        // 缓存未命中，解析并插入        let doc = Document::load(path)?;        let mut write_lock = self.cache.write().unwrap();        write_lock.insert(path.to_string(), doc.clone());        Ok(doc)    }}

import pdf_inspectorimport redef enhanced_heading_detection(pdf_path: str):    """在pdf-inspector输出的基础上做标题二次增强"""    result = pdf_inspector.process_pdf(pdf_path)    # 1. 获取字体统计信息    items = pdf_inspector.extract_text_with_positions(pdf_path)    # 2. 计算全文档的字体大小分布    font_sizes = [item.font_size for item in items if item.font_size > 0]    if not font_sizes:        return result.markdown    avg_font_size = sum(font_sizes) / len(font_sizes)    body_font_size = sorted(font_sizes)[len(font_sizes) // 3]  # 近似正文大小    # 3. 对Markdown内容做二次后处理    # 策略：任何字体 >= body_font_size*1.2 且独立成行的文本，如果未被标记为H1-H4，手动补充标记    lines = result.markdown.split('\n')    enhanced_lines = []    for line in lines:        # 查找未被标准库标记的可能标题        # 简化实现：检测行首无#标记的短文本行        if line.strip() and not line.startswith('#'):            # 通过查找对应文本项的字体大小决定是否提升为标题            # （实际实现需要维护行号到TextItem的映射，此处为示意）            enhanced_lines.append(line)        else:            enhanced_lines.append(line)    return '\n'.join(enhanced_lines)

import asyncioimport pdf_inspectorfrom typing import List, Tupleclass SmartOCRClient:    """    智能OCR客户端：使用pdf-inspector做前置判断，仅在必要时调用OCR服务    """    def __init__(self, ocr_service, confidence_threshold: float = 0.7):        self.ocr_service = ocr_service        self.confidence_threshold = confidence_threshold    async def process_document(self, pdf_path: str) -> dict:        # 步骤1：快速分类（<50ms）        detection = pdf_inspector.detect_pdf(pdf_path)        # 步骤2：根据类型决定处理路径        if detection.pdf_type == "text_based" and detection.confidence > self.confidence_threshold:            # 高置信度文本型PDF：本地提取            result = pdf_inspector.process_pdf(pdf_path)            return {                "source": "pdf_inspector",                "content": result.markdown,                "processing_time_ms": result.processing_time_ms,                "needs_ocr": False            }        elif detection.pdf_type == "mixed":            # 混合型：分页路由，仅将pages_needing_ocr传入OCR            tasks = []            ocr_pages = set(detection.pages_needing_ocr)            for page_num in range(detection.page_count):                if page_num in ocr_pages:                    tasks.append(self._ocr_page(pdf_path, page_num))                else:                    tasks.append(self._direct_extract_page(pdf_path, page_num))            page_results = await asyncio.gather(*tasks)            return {                "source": "hybrid",                "pages": page_results,                "needs_ocr": len(ocr_pages) > 0,                "stats": {                    "total_pages": detection.page_count,                    "ocr_pages": len(ocr_pages),                    "confidence": detection.confidence                }            }        else:  # scanned / image_based / low_confidence            # 全文档OCR            ocr_result = await self.ocr_service.extract_async(pdf_path)            return {                "source": "ocr_service",                "content": ocr_result.text,                "needs_ocr": True            }    async def _ocr_page(self, pdf_path: str, page_num: int):        """单页OCR"""        # 具体实现取决于OCR服务的API        return await self.ocr_service.process_page(pdf_path, page_num)    async def _direct_extract_page(self, pdf_path: str, page_num: int):        """单页直接提取"""        # 需要分页API，目前pdf_inspector已支持extract_pages_markdown        result = pdf_inspector.extract_pages_markdown(pdf_path, pages=[page_num])        return {            "page": page_num,            "content": result.pages[0].markdown,            "source": "direct"        }

def post_process_tables(markdown_content: str) -> str:    """    对pdf-inspector输出的Markdown表格进行后处理增强    """    lines = markdown_content.split('\n')    result_lines = []    i = 0    while i < len(lines):        line = lines[i]        # 检测可能是表格分割线但未被识别的情况        # 例如：连续的 | ... | 模式        if '|' in line and line.count('|') >= 3:            # 检查下一行是否是分隔符模式（如 |---|）            if i + 1 < len(lines) and '---' in lines[i + 1]:                # 已被标准库正确识别，直接保留                result_lines.append(line)            else:                # 可能是一个误报，或者格式不规范的表格                # 可选：调用外部表格修复逻辑                result_lines.append(line)        else:            result_lines.append(line)        i += 1    return '\n'.join(result_lines)

# 多阶段构建：分离编译环境与运行环境，减小最终镜像体积FROM rust:1.75-slim as builder# 安装Python编译依赖（Python绑定需要）RUN apt-get update && apt-get install -y python3-dev python3-pipWORKDIR /appCOPY . .# 构建Python绑定RUN pip3 install maturin && maturin build --release --out dist# 运行时镜像FROM python:3.11-slimCOPY --from=builder /app/dist/*.whl /tmp/RUN pip install /tmp/*.whl && rm -rf /tmp/*.whlCOPY ./app.py /app/WORKDIR /appCMD ["python", "app.py"]

import signalfrom contextlib import contextmanagerclass TimeoutError(Exception):    pass@contextmanagerdef timeout(seconds: int):    """简单超时实现"""    def handler(signum, frame):        raise TimeoutError("处理超时")    original_handler = signal.signal(signal.SIGALRM, handler)    signal.alarm(seconds)    try:        yield    finally:        signal.alarm(0)        signal.signal(signal.SIGALRM, original_handler)def robust_process(pdf_path: str, ocr_fallback_service):    """带超时和降级的鲁棒处理"""    try:        with timeout(5):  # 5秒总时长限制            detection = pdf_inspector.detect_pdf(pdf_path)            if detection.pdf_type == "text_based":                return pdf_inspector.process_pdf(pdf_path)            else:                # 时间不足时直接降级到OCR                return ocr_fallback_service.extract(pdf_path)    except TimeoutError:        print(f"处理超时，降级到OCR: {pdf_path}")        return ocr_fallback_service.extract(pdf_path)    except Exception as e:        print(f"处理失败，降级到OCR: {e}")        return ocr_fallback_service.extract(pdf_path)