Как парсить PDF

BT
210 700 Td  (раму) Tj
100 700 Td  (Мама) Tj
160 700 Td  (мыла) Tj
ET

import asyncio
import os
import sys
from pathlib import Path
from typing import Literal

from dotenv import load_dotenv
from loguru import logger
from openai import AsyncOpenAI
from pdf2image import convert_from_path
from pydantic_ai import Agent, BinaryContent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic import BaseModel, Field

SYSTEM_PROMPT = (
    "You extract PDF page content into structured markdown. "
    "Return strict JSON matching the DocumentPage schema. "
    "Rules: preserve all visible text as markdown; "
    "format tables strictly as markdown tables; "
    "describe images as separate blocks with type=image_description in the correct reading position; "
    "preserve reading order top-to-bottom, left-to-right; "
    "do not skip headings, captions, footnotes, or lists; "
    "do not invent missing text; if uncertain, mark it explicitly in markdown as [Uncertain: ...]."
)


class ContentBlock(BaseModel):
    # Тип блока нужен, чтобы дальше можно было отдельно обрабатывать обычный текст,
    # таблицы и описания изображений (например, для пост-обработки или аналитики).
    type: Literal["text", "table", "image_description"]
    # Markdown хранит финальное представление блока в универсальном формате,
    # который удобно сразу склеивать в итоговый output.md без дополнительного парсинга.
    markdown: str = Field(min_length=1)


class DocumentPage(BaseModel):
    # Номер страницы сохраняем явно, чтобы гарантированно восстановить правильный
    # порядок документа даже при параллельной обработке.
    page_number: int = Field(ge=1)
    # Страница состоит из последовательности блоков; список отражает порядок чтения,
    # что важно для сохранения смысла при рендеринге в markdown.
    blocks: list[ContentBlock] = Field(default_factory=list)


class PdfConverter:
    def convert(self, pdf_path: Path, images_dir: Path) -> list[Path]:
        """Конвертирует PDF в PNG-изображения страниц и возвращает пути к файлам."""
        logger.info(f"Starting PDF conversion: {pdf_path}")

        images_dir.mkdir(parents=True, exist_ok=True)
        pages = convert_from_path(str(pdf_path))
        logger.info(f"Pages found for conversion: {len(pages)}")

        image_paths: list[Path] = []
        for index, page_image in enumerate(pages, start=1):
            output_path = images_dir / f"page_{index:04d}.png"
            page_image.save(output_path, format="PNG")
            image_paths.append(output_path)

        sorted_paths = sorted(image_paths)
        logger.info(f"Images saved to directory: {images_dir}")
        return sorted_paths


class PdfExtractorService:
    def __init__(self, model_name: str) -> None:
        load_dotenv()

        api_key = os.getenv("OPENROUTER_API_KEY")
        if not api_key:
            raise ValueError(
                "OPENROUTER_API_KEY is missing. Create a .env file and add: OPENROUTER_API_KEY=..."
            )

        self.client = AsyncOpenAI(
            base_url="https://openrouter.ai/api/v1",
            api_key=api_key,
        )
        provider = OpenAIProvider(openai_client=self.client)
        model = OpenAIChatModel(model_name=model_name, provider=provider)

        self.agent = Agent[None, DocumentPage](
            model=model,
            output_type=DocumentPage,
            instructions=SYSTEM_PROMPT,
        )

    @staticmethod
    def _render_page_markdown(page: DocumentPage) -> str:
        blocks = [
            block.markdown.strip() for block in page.blocks if block.markdown.strip()
        ]
        if not blocks:
            blocks = ["> [warning] Could not extract content from this page."]

        return f"## Page {page.page_number}\n\n" + "\n\n".join(blocks)

    async def _process_page(
        self,
        image_path: Path,
        page_number: int,
        semaphore: asyncio.Semaphore,
    ) -> tuple[int, str]:
        async with semaphore:
            logger.info(f"Starting page processing {page_number}: {image_path}")

            prompt = [
                (
                    f"page_number={page_number}\n"
                    "Extract this page content from the provided image following the system prompt rules."
                ),
                BinaryContent(data=image_path.read_bytes(), media_type="image/png"),
            ]

            result = await self.agent.run(prompt)
            page = result.output
            if page.page_number != page_number:
                page.page_number = page_number

            return page_number, self._render_page_markdown(page)

    async def extract_to_markdown(
        self, image_paths: list[Path], output_md: Path
    ) -> None:
        semaphore = asyncio.Semaphore(10)
        sorted_paths = sorted(image_paths)

        tasks = [
            asyncio.create_task(self._process_page(image_path, index, semaphore))
            for index, image_path in enumerate(sorted_paths, start=1)
        ]
        logger.info(f"Submitted page processing tasks: {len(tasks)}")

        results = await asyncio.gather(*tasks)

        logger.info(f"Processed pages: {len(results)}/{len(tasks)}")
        ordered_markdown = [
            content for _, content in sorted(results, key=lambda item: item[0])
        ]

        output_md.parent.mkdir(parents=True, exist_ok=True)
        output_md.write_text("\n\n---\n\n".join(ordered_markdown), encoding="utf-8")
        logger.info(f"Final markdown saved to: {output_md}")


async def main() -> None:
    if len(sys.argv) < 2:
        raise ValueError(
            "Pass PDF path as the first argument: uv run python pdf_parser.py ./file.pdf"
        )

    # Эти параметры вынесены в начало main для учебного сценария: новичку проще
    # быстро поменять папки/модель в одном месте и перезапустить скрипт.
    pdf_path = Path(sys.argv[1])
    images_dir = Path("images")
    output_md = Path("output.md")
    model_name = os.getenv("OPENROUTER_MODEL", "openai/gpt-4.1-mini")

    if not pdf_path.exists():
        raise FileNotFoundError(
            f"PDF file not found: {pdf_path}. Please provide a valid path."
        )

    converter = PdfConverter()
    image_paths = converter.convert(pdf_path, images_dir)
    logger.info(f"Conversion complete. Files saved: {len(image_paths)}")

    extractor = PdfExtractorService(model_name=model_name)
    await extractor.extract_to_markdown(image_paths=image_paths, output_md=output_md)


if __name__ == "__main__":
    # Run example:
    # uv run python pdf_parser.py ./path/to/document.pdf
    # Optional env vars:
    # OPENROUTER_API_KEY=your_key
    # OPENROUTER_MODEL=openai/gpt-4.1-mini
    # main асинхронный, поэтому используем asyncio.run: это корректно создает
    # и закрывает event loop для всего конвейера обработки.
    asyncio.run(main())