3. Document Processing

Description

The first component of the Document Analyzer is the DocumentProcessor: the layer that receives a raw file (PDF or image) and transforms it into data the other modules can consume. Without this component, the VisionAnalyzer has no images to analyze, the RAGModule has no text to index, and the Summarizer has no content to summarize.

Why it matters: Document processing is the foundation of the whole pipeline. A PDF can have pages with selectable text (digitally generated), scanned pages (images), or a mix of both. Correctly detecting the type of each page determines whether text is extracted directly (fast, free) or sent to a Vision API (slow, expensive). An error here propagates to all downstream modules.

Connection with the module: In Module 3 you built individual functions to extract text and images. Here you integrate them into a DocumentProcessor class that encapsulates all the processing logic, handles edge cases (mixed PDFs, corrupt images, large files), and produces a standardized ProcessedDocument that the rest of the system consumes.

Processing Pipeline

Flow for PDFs

PDF → Open with PyMuPDF → Iterate pages
    │
    ├─── Page with text (>50 chars) → Extract text → PageContent(type="text")
    │
    ├─── Scanned page (<50 chars) → Render to image → Base64 → PageContent(type="image")
    │
    └─── Page with embedded images → Extract images → Base64 → PageContent(type="image")

ProcessedDocument with a list of PageContent + metadata

Flow for images

Image → Verify format → Read bytes → Base64 → PageContent(type="image")

ProcessedDocument with 1 PageContent + metadata

Decision: text vs image per page

The key heuristic is: does the page have enough extractable text?

TEXT_THRESHOLD = 50  # minimum characters to consider a page as "text"

If len(page.get_text().strip()) > TEXT_THRESHOLD, the page has native text. If not, it's scanned or an image and needs visual processing.

This heuristic works well for 95% of cases. Edge cases (pages with only numbers or with text in embedded images) are handled with additional logic.

Data Models

Models produced by the DocumentProcessor

from pydantic import BaseModel
from typing import Optional


class PageContent(BaseModel):
    page_number: int
    content_type: str  # "text" | "image"
    text: Optional[str] = None
    image_base64: Optional[str] = None
    char_count: int = 0
    image_size_bytes: int = 0


class ProcessedDocument(BaseModel):
    file_path: str
    file_type: str  # "pdf" | "image"
    total_pages: int
    pages: list[PageContent]
    full_text: Optional[str] = None
    has_text_pages: bool = False
    has_image_pages: bool = False
    text_page_count: int = 0
    image_page_count: int = 0

These models are the contract between DocumentProcessor and all downstream modules. VisionAnalyzer consumes the pages with content_type="image". RAGModule consumes the pages with content_type="text" and the Vision-generated descriptions for the images.

Implementation: DocumentProcessor

Complete class

import base64
import logging
from pathlib import Path
from typing import Optional

import fitz

logger = logging.getLogger(__name__)

TEXT_THRESHOLD = 50
IMAGE_DPI_SCALE = 150 / 72  # 150 DPI for page rendering
SUPPORTED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp"}


class DocumentProcessor:
    def __init__(self, text_threshold: int = TEXT_THRESHOLD, dpi_scale: float = IMAGE_DPI_SCALE):
        self.text_threshold = text_threshold
        self.dpi_scale = dpi_scale

    def process(self, file_path: str) -> ProcessedDocument:
        path = Path(file_path)

        if path.suffix.lower() == ".pdf":
            return self._process_pdf(file_path)
        elif path.suffix.lower() in SUPPORTED_IMAGE_EXTENSIONS:
            return self._process_image(file_path)
        else:
            raise ValueError(f"Unsupported format: {path.suffix}")

    def _process_pdf(self, file_path: str) -> ProcessedDocument:
        doc = fitz.open(file_path)
        pages: list[PageContent] = []
        full_text_parts: list[str] = []
        text_count = 0
        image_count = 0

        try:
            for i in range(len(doc)):
                page = doc[i]
                text = page.get_text()

                if len(text.strip()) > self.text_threshold:
                    pages.append(PageContent(
                        page_number=i + 1,
                        content_type="text",
                        text=text,
                        char_count=len(text)
                    ))
                    full_text_parts.append(text)
                    text_count += 1
                else:
                    image_b64 = self._page_to_base64(page)
                    pages.append(PageContent(
                        page_number=i + 1,
                        content_type="image",
                        image_base64=image_b64,
                        image_size_bytes=len(image_b64) * 3 // 4
                    ))
                    image_count += 1

                embedded = self._extract_embedded_images(page, i + 1)
                pages.extend(embedded)
                image_count += len(embedded)

        finally:
            doc.close()

        full_text = "\n\n".join(full_text_parts) if full_text_parts else None

        return ProcessedDocument(
            file_path=file_path,
            file_type="pdf",
            total_pages=len(doc),
            pages=pages,
            full_text=full_text,
            has_text_pages=text_count > 0,
            has_image_pages=image_count > 0,
            text_page_count=text_count,
            image_page_count=image_count
        )

    def _process_image(self, file_path: str) -> ProcessedDocument:
        with open(file_path, "rb") as f:
            raw = f.read()

        b64 = base64.b64encode(raw).decode()

        page = PageContent(
            page_number=1,
            content_type="image",
            image_base64=b64,
            image_size_bytes=len(raw)
        )

        return ProcessedDocument(
            file_path=file_path,
            file_type="image",
            total_pages=1,
            pages=[page],
            full_text=None,
            has_text_pages=False,
            has_image_pages=True,
            text_page_count=0,
            image_page_count=1
        )

    def _page_to_base64(self, page: fitz.Page) -> str:
        mat = fitz.Matrix(self.dpi_scale, self.dpi_scale)
        pix = page.get_pixmap(matrix=mat, alpha=False)
        png_bytes = pix.tobytes("png")
        return base64.b64encode(png_bytes).decode()

    def _extract_embedded_images(
        self, page: fitz.Page, page_number: int, min_size: int = 10000
    ) -> list[PageContent]:
        images = []
        image_list = page.get_images(full=True)

        for img_index, img_info in enumerate(image_list):
            xref = img_info[0]
            try:
                base_image = page.parent.extract_image(xref)
                if base_image and len(base_image["image"]) > min_size:
                    b64 = base64.b64encode(base_image["image"]).decode()
                    images.append(PageContent(
                        page_number=page_number,
                        content_type="image",
                        image_base64=b64,
                        image_size_bytes=len(base_image["image"])
                    ))
            except Exception as e:
                logger.warning(f"Error extracting image {img_index} from page {page_number}: {e}")

        return images

Chunking for RAG

The extracted text needs to be split into chunks for indexing in ChromaDB. Chunks that are too large dilute relevance; too small and they lose context.

Chunking strategy

Full text → Split by paragraphs → If paragraph > CHUNK_SIZE: subdivide
                                  → If paragraph < MIN_CHUNK: merge with next
                                  → Add overlap between chunks

Implementation

class TextChunker:
    def __init__(
        self,
        chunk_size: int = 1500,
        chunk_overlap: int = 200,
        min_chunk_size: int = 100
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.min_chunk_size = min_chunk_size

    def chunk_text(self, text: str, doc_id: str) -> list[DocumentChunk]:
        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
        chunks: list[DocumentChunk] = []
        current_chunk = ""
        chunk_index = 0

        for paragraph in paragraphs:
            if len(current_chunk) + len(paragraph) + 2 <= self.chunk_size:
                current_chunk += ("\n\n" + paragraph if current_chunk else paragraph)
            else:
                if len(current_chunk) >= self.min_chunk_size:
                    chunks.append(self._make_chunk(current_chunk, doc_id, chunk_index))
                    chunk_index += 1

                if len(paragraph) > self.chunk_size:
                    sub_chunks = self._split_long_paragraph(paragraph, doc_id, chunk_index)
                    chunks.extend(sub_chunks)
                    chunk_index += len(sub_chunks)
                    current_chunk = ""
                else:
                    overlap_text = current_chunk[-self.chunk_overlap:] if current_chunk else ""
                    current_chunk = overlap_text + "\n\n" + paragraph if overlap_text else paragraph

        if len(current_chunk) >= self.min_chunk_size:
            chunks.append(self._make_chunk(current_chunk, doc_id, chunk_index))

        return chunks

    def _split_long_paragraph(self, text: str, doc_id: str, start_index: int) -> list[DocumentChunk]:
        chunks = []
        for i in range(0, len(text), self.chunk_size - self.chunk_overlap):
            chunk_text = text[i:i + self.chunk_size]
            if len(chunk_text) >= self.min_chunk_size:
                chunks.append(self._make_chunk(chunk_text, doc_id, start_index + len(chunks)))
        return chunks

    def _make_chunk(self, text: str, doc_id: str, index: int) -> DocumentChunk:
        return DocumentChunk(
            chunk_id=f"{doc_id}_chunk_{index}",
            doc_id=doc_id,
            text=text,
            page_number=0,
            content_type="text",
            metadata={"char_count": len(text), "chunk_index": index}
        )


class DocumentChunk(BaseModel):
    chunk_id: str
    doc_id: str
    text: str
    page_number: int
    content_type: str
    metadata: dict = {}

Processing Mixed Documents

The problem

A financial PDF can have:

• Pages 1-3: digitally generated text (account data, transactions)
• Pages 4-5: scanned images of receipts

The DocumentProcessor already handles this by design: it iterates page by page and classifies each one. But for downstream, we need to be able to filter easily:

class ProcessedDocument(BaseModel):
    # ... previous fields ...

    def get_text_pages(self) -> list[PageContent]:
        return [p for p in self.pages if p.content_type == "text"]

    def get_image_pages(self) -> list[PageContent]:
        return [p for p in self.pages if p.content_type == "image"]

    def get_text_for_rag(self) -> str:
        return "\n\n".join(p.text for p in self.get_text_pages() if p.text)

    def get_images_for_vision(self) -> list[dict]:
        return [
            {"page": p.page_number, "base64": p.image_base64}
            for p in self.get_image_pages()
            if p.image_base64
        ]

Image Optimization

Reduce size before sending to the Vision API

Images rendered at 150 DPI can be larger than necessary. Reducing resolution saves costs (OpenAI charges per image token, which depend on size):

from PIL import Image
import io


def optimize_image_for_vision(
    image_base64: str,
    max_dimension: int = 2048,
    quality: int = 85
) -> str:
    raw = base64.b64decode(image_base64)
    img = Image.open(io.BytesIO(raw))

    if max(img.size) > max_dimension:
        ratio = max_dimension / max(img.size)
        new_size = (int(img.width * ratio), int(img.height * ratio))
        img = img.resize(new_size, Image.LANCZOS)

    buffer = io.BytesIO()
    img.save(buffer, format="PNG", optimize=True, quality=quality)
    return base64.b64encode(buffer.getvalue()).decode()

Cost table by resolution

For documents, 1024×1024 or 2048×2048 is usually enough. You only need higher resolution for documents with very small text.

Error Handling

Common errors and solutions

Error-handling implementation

def process_safe(self, file_path: str) -> ProcessedDocument | dict:
    try:
        return self.process(file_path)
    except fitz.FileDataError:
        logger.error(f"Corrupt PDF: {file_path}")
        return {"error": "Corrupt or unreadable PDF", "file": file_path}
    except MemoryError:
        logger.error(f"Insufficient memory processing: {file_path}")
        return {"error": "Document too large to process", "file": file_path}
    except Exception as e:
        logger.error(f"Unexpected error: {e}", exc_info=True)
        return {"error": f"Processing error: {str(e)}", "file": file_path}

Using the DocumentProcessor

Complete example

processor = DocumentProcessor(text_threshold=50, dpi_scale=150/72)
chunker = TextChunker(chunk_size=1500, chunk_overlap=200)

result = processor.process("invoice_march.pdf")

print(f"Type: {result.file_type}")
print(f"Total pages: {result.total_pages}")
print(f"Text pages: {result.text_page_count}")
print(f"Image pages: {result.image_page_count}")

if result.full_text:
    chunks = chunker.chunk_text(result.full_text, doc_id="invoice_001")
    print(f"Chunks generated: {len(chunks)}")
    for chunk in chunks[:3]:
        print(f"  - {chunk.chunk_id}: {len(chunk.text)} chars")

images = result.get_images_for_vision()
if images:
    print(f"Images for Vision: {len(images)}")
    for img in images:
        print(f"  - Page {img['page']}: {len(img['base64'])} chars base64")

Expected output

Type: pdf
Total pages: 3
Text pages: 2
Image pages: 1
Chunks generated: 4
  - invoice_001_chunk_0: 1423 chars
  - invoice_001_chunk_1: 1387 chars
  - invoice_001_chunk_2: 892 chars
  - invoice_001_chunk_3: 456 chars
Images for Vision: 1
  - Page 3: 48920 chars base64

Troubleshooting

"PyMuPDF doesn't extract text from my PDF"

Probable cause: The PDF is scanned (the "letters" are pixels, not text).

Check:

doc = fitz.open("document.pdf")
for i in range(len(doc)):
    text = doc[i].get_text()
    print(f"Page {i+1}: {len(text.strip())} characters")
doc.close()

If all pages have 0 or few characters, the PDF is scanned. The DocumentProcessor already handles this case: it converts the pages to images for Vision.

"The extracted images are too large"

Solution: Use the optimize_image_for_vision() function before sending to the API:

for page in result.get_image_pages():
    optimized = optimize_image_for_vision(page.image_base64, max_dimension=1024)
    page.image_base64 = optimized

"Memory error with large PDFs"

Solution: Process pages in batches:

def process_pdf_batched(self, file_path: str, batch_size: int = 10) -> ProcessedDocument:
    doc = fitz.open(file_path)
    all_pages = []

    for batch_start in range(0, len(doc), batch_size):
        batch_end = min(batch_start + batch_size, len(doc))
        for i in range(batch_start, batch_end):
            page = doc[i]
            text = page.get_text()
            if len(text.strip()) > self.text_threshold:
                all_pages.append(PageContent(
                    page_number=i + 1, content_type="text",
                    text=text, char_count=len(text)
                ))
            else:
                b64 = self._page_to_base64(page)
                all_pages.append(PageContent(
                    page_number=i + 1, content_type="image",
                    image_base64=b64
                ))

    doc.close()
    # ... build ProcessedDocument ...

Exercises

Exercise 1: Chunking with page metadata

Modify the TextChunker so that each chunk includes the page number it comes from. You need to pass the page information from ProcessedDocument to the chunker.

def chunk_pages(self, pages: list[PageContent], doc_id: str) -> list[DocumentChunk]:
    chunks: list[DocumentChunk] = []
    chunk_index = 0

    for page in pages:
        if page.content_type != "text" or not page.text:
            continue

        paragraphs = [p.strip() for p in page.text.split("\n\n") if p.strip()]
        current_chunk = ""

        for paragraph in paragraphs:
            if len(current_chunk) + len(paragraph) + 2 <= self.chunk_size:
                current_chunk += ("\n\n" + paragraph if current_chunk else paragraph)
            else:
                if len(current_chunk) >= self.min_chunk_size:
                    chunks.append(DocumentChunk(
                        chunk_id=f"{doc_id}_chunk_{chunk_index}",
                        doc_id=doc_id,
                        text=current_chunk,
                        page_number=page.page_number,
                        content_type="text",
                        metadata={"source_page": page.page_number}
                    ))
                    chunk_index += 1
                current_chunk = paragraph

        if len(current_chunk) >= self.min_chunk_size:
            chunks.append(DocumentChunk(
                chunk_id=f"{doc_id}_chunk_{chunk_index}",
                doc_id=doc_id,
                text=current_chunk,
                page_number=page.page_number,
                content_type="text",
                metadata={"source_page": page.page_number}
            ))
            chunk_index += 1

    return chunks

Exercise 2: Detecting when OCR is needed

Implement a function that analyzes a ProcessedDocument and returns a report: how many pages need OCR (Vision), how many have direct text, and a cost estimate based on the number of image-pages.

COST_PER_IMAGE_PAGE = 0.01  # estimated Vision cost per image
COST_PER_TEXT_PAGE = 0.0001  # text processing cost


def analyze_processing_needs(doc: ProcessedDocument) -> dict:
    text_pages = doc.get_text_pages()
    image_pages = doc.get_image_pages()

    text_cost = len(text_pages) * COST_PER_TEXT_PAGE
    image_cost = len(image_pages) * COST_PER_IMAGE_PAGE
    total_cost = text_cost + image_cost

    return {
        "total_pages": doc.total_pages,
        "text_pages": len(text_pages),
        "image_pages_needing_ocr": len(image_pages),
        "estimated_cost_usd": round(total_cost, 4),
        "cost_breakdown": {
            "text_processing": round(text_cost, 4),
            "vision_ocr": round(image_cost, 4)
        },
        "recommendation": (
            "Text only — economical processing"
            if len(image_pages) == 0
            else f"{len(image_pages)} pages require the Vision API — "
                 f"estimated cost ${image_cost:.3f}"
        )
    }


doc = processor.process("financial_report.pdf")
report = analyze_processing_needs(doc)
print(report)

Summary

• The DocumentProcessor is the foundation of the pipeline: it transforms raw files into a standardized ProcessedDocument.
• It detects content type per page: native text (direct extraction) vs scanned (requires Vision).
• The TextChunker splits text into chunks with overlap for optimal RAG indexing.
• Images are optimized before sending to the Vision API to reduce costs.
• Error handling covers corrupt PDFs, insufficient memory and unsupported formats.
• Batch processing lets you handle large PDFs without exhausting memory.

Additional Resources

1. PyMuPDF Documentation — Complete reference
2. PyMuPDF Recipes — Text extraction recipes
3. Pillow Documentation — Image processing
4. Module 3 of this guide — Foundation of document processing