app/services/parsers/simple_parser.py

"""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",
            },
        )