pdf-document-processing

name	pdf-document-processing
description	Expert guidance for extracting text and data from PDF documents in Python. Use when implementing document ingestion pipelines, text extraction, chunking strategies, OCR for scanned documents, or metadata extraction. Covers pypdf, pdfplumber, and integration with LLM pipelines.

PDF Document Processing Skill

Overview

PDF processing is essential for document analysis pipelines. This skill covers text extraction, chunking, metadata handling, and integration with LLMs for entity extraction.

When to Use This Skill

• Extracting text from PDF contracts, reports, or legal documents
• Building document ingestion pipelines
• Creating chunking strategies for LLM processing
• Handling multi-page documents
• Extracting tables and structured data
• Processing scanned documents with OCR

Libraries

Library	Use Case	Notes
pypdf	Basic text extraction	Fast, pure Python
pdfplumber	Tables and structured data	Detailed layout info
pdf2image + pytesseract	Scanned documents (OCR)	Requires Tesseract
PyMuPDF (fitz)	Fast rendering and extraction	C bindings

Basic Text Extraction

Using pypdf (Recommended for Simple PDFs)

from pypdf import PdfReader
from pathlib import Path

def extract_text_pypdf(pdf_path: str) -> str:
    """Extract all text from a PDF file."""
    reader = PdfReader(pdf_path)
    
    text_parts = []
    for page in reader.pages:
        text = page.extract_text()
        if text:
            text_parts.append(text)
    
    return "\n\n".join(text_parts)

def extract_pages_pypdf(pdf_path: str) -> list[dict]:
    """Extract text with page information."""
    reader = PdfReader(pdf_path)
    
    pages = []
    for i, page in enumerate(reader.pages, start=1):
        text = page.extract_text() or ""
        pages.append({
            "page_number": i,
            "content": text,
            "char_count": len(text),
        })
    
    return pages

# Usage
text = extract_text_pypdf("/docs/contract.pdf")
pages = extract_pages_pypdf("/docs/contract.pdf")

Using pdfplumber (Better for Complex Layouts)

import pdfplumber

def extract_text_pdfplumber(pdf_path: str) -> str:
    """Extract text with better layout handling."""
    text_parts = []
    
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            text = page.extract_text()
            if text:
                text_parts.append(text)
    
    return "\n\n".join(text_parts)

def extract_with_metadata(pdf_path: str) -> dict:
    """Extract text with document metadata."""
    with pdfplumber.open(pdf_path) as pdf:
        return {
            "metadata": pdf.metadata,
            "page_count": len(pdf.pages),
            "pages": [
                {
                    "page_number": i + 1,
                    "content": page.extract_text() or "",
                    "width": page.width,
                    "height": page.height,
                }
                for i, page in enumerate(pdf.pages)
            ]
        }

Chunking Strategies

Fixed-Size Chunking

def chunk_text_fixed(
    text: str,
    chunk_size: int = 1000,
    overlap: int = 100,
) -> list[str]:
    """Split text into fixed-size chunks with overlap."""
    chunks = []
    start = 0
    
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap
    
    return chunks

# Usage
text = extract_text_pypdf("contract.pdf")
chunks = chunk_text_fixed(text, chunk_size=1000, overlap=100)

Sentence-Based Chunking

import re

def chunk_by_sentences(
    text: str,
    max_chunk_size: int = 1000,
    overlap_sentences: int = 1,
) -> list[str]:
    """Chunk text by sentences, respecting max size."""
    # Split into sentences
    sentences = re.split(r'(?<=[.!?])\s+', text)
    
    chunks = []
    current_chunk = []
    current_size = 0
    
    for sentence in sentences:
        sentence_size = len(sentence)
        
        if current_size + sentence_size > max_chunk_size and current_chunk:
            # Save current chunk
            chunks.append(" ".join(current_chunk))
            # Keep overlap sentences
            current_chunk = current_chunk[-overlap_sentences:] if overlap_sentences else []
            current_size = sum(len(s) for s in current_chunk)
        
        current_chunk.append(sentence)
        current_size += sentence_size
    
    # Add final chunk
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

Page-Aware Chunking

from dataclasses import dataclass

@dataclass
class DocumentChunk:
    content: str
    start_page: int
    end_page: int
    chunk_index: int

def chunk_by_pages(
    pdf_path: str,
    pages_per_chunk: int = 2,
) -> list[DocumentChunk]:
    """Create chunks based on page boundaries."""
    from pypdf import PdfReader
    
    reader = PdfReader(pdf_path)
    chunks = []
    chunk_index = 0
    
    for i in range(0, len(reader.pages), pages_per_chunk):
        end_idx = min(i + pages_per_chunk, len(reader.pages))
        
        # Extract text from pages in this chunk
        text_parts = []
        for page in reader.pages[i:end_idx]:
            text = page.extract_text()
            if text:
                text_parts.append(text)
        
        if text_parts:
            chunks.append(DocumentChunk(
                content="\n\n".join(text_parts),
                start_page=i + 1,
                end_page=end_idx,
                chunk_index=chunk_index,
            ))
            chunk_index += 1
    
    return chunks

Table Extraction

import pdfplumber
import pandas as pd

def extract_tables(pdf_path: str) -> list[pd.DataFrame]:
    """Extract all tables from a PDF."""
    tables = []
    
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            page_tables = page.extract_tables()
            for table in page_tables:
                if table:
                    # Convert to DataFrame
                    df = pd.DataFrame(table[1:], columns=table[0])
                    tables.append(df)
    
    return tables

def extract_tables_as_text(pdf_path: str) -> list[str]:
    """Extract tables and convert to text representation."""
    tables = extract_tables(pdf_path)
    
    text_tables = []
    for df in tables:
        text_tables.append(df.to_string(index=False))
    
    return text_tables

Parallel Processing with Ray

import ray
from pypdf import PdfReader

@ray.remote
def process_pdf_chunk(pdf_path: str, start_page: int, end_page: int) -> dict:
    """Process a range of pages from a PDF."""
    reader = PdfReader(pdf_path)
    
    text_parts = []
    for page in reader.pages[start_page:end_page]:
        text = page.extract_text()
        if text:
            text_parts.append(text)
    
    return {
        "start_page": start_page + 1,
        "end_page": end_page,
        "content": "\n\n".join(text_parts),
    }

def parallel_extract(pdf_path: str, num_workers: int = 4) -> list[dict]:
    """Extract PDF text in parallel using Ray."""
    reader = PdfReader(pdf_path)
    total_pages = len(reader.pages)
    pages_per_worker = max(1, total_pages // num_workers)
    
    # Launch parallel tasks
    futures = []
    for i in range(0, total_pages, pages_per_worker):
        end = min(i + pages_per_worker, total_pages)
        futures.append(process_pdf_chunk.remote(pdf_path, i, end))
    
    # Collect results
    results = ray.get(futures)
    return sorted(results, key=lambda x: x["start_page"])

Integration with Temporal Activities

from temporalio import activity
from pypdf import PdfReader
from pathlib import Path

@activity.defn(name="load_and_chunk_document")
async def load_and_chunk_document(
    pdf_path: str,
    chunk_size: int = 1000,
) -> list[dict]:
    """Activity to load PDF and create chunks."""
    
    activity.heartbeat("Loading PDF...")
    
    # Validate file exists
    if not Path(pdf_path).exists():
        raise ValueError(f"PDF not found: {pdf_path}")
    
    # Extract text
    reader = PdfReader(pdf_path)
    activity.heartbeat(f"Processing {len(reader.pages)} pages...")
    
    chunks = []
    for i, page in enumerate(reader.pages):
        text = page.extract_text()
        if text:
            chunks.append({
                "content": text,
                "page": i + 1,
                "source": pdf_path,
            })
        
        if i % 10 == 0:
            activity.heartbeat(f"Processed {i+1}/{len(reader.pages)} pages")
    
    activity.heartbeat(f"Created {len(chunks)} chunks")
    return chunks

OCR for Scanned Documents

from pdf2image import convert_from_path
import pytesseract
from pathlib import Path

def extract_text_ocr(pdf_path: str, dpi: int = 300) -> str:
    """Extract text from scanned PDF using OCR."""
    
    # Convert PDF pages to images
    images = convert_from_path(pdf_path, dpi=dpi)
    
    text_parts = []
    for i, image in enumerate(images):
        # Run OCR on each page
        text = pytesseract.image_to_string(image)
        if text.strip():
            text_parts.append(f"--- Page {i+1} ---\n{text}")
    
    return "\n\n".join(text_parts)

def needs_ocr(pdf_path: str) -> bool:
    """Check if PDF needs OCR (has minimal extractable text)."""
    reader = PdfReader(pdf_path)
    
    total_chars = 0
    for page in reader.pages[:3]:  # Check first 3 pages
        text = page.extract_text() or ""
        total_chars += len(text.strip())
    
    # If very little text, likely scanned
    return total_chars < 100

Document Validation

from pypdf import PdfReader
from pathlib import Path

def validate_pdf(pdf_path: str) -> dict:
    """Validate PDF and return metadata."""
    path = Path(pdf_path)
    
    if not path.exists():
        return {"valid": False, "error": "File not found"}
    
    if not path.suffix.lower() == ".pdf":
        return {"valid": False, "error": "Not a PDF file"}
    
    try:
        reader = PdfReader(pdf_path)
        
        return {
            "valid": True,
            "page_count": len(reader.pages),
            "encrypted": reader.is_encrypted,
            "metadata": {
                "title": reader.metadata.title if reader.metadata else None,
                "author": reader.metadata.author if reader.metadata else None,
                "creator": reader.metadata.creator if reader.metadata else None,
            },
            "file_size_bytes": path.stat().st_size,
        }
    except Exception as e:
        return {"valid": False, "error": str(e)}

Complete Processing Pipeline

from dataclasses import dataclass
from pypdf import PdfReader
from pathlib import Path
import hashlib

@dataclass
class ProcessedDocument:
    path: str
    file_hash: str
    page_count: int
    chunks: list[dict]
    metadata: dict

def process_document(
    pdf_path: str,
    chunk_size: int = 1000,
    overlap: int = 100,
) -> ProcessedDocument:
    """Complete document processing pipeline."""
    
    path = Path(pdf_path)
    
    # Compute file hash for deduplication
    with open(path, "rb") as f:
        file_hash = hashlib.md5(f.read()).hexdigest()
    
    # Extract text
    reader = PdfReader(pdf_path)
    
    # Process pages
    all_text = []
    page_boundaries = [0]
    
    for page in reader.pages:
        text = page.extract_text() or ""
        all_text.append(text)
        page_boundaries.append(page_boundaries[-1] + len(text))
    
    full_text = "\n\n".join(all_text)
    
    # Create chunks with page references
    chunks = []
    start = 0
    chunk_idx = 0
    
    while start < len(full_text):
        end = min(start + chunk_size, len(full_text))
        chunk_text = full_text[start:end]
        
        # Find which page(s) this chunk spans
        start_page = next(
            i for i, boundary in enumerate(page_boundaries)
            if boundary > start
        )
        end_page = next(
            (i for i, boundary in enumerate(page_boundaries) if boundary >= end),
            len(reader.pages)
        )
        
        chunks.append({
            "index": chunk_idx,
            "content": chunk_text,
            "start_page": start_page,
            "end_page": end_page,
            "char_start": start,
            "char_end": end,
        })
        
        start = end - overlap
        chunk_idx += 1
    
    return ProcessedDocument(
        path=str(path),
        file_hash=file_hash,
        page_count=len(reader.pages),
        chunks=chunks,
        metadata={
            "title": reader.metadata.title if reader.metadata else None,
            "author": reader.metadata.author if reader.metadata else None,
        }
    )

Installation

# Basic PDF processing
pip install pypdf>=5.1.0

# Advanced layout and tables
pip install pdfplumber

# OCR support
pip install pdf2image pytesseract
# Also requires: apt install tesseract-ocr (Linux) or brew install tesseract (Mac)

# Fast rendering
pip install PyMuPDF

Performance Tips

1. Use pypdf for simple extraction - Fastest for text-only
2. Batch process pages - Don't open/close PDF repeatedly
3. Parallelize with Ray - For large document batches
4. Skip OCR when possible - Check if text is extractable first
5. Cache extracted text - Store with file hash for deduplication
6. Stream large files - Process page by page for memory efficiency

References

• pypdf Documentation
• pdfplumber Documentation
• PyMuPDF Documentation
• Tesseract OCR