7. Troubleshooting Documents

Description

In this capsule we cover the most common problems when processing documents: corrupt PDFs, low scan quality, API errors, encoding, badly extracted tables and cost optimization. You'll learn to diagnose, solve and prevent failures in document pipelines with diagnostic functions, preprocessing and fallback strategies.

Why it matters: In production, real documents are imperfect. Knowing how to anticipate and handle these cases is what separates a prototype from a robust system.

Quick Error Reference Table

Common Problems and Solutions

1. Corrupt or damaged PDF

Symptoms: fitz.open() raises FileDataError, pdf2image fails.

import fitz
from pathlib import Path

def diagnose_pdf(path: str) -> dict:
    result = {"path": path, "valid": False, "pages": 0, "issues": []}
    if not Path(path).exists():
        result["issues"].append("File not found")
        return result
    try:
        doc = fitz.open(path)
        result["pages"] = len(doc)
        result["valid"] = True
        for i, page in enumerate(doc):
            if len(page.get_text().strip()) < 5:
                result["issues"].append(f"Page {i+1}: no extractable text")
        doc.close()
    except fitz.FileDataError as e:
        result["issues"].append(f"Corrupt PDF: {e}")
    return result

Fallback with pypdf (more tolerant of malformed PDFs):

from pypdf import PdfReader

def try_pypdf_fallback(path: str) -> str:
    reader = PdfReader(path)
    return "\n\n".join(p.extract_text() or "" for p in reader.pages)

2. Password-protected PDF

Symptoms: fitz.open() fails with an encryption message.

def open_pdf_with_password(path: str, passwords: list[str] | None = None):
    passwords = passwords or ["", "1234", "password"]
    for pwd in passwords:
        try:
            doc = fitz.open(path)
            if doc.is_encrypted:
                if doc.authenticate(pwd):
                    return doc
                doc.close()
            else:
                return doc
        except Exception:
            continue
    return None

3. Scanned PDF without text (images only)

Symptoms: page.get_text() returns an empty string or fewer than 20 characters.

import io
from PIL import Image
import pytesseract

def classify_pdf_type(path: str) -> str:
    doc = fitz.open(path)
    digital, scanned = 0, 0
    for page in doc:
        if len(page.get_text().strip()) > 50:
            digital += 1
        elif len(page.get_images()) > 0:
            scanned += 1
    doc.close()
    if scanned == 0: return "digital"
    if digital == 0: return "scanned"
    return "mixed"

def extract_from_scanned_page(doc, page_num: int, dpi: int = 300) -> str:
    mat = fitz.Matrix(dpi / 72, dpi / 72)
    pix = doc[page_num].get_pixmap(matrix=mat)
    img = Image.open(io.BytesIO(pix.tobytes("png")))
    return pytesseract.image_to_string(img, lang="eng")

4. Scanned documents with quality problems

Symptoms: OCR returns text with many errors, random characters.

a) Blurry image:

from PIL import ImageEnhance, ImageFilter

def fix_blurry_scan(img: Image.Image) -> Image.Image:
    img = img.convert("L")
    img = ImageEnhance.Sharpness(img).enhance(2.5)
    return ImageEnhance.Contrast(img).enhance(1.8)

b) Rotated or skewed image:

import numpy as np

def deskew_image(img: Image.Image) -> Image.Image:
    arr = np.array(img.convert("L"))
    coords = np.column_stack(np.where((arr < 128) > 0))
    if len(coords) < 100:
        return img
    best_angle, best_score = 0, 0
    for angle in np.arange(-10, 10, 0.5):
        rad = np.radians(angle)
        proj = coords[:, 0] * np.cos(rad) + coords[:, 1] * np.sin(rad)
        hist, _ = np.histogram(proj, bins=50)
        if np.var(hist) > best_score:
            best_score = np.var(hist)
            best_angle = angle
    if abs(best_angle) > 0.5:
        return img.rotate(best_angle, fillcolor=255, expand=True)
    return img

5. Table extraction failures

Symptoms: Tables as plain text, mixed columns, lost rows.

import camelot
import re, json

def extract_tables_from_pdf(path: str, page: str = "1") -> list[dict]:
    tables = camelot.read_pdf(path, pages=page, flavor="lattice")
    if not tables:
        tables = camelot.read_pdf(path, pages=page, flavor="stream")
    return [{"accuracy": t.parsing_report.get("accuracy", 0),
             "data": t.df.to_dict(orient="records")} for t in tables]

TABLE_PROMPT = "Extract the table as JSON. Array of objects (row=object). Headers as keys. Empty cells: null."

def extract_table_with_vision(image_path: str, client) -> list[dict]:
    import base64
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": [
            {"type": "text", "text": TABLE_PROMPT},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
        ]}], temperature=0)
    text = response.choices[0].message.content
    match = re.search(r'\[[\s\S]*\]', text)
    return json.loads(match.group()) if match else []

6. Encoding problems

Symptoms: UnicodeDecodeError, characters like é instead of é.

import chardet

def detect_and_decode(raw_bytes: bytes) -> str:
    detection = chardet.detect(raw_bytes)
    encoding = detection.get("encoding", "utf-8")
    if detection.get("confidence", 0) < 0.5:
        for enc in ["utf-8", "latin-1", "cp1252", "iso-8859-1"]:
            try:
                return raw_bytes.decode(enc)
            except UnicodeDecodeError:
                continue
    return raw_bytes.decode(encoding, errors="replace")

def clean_extracted_text(text: str) -> str:
    text = text.replace("\x00", "").replace("\ufffd", "")
    text = re.sub(r'[\x01-\x08\x0b\x0c\x0e-\x1f]', '', text)
    return re.sub(r'\s+', ' ', text).strip()

7. Very large PDFs (memory management)

Symptoms: MemoryError, the process hangs.

import gc

def process_large_pdf(path: str, process_fn, batch_size: int = 5) -> list:
    doc = fitz.open(path)
    results = []
    for start in range(0, len(doc), batch_size):
        for page_num in range(start, min(start + batch_size, len(doc))):
            results.append(process_fn(doc[page_num], page_num))
        gc.collect()
    doc.close()
    return results

def estimate_cost(path: str, cost_per_page: float = 0.01) -> dict:
    doc = fitz.open(path)
    pages = len(doc)
    doc.close()
    return {"pages": pages,
            "file_mb": round(Path(path).stat().st_size / 1048576, 2),
            "est_cost_usd": round(pages * cost_per_page, 2),
            "recommendation": "batch" if pages > 50 else "direct"}

8. Documents with multiple languages

Symptoms: Tesseract returns incorrect characters in bilingual documents.

from langdetect import detect, DetectorFactory
DetectorFactory.seed = 0

LANG_MAP = {"es": "spa", "en": "eng", "fr": "fra", "de": "deu", "pt": "por"}

def ocr_multilang(img: Image.Image, text_sample: str = "") -> str:
    lang = "en"
    if len(text_sample.strip()) > 20:
        try: lang = detect(text_sample)
        except Exception: pass
    tess_lang = LANG_MAP.get(lang, "eng")
    try:
        return pytesseract.image_to_string(img, lang=tess_lang)
    except pytesseract.TesseractError:
        return pytesseract.image_to_string(img, lang="eng")

9. Rate limits in batch processing

Symptoms: 429 Too Many Requests when processing many documents.

import time
from dataclasses import dataclass, field

@dataclass
class RateLimiter:
    max_requests: int = 50
    window_seconds: float = 60.0
    _timestamps: list = field(default_factory=list)

    def wait_if_needed(self):
        now = time.time()
        self._timestamps = [t for t in self._timestamps if now - t < self.window_seconds]
        if len(self._timestamps) >= self.max_requests:
            sleep_time = self._timestamps[0] + self.window_seconds - now
            if sleep_time > 0:
                time.sleep(sleep_time)
        self._timestamps.append(time.time())

def call_with_retry(func, *args, max_retries: int = 5, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if any(k in str(e).lower() for k in ["rate", "429", "timeout"]):
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
            raise

10. Invalid JSON in structured extraction

Symptoms: json.JSONDecodeError, Pydantic ValidationError.

def extract_json_from_response(text: str) -> dict | list:
    text = text.strip()
    code_block = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
    if code_block:
        text = code_block.group(1).strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    for pattern in [r'\{[\s\S]*\}', r'\[[\s\S]*\]']:
        match = re.search(pattern, text)
        if match:
            try: return json.loads(match.group())
            except json.JSONDecodeError: continue
    raise ValueError(f"No valid JSON found: {text[:200]}")

Diagnostic Function

Analyzes a document before processing it, detecting problems and suggesting the best strategy.

def full_document_diagnostic(path: str) -> dict:
    report = {"path": path, "exists": False, "file_type": None, "pages": 0,
              "pdf_type": None, "issues": [], "recommended_strategy": None}
    p = Path(path)
    if not p.exists():
        report["issues"].append("File not found")
        return report
    report["exists"] = True
    report["file_type"] = p.suffix.lower()

    if report["file_type"] in [".png", ".jpg", ".jpeg", ".tiff", ".bmp"]:
        report["pdf_type"] = "image"
        report["pages"] = 1
        report["recommended_strategy"] = "ocr_or_vision"
        return report
    if report["file_type"] != ".pdf":
        report["issues"].append(f"Unsupported format: {report['file_type']}")
        return report

    try:
        doc = fitz.open(path)
    except Exception as e:
        report["issues"].append(f"Cannot open: {e}")
        report["recommended_strategy"] = "repair_first"
        return report

    if doc.is_encrypted:
        report["issues"].append("Encrypted PDF")
        report["recommended_strategy"] = "decrypt_first"
        doc.close()
        return report

    report["pages"] = len(doc)
    digital, scanned = 0, 0
    for i in range(min(len(doc), 10)):
        text = doc[i].get_text().strip()
        if len(text) > 50: digital += 1
        elif len(doc[i].get_images()) > 0: scanned += 1
    doc.close()

    if scanned > digital: report["pdf_type"], report["recommended_strategy"] = "scanned", "ocr_pipeline"
    elif digital > scanned: report["pdf_type"], report["recommended_strategy"] = "digital", "text_extraction"
    else: report["pdf_type"], report["recommended_strategy"] = "mixed", "hybrid_pipeline"
    if report["pages"] > 50: report["issues"].append("Long document: use chunking")
    return report

Preprocessing Pipeline

A configurable image preprocessing pipeline to improve OCR quality.

from PIL import Image, ImageEnhance, ImageFilter

class DocumentPreprocessor:
    def __init__(self, denoise=True, deskew=True, binarize=True):
        self.denoise = denoise
        self.deskew = deskew
        self.binarize = binarize
        self.steps_applied = []

    def process(self, img: Image.Image) -> Image.Image:
        self.steps_applied = []
        img = img.convert("L"); self.steps_applied.append("grayscale")
        if self.denoise:
            img = img.filter(ImageFilter.MedianFilter(size=3)); self.steps_applied.append("denoise")
        if self.deskew:
            img = self._deskew(img)
        img = ImageEnhance.Contrast(img).enhance(2.0); self.steps_applied.append("contrast")
        if self.binarize:
            img = img.point(lambda x: 0 if x < 140 else 255); self.steps_applied.append("binarize")
        return img

    def _deskew(self, img: Image.Image) -> Image.Image:
        arr = np.array(img)
        coords = np.column_stack(np.where((arr < 128) > 0))
        if len(coords) < 100: return img
        best_angle, best_score = 0, 0
        for a in np.arange(-10, 10, 0.5):
            rad = np.radians(a)
            proj = coords[:, 0] * np.cos(rad) + coords[:, 1] * np.sin(rad)
            h, _ = np.histogram(proj, bins=50)
            if np.var(h) > best_score: best_score, best_angle = np.var(h), a
        if abs(best_angle) > 0.5:
            self.steps_applied.append(f"deskew({best_angle:.1f}°)")
            return img.rotate(best_angle, fillcolor=255, expand=True)
        return img

Fallback Strategy for Documents

A cascade that tries the cheapest option first and only escalates if it fails: direct text → OCR → Vision API → log and skip.

import logging

logger = logging.getLogger("document_fallback")

def fallback_extract(path: str, vision_client=None) -> dict:
    result = {"path": path, "text": "", "method": "failed", "attempts": []}

    # 1. Direct text (free, fast)
    try:
        doc = fitz.open(path)
        combined = "\n".join(page.get_text() for page in doc).strip()
        doc.close()
        result["attempts"].append({"method": "text", "chars": len(combined)})
        if len(combined) > 50:
            result["text"], result["method"] = combined, "text_extraction"
            return result
    except Exception as e:
        result["attempts"].append({"method": "text", "error": str(e)})

    # 2. OCR with Tesseract + preprocessing (free, slower)
    try:
        doc = fitz.open(path)
        preprocessor = DocumentPreprocessor()
        ocr_texts = []
        for i in range(min(len(doc), 20)):
            pix = doc[i].get_pixmap(matrix=fitz.Matrix(2, 2))
            img = preprocessor.process(Image.open(io.BytesIO(pix.tobytes("png"))))
            ocr_texts.append(pytesseract.image_to_string(img, lang="eng"))
        doc.close()
        combined = "\n".join(ocr_texts).strip()
        result["attempts"].append({"method": "ocr", "chars": len(combined)})
        if len(combined) > 50:
            result["text"], result["method"] = combined, "ocr_tesseract"
            return result
    except Exception as e:
        result["attempts"].append({"method": "ocr", "error": str(e)})

    # 3. Vision API (cost per request, more accurate)
    if vision_client:
        try:
            import base64
            doc = fitz.open(path)
            b64 = base64.b64encode(doc[0].get_pixmap(matrix=fitz.Matrix(2,2)).tobytes("png")).decode()
            doc.close()
            resp = vision_client.chat.completions.create(model="gpt-4o", messages=[{"role":"user","content":[
                {"type":"text","text":"Extract all the text from this document."},
                {"type":"image_url","image_url":{"url":f"data:image/png;base64,{b64}"}}
            ]}], temperature=0)
            text = resp.choices[0].message.content.strip()
            result["attempts"].append({"method": "vision", "chars": len(text)})
            if len(text) > 20:
                result["text"], result["method"] = text, "vision_api"
                return result
        except Exception as e:
            result["attempts"].append({"method": "vision", "error": str(e)})

    logger.error(f"All methods failed: {path}")
    return result

Monitoring and Logging

Structured logging for document pipelines in production.

import json, time, logging
from dataclasses import dataclass, field, asdict
from datetime import datetime

@dataclass
class ProcessingRecord:
    path: str
    method: str = ""
    success: bool = False
    duration: float = 0.0
    error: str = ""
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class DocumentPipelineMonitor:
    def __init__(self):
        self.records: list[ProcessingRecord] = []

    def track(self, func, path: str, **kwargs) -> dict:
        rec = ProcessingRecord(path=path)
        start = time.time()
        try:
            result = func(path, **kwargs)
            rec.success = True
            rec.method = result.get("method", "unknown")
            return result
        except Exception as e:
            rec.error = str(e)
            return {"error": str(e)}
        finally:
            rec.duration = round(time.time() - start, 3)
            self.records.append(rec)

    def get_summary(self) -> dict:
        total = len(self.records)
        if total == 0: return {"total": 0}
        ok = sum(1 for r in self.records if r.success)
        errors = {}
        for r in self.records:
            if r.error: errors[r.error[:80]] = errors.get(r.error[:80], 0) + 1
        return {"total": total, "successes": ok, "failures": total - ok,
                "success_rate": f"{ok/total*100:.1f}%",
                "avg_duration": round(sum(r.duration for r in self.records)/total, 3),
                "top_errors": dict(sorted(errors.items(), key=lambda x:-x[1])[:5])}

    def export_records(self, path: str):
        with open(path, "w") as f:
            json.dump([asdict(r) for r in self.records], f, indent=2, ensure_ascii=False)

Usage:

monitor = DocumentPipelineMonitor()
for pdf_path in pdf_files:
    monitor.track(fallback_extract, pdf_path, vision_client=client)
print(monitor.get_summary())

Robustness Checklist

• Validate the PDF before processing (full diagnostic)
• Handle password-protected PDFs
• Classify the PDF type (digital / scanned / mixed)
• Fallback: text → OCR → Vision API
• Preprocess images (deskew, denoise, binarize)
• Detect and handle encoding correctly
• Retry with backoff + rate limiting for APIs
• Batch processing for large PDFs
• Structured logging and configurable timeout
• Validate extracted JSON before using it

Exercises

Exercise 1: Complete diagnostic function

Create diagnose_document(path) that returns the file type, validity, number of pages, PDF type (digital/scanned/mixed), a list of problems and a recommended strategy.

def diagnose_document(path: str) -> dict:
    report = {"path": path, "exists": False, "file_type": None, "valid": False,
              "pages": 0, "pdf_type": None, "issues": [], "strategy": None}
    p = Path(path)
    if not p.exists():
        report["issues"].append("File not found")
        return report
    report["exists"] = True
    report["file_type"] = p.suffix.lower()

    if report["file_type"] in {".png", ".jpg", ".jpeg", ".tiff"}:
        return {**report, "valid": True, "pages": 1, "strategy": "ocr_or_vision"}
    if report["file_type"] != ".pdf":
        report["issues"].append(f"Unsupported: {report['file_type']}")
        return report
    try:
        doc = fitz.open(path)
    except Exception as e:
        report["issues"].append(f"Cannot open: {e}")
        return {**report, "strategy": "repair_first"}

    report["valid"], report["pages"] = True, len(doc)
    if doc.is_encrypted:
        doc.close()
        return {**report, "issues": ["Encrypted PDF"], "strategy": "decrypt_first"}

    digital, scanned = 0, 0
    for i in range(min(len(doc), 10)):
        if len(doc[i].get_text().strip()) > 50: digital += 1
        elif len(doc[i].get_images()) > 0: scanned += 1
    doc.close()

    if scanned > digital: report["pdf_type"], report["strategy"] = "scanned", "ocr_pipeline"
    elif digital > scanned: report["pdf_type"], report["strategy"] = "digital", "text_extraction"
    else: report["pdf_type"], report["strategy"] = "mixed", "hybrid_pipeline"
    return report

Exercise 2: Configurable preprocessing pipeline

Create OCRPreprocessor with methods for: grayscale, noise filter, skew correction, contrast and binarization. The run(img) method executes all the steps and returns (image, list_of_steps).

class OCRPreprocessor:
    def __init__(self, denoise=True, deskew=True, binarize=True, contrast=2.0):
        self.denoise = denoise
        self.deskew = deskew
        self.binarize = binarize
        self.contrast = contrast

    def run(self, img: Image.Image) -> tuple[Image.Image, list[str]]:
        steps = []
        img = img.convert("L"); steps.append("grayscale")
        if self.denoise:
            img = img.filter(ImageFilter.MedianFilter(size=3)); steps.append("denoise")
        if self.deskew:
            arr = np.array(img)
            coords = np.column_stack(np.where((arr < 128) > 0))
            if len(coords) > 100:
                best_a, best_v = 0, 0
                for a in np.arange(-10, 10, 0.5):
                    r = np.radians(a)
                    h, _ = np.histogram(coords[:,0]*np.cos(r)+coords[:,1]*np.sin(r), bins=50)
                    if np.var(h) > best_v: best_v, best_a = np.var(h), a
                if abs(best_a) > 0.5:
                    img = img.rotate(best_a, fillcolor=255, expand=True)
                    steps.append(f"deskew({best_a:.1f}°)")
        img = ImageEnhance.Contrast(img).enhance(self.contrast); steps.append("contrast")
        if self.binarize:
            img = img.point(lambda x: 0 if x < 140 else 255); steps.append("binarize")
        return img, steps

Exercise 3: Fallback cascade for documents

Implement robust_extract(path, client=None) with a cascade: (1) PyMuPDF text, (2) OCR Tesseract, (3) Vision API, (4) error with the methods attempted. Each step must be recorded in attempts.

def robust_extract(path: str, client=None) -> dict:
    result = {"path": path, "text": "", "method": "none", "attempts": []}

    # Step 1: direct text
    try:
        doc = fitz.open(path)
        text = "\n".join(p.get_text() for p in doc).strip(); doc.close()
        result["attempts"].append({"method": "pymupdf", "chars": len(text)})
        if len(text) > 50:
            return {**result, "text": text, "method": "pymupdf"}
    except Exception as e:
        result["attempts"].append({"method": "pymupdf", "error": str(e)})

    # Step 2: OCR
    try:
        doc = fitz.open(path)
        parts = []
        for i in range(min(len(doc), 10)):
            pix = doc[i].get_pixmap(matrix=fitz.Matrix(2, 2))
            img = ImageEnhance.Contrast(Image.open(io.BytesIO(pix.tobytes("png"))).convert("L")).enhance(2.0)
            parts.append(pytesseract.image_to_string(img, lang="eng"))
        doc.close()
        text = "\n".join(parts).strip()
        result["attempts"].append({"method": "tesseract", "chars": len(text)})
        if len(text) > 50:
            return {**result, "text": text, "method": "tesseract"}
    except Exception as e:
        result["attempts"].append({"method": "tesseract", "error": str(e)})

    # Step 3: Vision API
    if client:
        try:
            import base64
            doc = fitz.open(path)
            b64 = base64.b64encode(doc[0].get_pixmap(matrix=fitz.Matrix(2,2)).tobytes("png")).decode()
            doc.close()
            resp = client.chat.completions.create(model="gpt-4o", messages=[{"role":"user","content":[
                {"type":"text","text":"Extract all the text from this document."},
                {"type":"image_url","image_url":{"url":f"data:image/png;base64,{b64}"}}
            ]}], temperature=0)
            text = resp.choices[0].message.content.strip()
            result["attempts"].append({"method": "vision", "chars": len(text)})
            if len(text) > 20:
                return {**result, "text": text, "method": "vision"}
        except Exception as e:
            result["attempts"].append({"method": "vision", "error": str(e)})

    return {**result, "method": "failed"}

Exercise 4: Batch processor with error handling

Create BatchDocumentProcessor that processes a list of PDFs with: rate limiting (N per minute), retry with backoff, logging, and a summary() method with the success rate, frequent errors and total time.

@dataclass
class DocResult:
    path: str
    success: bool = False
    method: str = ""
    duration: float = 0.0
    error: str = ""

class BatchDocumentProcessor:
    def __init__(self, extract_fn, max_per_minute=30, max_retries=3):
        self.extract_fn = extract_fn
        self.max_per_minute = max_per_minute
        self.max_retries = max_retries
        self.results: list[DocResult] = []
        self._times: list[float] = []

    def _throttle(self):
        now = time.time()
        self._times = [t for t in self._times if now - t < 60]
        if len(self._times) >= self.max_per_minute:
            wait = 60 - (now - self._times[0])
            if wait > 0: time.sleep(wait)
        self._times.append(time.time())

    def process_batch(self, paths: list[str]) -> list[DocResult]:
        for path in paths:
            rec = DocResult(path=path)
            start = time.time()
            for attempt in range(self.max_retries):
                try:
                    self._throttle()
                    r = self.extract_fn(path)
                    rec.success, rec.method = True, r.get("method", "")
                    break
                except Exception as e:
                    rec.error = str(e)
                    if attempt < self.max_retries - 1: time.sleep(2**attempt)
            rec.duration = round(time.time() - start, 3)
            self.results.append(rec)
        return self.results

    def summary(self) -> dict:
        total = len(self.results)
        if total == 0: return {"total": 0}
        ok = sum(1 for r in self.results if r.success)
        errors = {}
        for r in self.results:
            if r.error: errors[r.error[:60]] = errors.get(r.error[:60], 0) + 1
        return {"total": total, "success": ok, "failed": total-ok,
                "rate": f"{ok/total*100:.1f}%",
                "total_time": round(sum(r.duration for r in self.results), 2),
                "top_errors": dict(sorted(errors.items(), key=lambda x:-x[1])[:3])}

Additional Resources

1. PyMuPDF · PIL ImageEnhance · Tenacity
2. Circuit Breaker pattern · Camelot · chardet