8. Project: Document Extractor

Description

This project wraps up Module 3 by building a complete Document Extractor: a system that receives a PDF or document image, detects its type, extracts text with the appropriate technique (PyMuPDF, Tesseract, Vision), turns the information into structured data with Pydantic, and handles long documents with chunking. It integrates every capsule of the module into a working pipeline.

It isn't an isolated script. It's the pattern that production document-processing systems use: a router that decides the extraction route based on the document type, and a parser that turns free text into typed, validated data.

Why it matters: In production, documents arrive in any format — PDFs with text, scanned PDFs, photos of invoices. Without a system that detects the type and applies the correct technique, you end up with fragile pipelines that fail on the first unexpected document.

Connection with the module: Each component comes from a capsule: PDFs (02), OCR/Vision (03), document images (04), Pydantic (05), chunking (06), errors (07).

Connection with the guide: This extractor is the foundation of the Document Analyzer in Module 8. There you'll add RAG for Q&A over the extracted content, and optionally TTS for spoken summaries.

Technical Specifications

Input

The extractor accepts two arguments:

Output

@dataclass
class ExtractionResult:
    success: bool                    # Whether the extraction succeeded
    data: dict | None                # Structured data validated by Pydantic
    raw_text: str                    # Raw text extracted from the document
    pages_processed: int             # Number of pages processed
    method: str                      # "pymupdf" | "tesseract" | "vision"
    document_type: str               # "pdf_text" | "pdf_scanned" | "image"
    schema_used: str                 # "invoice" | "receipt" | "contract"
    cost_usd: float                  # Estimated extraction cost
    confidence: float                # 0.0-1.0 confidence in the extraction
    errors: list[str]                # Errors found during the process
    latency_seconds: float           # Total execution time

Functional requirements

1. Detect the type of document: PDF with text, scanned PDF, image
2. Extract text with the optimal technique for the detected type
3. Define schemas in Pydantic for invoices, receipts and contracts
4. Extract structured data by sending text/image to the LLM with the schema
5. Handle long documents with page-based chunking and result merging
6. Report cost, confidence and method used in each extraction

Step 1: Configuration and Data Structures

We define the enums, dataclasses and base configuration. Separating constants from logic lets you change prices or methods without touching the pipeline.

from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Optional
import base64
import json
import time

import fitz
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError


class DocumentType(Enum):
    PDF_TEXT = "pdf_text"
    PDF_SCANNED = "pdf_scanned"
    IMAGE = "image"
    UNKNOWN = "unknown"


class ExtractionMethod(Enum):
    PYMUPDF = "pymupdf"
    TESSERACT = "tesseract"
    VISION = "vision"


SUPPORTED_EXTENSIONS = {
    ".pdf": "pdf",
    ".jpg": "image", ".jpeg": "image",
    ".png": "image", ".webp": "image",
}

METHOD_COSTS_PER_PAGE = {
    ExtractionMethod.PYMUPDF: 0.0,
    ExtractionMethod.TESSERACT: 0.0,
    ExtractionMethod.VISION: 0.003,
}

MIN_TEXT_CHARS_FOR_DIGITAL = 50


@dataclass
class ExtractionResult:
    success: bool = False
    data: dict | None = None
    raw_text: str = ""
    pages_processed: int = 0
    method: str = ""
    document_type: str = ""
    schema_used: str = ""
    cost_usd: float = 0.0
    confidence: float = 0.0
    errors: list[str] = field(default_factory=list)
    latency_seconds: float = 0.0


client = OpenAI()

DocumentType encodes the three cases the system must handle. ExtractionMethod determines which technique to use. The per-page costs are approximate — the Vision API charges by image tokens, but for a quick estimate we use a per-page average.

Step 2: Detect the Document Type

Detection follows three rules: if it's an image, it's an image. If it's a PDF, we extract text with PyMuPDF — if it has more than 50 characters per page on average, it's a digital PDF; if not, it's a scanned PDF.

def detect_document_type(file_path: str) -> DocumentType:
    """Classifies the document as a digital PDF, scanned PDF, or image."""
    path = Path(file_path)

    if not path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")

    suffix = path.suffix.lower()
    if suffix not in SUPPORTED_EXTENSIONS:
        raise ValueError(f"Unsupported format: {suffix}")

    if SUPPORTED_EXTENSIONS[suffix] == "image":
        return DocumentType.IMAGE

    doc = fitz.open(file_path)
    total_text = ""
    for page_num in range(len(doc)):
        total_text += doc[page_num].get_text()
    doc.close()

    avg_chars_per_page = len(total_text.strip()) / max(len(doc), 1)
    if avg_chars_per_page > MIN_TEXT_CHARS_FOR_DIGITAL:
        return DocumentType.PDF_TEXT

    return DocumentType.PDF_SCANNED

Step 3: Extract Text

Three extraction methods. PyMuPDF for digital PDFs (free, fast). Tesseract for local OCR when Vision isn't an option. The Vision API for maximum quality on scanned documents or images.

def extract_text_pymupdf(file_path: str) -> tuple[str, int]:
    """Extracts text from a digital PDF with PyMuPDF. Returns (text, pages)."""
    doc = fitz.open(file_path)
    pages = []
    for page_num in range(len(doc)):
        page_text = doc[page_num].get_text()
        if page_text.strip():
            pages.append(page_text)
    doc.close()
    return "\n\n".join(pages), len(pages)


def extract_text_tesseract(file_path: str) -> tuple[str, int]:
    """Extracts text with Tesseract OCR. Converts the PDF to images first."""
    import pytesseract
    from pdf2image import convert_from_path
    from PIL import Image

    path = Path(file_path)
    if path.suffix.lower() == ".pdf":
        images = convert_from_path(file_path, dpi=200)
    else:
        images = [Image.open(file_path)]

    texts = []
    for img in images:
        text = pytesseract.image_to_string(img, lang="eng")
        texts.append(text)

    return "\n\n".join(texts), len(images)


def extract_text_vision(file_path: str) -> tuple[str, int]:
    """Extracts text by sending images to GPT-4 Vision."""
    path = Path(file_path)
    images_b64 = []

    if path.suffix.lower() == ".pdf":
        doc = fitz.open(file_path)
        for page_num in range(min(len(doc), 10)):
            page = doc[page_num]
            mat = fitz.Matrix(150 / 72, 150 / 72)
            pix = page.get_pixmap(matrix=mat, alpha=False)
            b64 = base64.b64encode(pix.tobytes("png")).decode()
            images_b64.append(b64)
        doc.close()
    else:
        with open(file_path, "rb") as f:
            b64 = base64.b64encode(f.read()).decode()
        images_b64.append(b64)

    prompt = "Extract ALL the visible text in this document. Keep the original structure (headings, lists, tables). Respond only with the extracted text."

    content = [{"type": "text", "text": prompt}]
    for b64 in images_b64:
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/png;base64,{b64}"}
        })

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        temperature=0,
        max_tokens=4096,
    )

    return response.choices[0].message.content, len(images_b64)


def select_and_extract(
    file_path: str, doc_type: DocumentType
) -> tuple[str, int, ExtractionMethod]:
    """Selects the extraction method based on the document type."""
    if doc_type == DocumentType.PDF_TEXT:
        text, pages = extract_text_pymupdf(file_path)
        return text, pages, ExtractionMethod.PYMUPDF

    if doc_type == DocumentType.PDF_SCANNED:
        text, pages = extract_text_vision(file_path)
        return text, pages, ExtractionMethod.VISION

    if doc_type == DocumentType.IMAGE:
        text, pages = extract_text_vision(file_path)
        return text, pages, ExtractionMethod.VISION

    raise ValueError(f"Unrecognized document type: {doc_type}")

select_and_extract is the central router. Digital PDFs go to PyMuPDF (zero cost). Scanned ones and images go to Vision. If you wanted a fallback to Tesseract for cost, you'd change one line here.

Step 4: Pydantic Schemas

Three schemas for the most common document types. The schema registry maps names to Pydantic classes and enables auto-detection by keywords.

class InvoiceItem(BaseModel):
    description: str = Field(default="", description="Product/service description")
    quantity: float = Field(default=1.0, description="Quantity")
    unit_price: float = Field(default=0.0, description="Price per unit")
    amount: float = Field(default=0.0, description="Total amount of the item")


class InvoiceSchema(BaseModel):
    date: Optional[str] = Field(default=None, description="Date in YYYY-MM-DD format")
    invoice_number: Optional[str] = Field(default=None, description="Invoice number")
    vendor: Optional[str] = Field(default=None, description="Vendor/issuer name")
    customer: Optional[str] = Field(default=None, description="Customer/recipient name")
    subtotal: Optional[float] = Field(default=None, description="Subtotal before taxes")
    tax: Optional[float] = Field(default=None, description="Tax amount")
    total: Optional[float] = Field(default=None, description="Total to pay")
    currency: Optional[str] = Field(default=None, description="Currency (USD, EUR, MXN)")
    items: list[InvoiceItem] = Field(default_factory=list, description="List of items")


class ReceiptItem(BaseModel):
    description: str = Field(default="", description="Product name")
    quantity: float = Field(default=1.0)
    price: float = Field(default=0.0)


class ReceiptSchema(BaseModel):
    date: Optional[str] = Field(default=None, description="Date YYYY-MM-DD")
    merchant: Optional[str] = Field(default=None, description="Merchant name")
    address: Optional[str] = Field(default=None, description="Merchant address")
    items: list[ReceiptItem] = Field(default_factory=list)
    subtotal: Optional[float] = None
    tax: Optional[float] = None
    total: Optional[float] = None
    payment_method: Optional[str] = Field(default=None, description="Cash, card, etc.")


class ContractSchema(BaseModel):
    title: Optional[str] = Field(default=None, description="Contract title")
    date: Optional[str] = Field(default=None, description="Signing date YYYY-MM-DD")
    parties: list[str] = Field(default_factory=list, description="Parties involved")
    subject: Optional[str] = Field(default=None, description="Subject of the contract")
    term: Optional[str] = Field(default=None, description="Term period")
    amount: Optional[float] = Field(default=None, description="Contract amount")
    key_clauses: list[str] = Field(
        default_factory=list, description="Main clauses summarized"
    )


SCHEMA_REGISTRY: dict[str, type[BaseModel]] = {
    "invoice": InvoiceSchema,
    "receipt": ReceiptSchema,
    "contract": ContractSchema,
}

SCHEMA_KEYWORDS: dict[str, list[str]] = {
    "invoice": ["invoice", "bill", "invoice no", "subtotal", "tax", "vendor"],
    "receipt": ["ticket", "receipt", "merchant", "cashier", "change"],
    "contract": ["contract", "agreement", "clause", "term", "parties", "signatory"],
}


def detect_schema_type(text: str) -> str:
    """Auto-detects the schema type by analyzing keywords in the text."""
    text_lower = text.lower()
    scores = {}
    for schema_name, keywords in SCHEMA_KEYWORDS.items():
        score = sum(1 for kw in keywords if kw in text_lower)
        scores[schema_name] = score

    best = max(scores, key=scores.get)
    if scores[best] == 0:
        return "invoice"
    return best

Each schema uses Field(default=None) — a document may not have all the fields, and we extract what's there without failing. Keyword auto-detection is simple but effective; in production you could replace it with an LLM classifier.

Step 5: Structured Extraction with the LLM

We send the extracted text to the LLM along with the JSON schema. The LLM returns JSON that we parse and validate with Pydantic.

def extract_structured_data(
    text: str,
    schema_type: str,
    images_b64: list[str] | None = None,
) -> tuple[dict, float]:
    """Extracts structured data using LLM + Pydantic schema.

    Returns:
        (validated_data, confidence)
    """
    schema_class = SCHEMA_REGISTRY.get(schema_type)
    if not schema_class:
        raise ValueError(f"Schema not registered: {schema_type}")

    schema_json = schema_class.model_json_schema()
    schema_str = json.dumps(schema_json, indent=2, ensure_ascii=False)

    prompt = f"""Extract the data from this document according to the provided schema.

SCHEMA (JSON Schema):
{schema_str}

RULES:
- Respond ONLY with valid JSON that complies with the schema
- If a field is not present in the document, use null
- Dates in YYYY-MM-DD format
- Amounts as numbers (no currency symbols)
- If there are items/lines, extract all you find

DOCUMENT:
{text[:6000]}"""

    content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]

    if images_b64:
        for b64 in images_b64[:5]:
            content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{b64}"}
            })

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        response_format={"type": "json_object"},
        temperature=0,
    )

    raw_json = json.loads(response.choices[0].message.content)

    try:
        validated = schema_class(**raw_json)
        data = validated.model_dump()
        confidence = _calculate_field_confidence(data)
        return data, confidence
    except ValidationError as e:
        data = _safe_partial_parse(raw_json, schema_class)
        confidence = _calculate_field_confidence(data) * 0.7
        return data, confidence


def _safe_partial_parse(raw: dict, schema_class: type[BaseModel]) -> dict:
    """Attempts a partial parse when full validation fails."""
    clean = {}
    for field_name, field_info in schema_class.model_fields.items():
        if field_name in raw:
            try:
                clean[field_name] = raw[field_name]
            except (TypeError, ValueError):
                clean[field_name] = None
        else:
            clean[field_name] = None
    return clean


def _calculate_field_confidence(data: dict) -> float:
    """Calculates confidence based on the ratio of non-null fields."""
    if not data:
        return 0.0
    total = len(data)
    filled = sum(1 for v in data.values() if v is not None and v != "" and v != [])
    return round(filled / max(total, 1), 2)

Step 6: Handling Long Documents

For PDFs of more than 5 pages, we process by chunks and merge the results. _safe_partial_parse acts as a fallback: if the LLM returns JSON that doesn't comply with the schema, we extract what's possible instead of losing the extraction.

CHUNK_SIZE_PAGES = 5


def extract_long_document(
    file_path: str,
    schema_type: str,
) -> tuple[str, list[dict], int]:
    """Extracts text from a long document in page-based chunks.

    Returns:
        (full_text, chunks_data, total_pages)
    """
    doc = fitz.open(file_path)
    total_pages = len(doc)
    all_text_parts = []
    chunks_data = []

    for start in range(0, total_pages, CHUNK_SIZE_PAGES):
        end = min(start + CHUNK_SIZE_PAGES, total_pages)
        chunk_text = ""
        for page_num in range(start, end):
            chunk_text += doc[page_num].get_text() + "\n\n"

        all_text_parts.append(chunk_text)

        if chunk_text.strip():
            data, conf = extract_structured_data(chunk_text, schema_type)
            chunks_data.append({"pages": f"{start+1}-{end}", "data": data, "confidence": conf})

    doc.close()
    full_text = "\n\n".join(all_text_parts)
    return full_text, chunks_data, total_pages


def merge_chunk_results(chunks_data: list[dict], schema_type: str) -> tuple[dict, float]:
    """Combines the results of multiple chunks into a single result.

    Strategy: for scalar fields it takes the first non-null value.
    For lists (items, clauses) it concatenates all of them.
    """
    if not chunks_data:
        return {}, 0.0

    if len(chunks_data) == 1:
        return chunks_data[0]["data"], chunks_data[0]["confidence"]

    schema_class = SCHEMA_REGISTRY[schema_type]
    merged = {}
    list_fields = set()

    for field_name, field_info in schema_class.model_fields.items():
        if hasattr(field_info.annotation, "__origin__") and field_info.annotation.__origin__ is list:
            list_fields.add(field_name)
            merged[field_name] = []
        else:
            merged[field_name] = None

    for chunk in chunks_data:
        data = chunk["data"]
        for key, value in data.items():
            if key in list_fields and isinstance(value, list):
                merged[key].extend(value)
            elif merged.get(key) is None and value is not None:
                merged[key] = value

    avg_confidence = sum(c["confidence"] for c in chunks_data) / len(chunks_data)
    return merged, round(avg_confidence, 2)

The merge prioritizes the first non-null value for scalar fields (date, total). For lists like items or clauses, it concatenates all the chunks.

Step 7: Main Function

extract_document() integrates all the steps into a clean pipeline. It's the only entry point the user needs.

def extract_document(
    file_path: str,
    schema_type: str = "auto",
) -> ExtractionResult:
    """Complete document extraction pipeline.

    Args:
        file_path: Path to a PDF or image
        schema_type: "invoice", "receipt", "contract", or "auto" to detect

    Returns:
        ExtractionResult with data, metadata and metrics
    """
    start = time.time()
    result = ExtractionResult()

    try:
        doc_type = detect_document_type(file_path)
        result.document_type = doc_type.value

        text, pages, method = select_and_extract(file_path, doc_type)
        result.raw_text = text
        result.pages_processed = pages
        result.method = method.value
        result.cost_usd = METHOD_COSTS_PER_PAGE[method] * pages

        if schema_type == "auto":
            schema_type = detect_schema_type(text)
        result.schema_used = schema_type

        if pages > CHUNK_SIZE_PAGES and doc_type == DocumentType.PDF_TEXT:
            full_text, chunks_data, total_pages = extract_long_document(
                file_path, schema_type
            )
            result.raw_text = full_text
            result.pages_processed = total_pages
            data, confidence = merge_chunk_results(chunks_data, schema_type)
        else:
            data, confidence = extract_structured_data(text, schema_type)

        result.data = data
        result.confidence = confidence
        result.success = True

    except FileNotFoundError as e:
        result.errors.append(f"File not found: {e}")
    except ValueError as e:
        result.errors.append(f"Format error: {e}")
    except Exception as e:
        result.errors.append(f"Unexpected error: {type(e).__name__}: {e}")

    result.latency_seconds = round(time.time() - start, 2)
    return result

Demo: Complete Usage

def print_result(result: ExtractionResult):
    """Prints the extraction result in a readable way."""
    status = "SUCCESS" if result.success else "ERROR"
    print(f"\n{'='*60}")
    print(f"  Status:        {status}")
    print(f"  Doc type:      {result.document_type}")
    print(f"  Method:        {result.method}")
    print(f"  Schema:        {result.schema_used}")
    print(f"  Pages:         {result.pages_processed}")
    print(f"  Confidence:    {result.confidence:.0%}")
    print(f"  Est. cost:     ${result.cost_usd:.4f}")
    print(f"  Latency:       {result.latency_seconds}s")

    if result.data:
        print(f"  Extracted data:")
        for key, value in result.data.items():
            if isinstance(value, list) and len(value) > 2:
                print(f"    {key}: [{len(value)} items]")
            else:
                print(f"    {key}: {value}")

    if result.errors:
        print(f"  Errors:")
        for err in result.errors:
            print(f"    - {err}")
    print(f"{'='*60}")


# --- Example 1: Digital PDF invoice ---
result = extract_document("digital_invoice.pdf", schema_type="invoice")
print_result(result)

# --- Example 2: Scanned receipt (image) ---
result = extract_document("receipt_photo.jpg", schema_type="receipt")
print_result(result)

# --- Example 3: Long contract with auto-detection ---
result = extract_document("contract_20pages.pdf", schema_type="auto")
print_result(result)

# --- Example 4: Direct access to the JSON ---
result = extract_document("invoice.pdf")
if result.success:
    print(json.dumps(result.data, indent=2, ensure_ascii=False))

Expected output (digital invoice):

============================================================
  Status:        SUCCESS
  Doc type:      pdf_text
  Method:        pymupdf
  Schema:        invoice
  Pages:         1
  Confidence:    89%
  Est. cost:     $0.0030
  Latency:       1.84s
  Extracted data:
    date: 2024-03-15
    invoice_number: INV-2024-0847
    vendor: Acme Technologies S.A.
    total: 5220.0
    items: [3 items]
============================================================

Extension 1: Batch Processing

Processes multiple documents with progress tracking, accumulated costs, and a statistical summary.

@dataclass
class BatchResult:
    total: int = 0
    successful: int = 0
    failed: int = 0
    total_cost_usd: float = 0.0
    total_pages: int = 0
    avg_confidence: float = 0.0
    by_type: dict = field(default_factory=dict)
    by_method: dict = field(default_factory=dict)
    results: list[ExtractionResult] = field(default_factory=list)


def extract_batch(
    file_paths: list[str],
    schema_type: str = "auto",
    budget_usd: float | None = None,
) -> BatchResult:
    """Processes multiple documents with progress tracking."""
    batch = BatchResult()
    confidences = []

    for i, path in enumerate(file_paths):
        print(f"  [{i+1}/{len(file_paths)}] Processing: {Path(path).name}...")

        if budget_usd and batch.total_cost_usd >= budget_usd:
            fail = ExtractionResult()
            fail.errors.append(f"Budget exhausted: ${batch.total_cost_usd:.4f}/{budget_usd}")
            batch.results.append(fail)
            batch.failed += 1
            batch.total += 1
            continue

        result = extract_document(path, schema_type=schema_type)
        batch.results.append(result)
        batch.total += 1

        if result.success:
            batch.successful += 1
            batch.total_cost_usd += result.cost_usd
            batch.total_pages += result.pages_processed
            confidences.append(result.confidence)

            batch.by_type[result.document_type] = batch.by_type.get(result.document_type, 0) + 1
            batch.by_method[result.method] = batch.by_method.get(result.method, 0) + 1
        else:
            batch.failed += 1

    batch.avg_confidence = round(sum(confidences) / len(confidences), 2) if confidences else 0.0
    return batch


def print_batch_report(batch: BatchResult):
    """Prints the batch processing report."""
    print(f"\n{'='*60}")
    print(f"  BATCH REPORT")
    print(f"{'='*60}")
    print(f"  Total documents:   {batch.total}")
    print(f"  Successful:        {batch.successful}")
    print(f"  Failed:            {batch.failed}")
    print(f"  Total pages:       {batch.total_pages}")
    print(f"  Total cost:        ${batch.total_cost_usd:.4f}")
    print(f"  Avg. confidence:   {batch.avg_confidence:.0%}")

    if batch.by_type:
        print(f"  By type:")
        for doc_type, count in batch.by_type.items():
            print(f"    {doc_type}: {count}")

    if batch.by_method:
        print(f"  By method:")
        for method, count in batch.by_method.items():
            print(f"    {method}: {count}")
    print(f"{'='*60}")

Extension 2: Detailed Confidence Scoring

A more granular scoring that analyzes the quality of each extracted field, not just the ratio of filled fields.

@dataclass
class FieldScore:
    field_name: str
    present: bool
    plausible: bool
    score: float


def score_extraction(data: dict, schema_type: str) -> tuple[float, list[FieldScore]]:
    """Evaluates the extraction quality field by field.

    Checks:
    - Presence: the field has a non-null value
    - Plausibility: the value has the expected format
    """
    field_scores = []

    plausibility_checks = {
        "date": lambda v: bool(v and len(str(v)) == 10 and "-" in str(v)),
        "total": lambda v: isinstance(v, (int, float)) and v > 0,
        "subtotal": lambda v: isinstance(v, (int, float)) and v > 0,
        "tax": lambda v: isinstance(v, (int, float)) and v >= 0,
        "amount": lambda v: isinstance(v, (int, float)) and v > 0,
        "currency": lambda v: bool(v and len(str(v)) == 3),
        "items": lambda v: isinstance(v, list) and len(v) > 0,
        "parties": lambda v: isinstance(v, list) and len(v) >= 2,
        "key_clauses": lambda v: isinstance(v, list) and len(v) > 0,
    }

    for field_name, value in data.items():
        present = value is not None and value != "" and value != []

        check = plausibility_checks.get(field_name)
        plausible = check(value) if check and present else present

        score = 0.0
        if present:
            score = 1.0 if plausible else 0.5

        field_scores.append(FieldScore(
            field_name=field_name,
            present=present,
            plausible=plausible,
            score=score,
        ))

    total_score = sum(fs.score for fs in field_scores) / max(len(field_scores), 1)
    return round(total_score, 2), field_scores


def print_confidence_report(data: dict, schema_type: str):
    """Prints a detailed per-field confidence report."""
    total_score, field_scores = score_extraction(data, schema_type)
    print(f"\n  Detailed confidence: {total_score:.0%}")
    for fs in field_scores:
        status = "OK" if fs.plausible else ("partial" if fs.present else "missing")
        print(f"    {fs.field_name:<20} {status:<10} {fs.score:.1f}")

Project Troubleshooting

Problem 1: A scanned PDF is detected as digital

Symptom: A scanned PDF has an invisible OCR layer embedded and PyMuPDF extracts it, but the text is garbage.

Solution: Add a quality check: compute alpha_ratio = sum(c.isalpha() for c in text) / len(text). If it's below 0.5, reclassify as scanned and use Vision. Integrate this check in select_and_extract as a post-validation of the PyMuPDF text.

Problem 2: Invoice fields with inconsistent formats

Symptom: The LLM returns "total": "$1,234.56" instead of "total": 1234.56.

Solution: Add a pre-processor between the LLM's JSON and Pydantic that cleans monetary values: value.replace("$", "").replace(",", "") and then float(). Apply it to numeric fields before passing to schema_class(**raw_json).

Problem 3: A long document exceeds the timeout

Symptom: A 50-page contract takes more than 60 seconds and fails due to the OpenAI client timeout.

Solution: Adjust the client timeout (OpenAI(timeout=120.0)) and reduce CHUNK_SIZE_PAGES = 3 for very long documents.

Problem 4: The Vision API fails with low-resolution images

Symptom: Document photos taken in poor lighting return partial or incorrect text.

Solution: Pre-process with PIL before sending: ImageEnhance.Contrast(img).enhance(1.5) and ImageEnhance.Sharpness(img).enhance(2.0). Save as JPEG quality=95 before encoding to Base64.

Problem 5: The schema is auto-detected incorrectly

Symptom: A receipt is classified as an invoice because it contains the phrase "simplified invoice".

Solution: Use differentiated weights per keyword instead of a simple count:

SCHEMA_WEIGHTS: dict[str, dict[str, float]] = {
    "invoice": {"invoice": 2.0, "vendor": 1.5, "tax": 1.0, "invoice no": 2.0},
    "receipt": {"ticket": 2.0, "receipt": 2.0, "cashier": 1.5, "change": 1.0},
    "contract": {"contract": 2.0, "clause": 2.0, "term": 1.5, "signatory": 1.5},
}

Completeness Checklist

Core pipeline:

• Accepts PDFs and images (JPG, PNG, WEBP)
• Detects digital PDF vs scanned vs image
• Extracts text with PyMuPDF (digital), Vision (scanned/image)
• Tesseract available as an alternative method
• Pydantic schemas: InvoiceSchema, ReceiptSchema, ContractSchema
• Schema registry with keyword auto-detection
• Structured extraction with LLM + Pydantic validation
• Partial fallback when validation fails

Long documents:

• Page-based chunking (configurable CHUNK_SIZE_PAGES)
• Independent processing per chunk
• Result merging (scalars: first non-null, lists: concatenate)

Quality and metrics:

• ExtractionResult with all specified fields
• Estimated cost per extraction
• Confidence based on extracted fields
• Latency measured
• Errors captured without crashes

Extensions:

• Batch processing with progress tracking and budget
• Detailed per-field confidence scoring

Exercises

Exercise 1: Add a Medical Receipt Schema (Easy)

Create a MedicalReceiptSchema with fields: patient, doctor, hospital, date, diagnosis, medications (list with name, dosage, quantity), total. Register it in SCHEMA_REGISTRY with appropriate keywords.

class Medication(BaseModel):
    name: str = Field(default="", description="Medication name")
    dosage: Optional[str] = Field(default=None, description="Indicated dosage")
    quantity: int = Field(default=1, description="Quantity prescribed")
    price: float = Field(default=0.0)


class MedicalReceiptSchema(BaseModel):
    patient: Optional[str] = Field(default=None, description="Patient name")
    doctor: Optional[str] = Field(default=None, description="Physician name")
    hospital: Optional[str] = Field(default=None, description="Medical center")
    date: Optional[str] = Field(default=None, description="Date YYYY-MM-DD")
    diagnosis: Optional[str] = Field(default=None, description="Diagnosis")
    medications: list[Medication] = Field(default_factory=list)
    total: Optional[float] = Field(default=None)


SCHEMA_REGISTRY["medical_receipt"] = MedicalReceiptSchema
SCHEMA_KEYWORDS["medical_receipt"] = [
    "patient", "doctor", "physician", "prescription", "diagnosis",
    "medication", "dosage", "hospital", "clinic",
]

result = extract_document("medical_prescription.pdf", schema_type="medical_receipt")
print(json.dumps(result.data, indent=2, ensure_ascii=False))

Exercise 2: Post-Extraction Validation Layer (Medium)

Create a validate_extraction(data, schema_type) function that applies business rules to the extracted data: total must be >= subtotal, date cannot be in the future, items must add up to approximately the subtotal (±10%). Returns a list of warnings and a boolean is_valid.

from datetime import date


def validate_extraction(data: dict, schema_type: str) -> tuple[bool, list[str]]:
    """Validates extracted data against business rules."""
    warnings = []

    if schema_type in ("invoice", "receipt"):
        total = data.get("total")
        subtotal = data.get("subtotal")

        if total is not None and subtotal is not None:
            if total < subtotal:
                warnings.append(
                    f"Total ({total}) less than subtotal ({subtotal})"
                )

        date_str = data.get("date")
        if date_str:
            try:
                doc_date = date.fromisoformat(date_str)
                if doc_date > date.today():
                    warnings.append(f"Future date detected: {date_str}")
            except ValueError:
                warnings.append(f"Date with invalid format: {date_str}")

        items = data.get("items", [])
        if items and subtotal is not None:
            items_sum = sum(
                item.get("amount", 0) or item.get("price", 0) * item.get("quantity", 1)
                for item in items
            )
            if items_sum > 0 and abs(items_sum - subtotal) / subtotal > 0.10:
                warnings.append(
                    f"Items sum to {items_sum:.2f}, subtotal is {subtotal:.2f} (difference >10%)"
                )

    if schema_type == "contract":
        parties = data.get("parties", [])
        if len(parties) < 2:
            warnings.append("Contract with fewer than 2 identified parties")

    is_valid = len(warnings) == 0
    return is_valid, warnings


result = extract_document("invoice.pdf", schema_type="invoice")
if result.success:
    is_valid, validation_warnings = validate_extraction(result.data, result.schema_used)
    print(f"Validation: {'PASS' if is_valid else 'WARN'}")
    for w in validation_warnings:
        print(f"  - {w}")

Summary

In this project you built a complete Document Extractor that:

• Detects the type of document (digital PDF, scanned PDF, image) using PyMuPDF to analyze embedded text
• Extracts text with the optimal method: PyMuPDF for digital ones (free), Tesseract for local OCR, the Vision API for maximum quality
• Defines schemas in Pydantic for invoices, receipts and contracts with keyword auto-detection
• Extracts structured data with LLM + Pydantic validation, with a partial fallback when validation fails
• Handles long documents with page-based chunking and intelligent result merging
• Reports metrics on cost, confidence, method used and latency in each extraction

This extractor is the foundation of the Document Analyzer in Module 8, where you'll integrate RAG for Q&A over the extracted content and TTS for spoken summaries.

Next module: Module 4 — Image Generation. From analysis you move to creation: DALL-E, Stable Diffusion, and controlled generation pipelines.

Additional Resources

1. PyMuPDF Documentation — Text extraction and PDF rendering
2. Tesseract OCR — Open-source OCR engine
3. Pydantic V2 Docs — Models, validation and JSON Schema
4. OpenAI Vision Guide — Image processing with GPT-4
5. OpenAI Structured Outputs — JSON mode and response format
6. pdf2image — PDF-to-image conversion for OCR