Spaces:
Sleeping
Sleeping
File size: 2,727 Bytes
8420e1e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """SimpleParser - 零依赖的 PDF 文本提取 fallback.
适用场景:
- Docling 失败 / 模型下载卡住 / Docling 不可用
- 文本型 PDF (非扫描件) - 用 pypdf 直接抽文字
特性:
- 零 ML 依赖 (只有 pypdf)
- 快速, 内存友好
- 拿不到结构 (无表格识别), 只做 "能分页 + 拿文本"
这是 marker-pdf 的替代方案, 因为 marker-pdf 1.10+ 与 pydantic-ai 的 anthropic 版本冲突.
"""
from __future__ import annotations
import asyncio
import logging
import time
from pathlib import Path
from app.services.parsers.base_parser import BaseParser, PageContent, ParsedDocument
logger = logging.getLogger(__name__)
class SimpleParser(BaseParser):
"""pypdf-based 轻量级 PDF 解析器. 兜底中的兜底."""
name = "simple"
def supported_extensions(self) -> set[str]:
# 只支持 PDF; 其它格式 (docx/png) 还是走 Docling
return {".pdf"}
async def parse(self, file_path: Path) -> ParsedDocument:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, self._parse_sync, file_path)
def _parse_sync(self, file_path: Path) -> ParsedDocument:
started = time.time()
logger.info("SimpleParser (pypdf) parsing: %s", file_path.name)
try:
from pypdf import PdfReader
except ImportError as e:
raise RuntimeError(
"pypdf not installed. Add 'pypdf>=4.0' to requirements.txt"
) from e
reader = PdfReader(str(file_path))
pages: list[PageContent] = []
page_texts: list[str] = []
for idx, page in enumerate(reader.pages):
try:
text = page.extract_text() or ""
except Exception as e: # noqa: BLE001
logger.warning("pypdf extract_text failed on page %d: %s", idx, e)
text = ""
pages.append(PageContent(
page_no=idx + 1,
text=text,
tables=[], # simple parser 不识别表格
images=[],
))
page_texts.append(text)
# 全文 markdown (无结构, 直接拼)
markdown = "\n\n".join(page_texts)
elapsed_ms = int((time.time() - started) * 1000)
logger.info(
"SimpleParser done: %s pages, %dms (using pypdf)",
len(pages), elapsed_ms,
)
return ParsedDocument(
markdown=markdown,
pages=pages,
tables=[],
images=[],
meta={
"parser": self.name,
"page_count": len(pages),
"elapsed_ms": elapsed_ms,
"backend": "pypdf",
},
)
|