Overview
Document processing pipeline: extract → clean → chunk → embed → query. Use the right tool for the document type: structured PDFs get pypdf/pdfplumber, scanned PDFs need OCR, complex layouts need specialized extractors.
Tool Selection
| Document Type | Tool | When | |--------------|------|------| | Digital PDF (text layer) | pypdf, pdfplumber | Default for PDFs with selectable text | | Scanned PDF / image | pytesseract + pdf2image | When text extraction returns garbage | | Complex tables | pdfplumber | Preserves cell boundaries | | Word/Office docs | python-docx, mammoth | DOCX, XLSX files | | Markdown conversion | markdownify, pandoc | HTML or complex formats to MD | | AI analysis | Claude Files API | When you need reasoning, not just extraction |
pypdf: Basic Extraction
from pypdf import PdfReader
def extract_pdf_text(path: str) -> str:
reader = PdfReader(path)
pages = []
for i, page in enumerate(reader.pages):
text = page.extract_text()
if text:
pages.append(f"--- Page {i+1} ---\n{text}")
return "\n\n".join(pages)
text = extract_pdf_text("document.pdf")
print(f"Extracted {len(text)} chars from {len(reader.pages)} pages")
pdfplumber: Tables and Layout
import pdfplumber
def extract_with_tables(path: str) -> dict:
result = {"text": [], "tables": []}
with pdfplumber.open(path) as pdf:
for i, page in enumerate(pdf.pages):
# Extract text
text = page.extract_text()
if text:
result["text"].append(f"Page {i+1}:\n{text}")
# Extract tables
tables = page.extract_tables()
for table in tables:
result["tables"].append({
"page": i + 1,
"rows": table,
})
return result
OCR for Scanned PDFs
import pdf2image
import pytesseract
from PIL import Image
def ocr_pdf(path: str, dpi: int = 300) -> str:
images = pdf2image.convert_from_path(path, dpi=dpi)
pages = []
for i, image in enumerate(images):
text = pytesseract.image_to_string(image, lang="eng")
pages.append(f"--- Page {i+1} ---\n{text}")
return "\n\n".join(pages)
# Install deps
# sudo apt install tesseract-ocr
# pip install pytesseract pdf2image
Detect If PDF Needs OCR
def needs_ocr(path: str, min_chars_per_page: int = 50) -> bool:
reader = PdfReader(path)
total_chars = sum(
len(page.extract_text() or "")
for page in reader.pages
)
avg_chars = total_chars / len(reader.pages) if reader.pages else 0
return avg_chars < min_chars_per_page
Chunking for RAG
from typing import Generator
def chunk_text(
text: str,
chunk_size: int = 500,
overlap: int = 50
) -> Generator[dict, None, None]:
words = text.split()
start = 0
while start < len(words):
end = min(start + chunk_size, len(words))
chunk = " ".join(words[start:end])
yield {
"content": chunk,
"word_start": start,
"word_end": end,
}
start = end - overlap # overlap for context continuity
# Usage
chunks = list(chunk_text(extracted_text, chunk_size=500, overlap=50))
print(f"Created {len(chunks)} chunks")
Markdown Conversion
# HTML to markdown
from markdownify import markdownify as md
html = "<h1>Title</h1><p>Paragraph with <strong>bold</strong></p>"
markdown = md(html, heading_style="ATX")
# Office docs via pandoc (shell)
import subprocess
def docx_to_md(input_path: str, output_path: str):
subprocess.run(
["pandoc", input_path, "-o", output_path, "--wrap=none"],
check=True
)
Claude API: Document Analysis via Files API
For complex reasoning about documents:
import anthropic
import base64
client = anthropic.Anthropic()
# Method 1: Upload to Files API (reuse across requests)
with open("report.pdf", "rb") as f:
file = client.files.create(file=f)
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=4_000,
messages=[{
"role": "user",
"content": [
{
"type": "document",
"source": { "type": "file", "file_id": file.id },
},
{ "type": "text", "text": "Extract all financial figures and organize them as JSON." },
],
}],
)
# Method 2: Inline base64 (single use)
with open("report.pdf", "rb") as f:
pdf_data = base64.standard_b64encode(f.read()).decode("utf-8")
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=4_000,
messages=[{
"role": "user",
"content": [
{
"type": "document",
"source": { "type": "base64", "media_type": "application/pdf", "data": pdf_data },
},
{ "type": "text", "text": "Summarize the key findings." },
],
}],
)
Batch PDF Processing
import asyncio
import httpx
from pathlib import Path
async def process_directory(pdf_dir: str, output_dir: str):
pdfs = list(Path(pdf_dir).glob("*.pdf"))
Path(output_dir).mkdir(parents=True, exist_ok=True)
for pdf in pdfs:
try:
# Check if needs OCR
if needs_ocr(str(pdf)):
text = ocr_pdf(str(pdf))
else:
text = extract_pdf_text(str(pdf))
# Save extracted text
output_path = Path(output_dir) / f"{pdf.stem}.txt"
output_path.write_text(text, encoding="utf-8")
print(f"Processed: {pdf.name} -> {output_path.name}")
except Exception as e:
print(f"Failed: {pdf.name}: {e}")
asyncio.run(process_directory("./pdfs", "./extracted"))
Document Metadata Extraction
from pypdf import PdfReader
def get_metadata(path: str) -> dict:
reader = PdfReader(path)
info = reader.metadata or {}
return {
"pages": len(reader.pages),
"title": info.get("/Title", ""),
"author": info.get("/Author", ""),
"subject": info.get("/Subject", ""),
"creator": info.get("/Creator", ""),
"created": str(info.get("/CreationDate", "")),
"encrypted": reader.is_encrypted,
}