Production Implementation: PDF & Text Extraction Engines for eDiscovery

import asyncio
import hashlib
import json
import logging
import gc
from pathlib import Path
from typing import Dict, Any, Optional
from enum import Enum

import pdfplumber
import pikepdf
import pytesseract
from pdf2image import convert_from_path
from pdfminer.pdfparser import PDFSyntaxError
from pydantic import BaseModel, ValidationError
from datetime import datetime, timezone

# Structured logging configuration for SIEM ingestion
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("pdf_extraction_engine")

class ExtractionStatus(str, Enum):
    SUCCESS = "SUCCESS"
    FALLBACK_OCR = "FALLBACK_OCR"
    QUARANTINED = "QUARANTINED"

class ErrorCategory(str, Enum):
    MALFORMED_XREF = "MALFORMED_XREF"
    MISSING_TEXT_LAYER = "MISSING_TEXT_LAYER"
    ENCRYPTED_NO_KEY = "ENCRYPTED_NO_KEY"
    STREAM_DECOMPRESSION_FAIL = "STREAM_DECOMPRESSION_FAIL"
    UNKNOWN = "UNKNOWN"

class ExtractionPayload(BaseModel):
    file_path: str
    sha256: str
    md5: str
    status: ExtractionStatus
    extracted_text: str
    page_count: int
    error_category: Optional[ErrorCategory] = None
    processing_timestamp: str
    engine_version: str = "1.0.0"

class ExtractionWorker:
    def __init__(self, queue_size: int = 50, batch_size: int = 15):
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=queue_size)
        self.batch_size = batch_size
        self._shutdown = False

    async def compute_hashes(self, file_path: Path) -> Dict[str, str]:
        """Deterministic cryptographic verification before extraction."""
        sha256 = hashlib.sha256()
        md5 = hashlib.md5()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                sha256.update(chunk)
                md5.update(chunk)
        return {"sha256": sha256.hexdigest(), "md5": md5.hexdigest()}

    async def extract_primary(self, file_path: Path) -> Dict[str, Any]:
        """High-fidelity text layer extraction via pdfplumber."""
        try:
            with pdfplumber.open(file_path) as pdf:
                pages = [page.extract_text() for page in pdf.pages]
                return {
                    "text": "\n".join(filter(None, pages)),
                    "page_count": len(pdf.pages),
                    "status": ExtractionStatus.SUCCESS
                }
        except PDFSyntaxError as e:
            return {"error": str(e), "category": ErrorCategory.MALFORMED_XREF}
        except Exception as e:
            return {"error": str(e), "category": ErrorCategory.UNKNOWN}

    async def extract_fallback(self, file_path: Path) -> Dict[str, Any]:
        """Secondary tier: repair-and-reparse, then rasterized OCR."""
        # First, attempt to recover a damaged cross-reference table with pikepdf,
        # then re-run the high-fidelity text-layer extractor on the repaired copy.
        try:
            repaired = file_path.with_suffix(".repaired.pdf")
            with pikepdf.open(file_path) as pdf:
                pdf.save(repaired)
            with pdfplumber.open(repaired) as pdf:
                pages = [page.extract_text() for page in pdf.pages]
            text = "\n".join(filter(None, pages))
            if text.strip():
                return {
                    "text": text,
                    "page_count": len(pages),
                    "status": ExtractionStatus.FALLBACK_OCR,
                }
        except Exception:
            pass

        # Final tier: rasterize each page and OCR it. pytesseract operates on
        # images, so the PDF must be converted to page images first.
        try:
            images = await asyncio.to_thread(convert_from_path, str(file_path))
            ocr_pages = [
                await asyncio.to_thread(pytesseract.image_to_string, image, lang="eng")
                for image in images
            ]
            return {
                "text": "\n".join(ocr_pages),
                "page_count": len(ocr_pages),
                "status": ExtractionStatus.FALLBACK_OCR,
            }
        except Exception as e:
            return {"error": str(e), "category": ErrorCategory.STREAM_DECOMPRESSION_FAIL}

    def build_payload(self, file_path: Path, hashes: Dict[str, str], result: Dict[str, Any]) -> ExtractionPayload:
        """Schema validation and payload normalization."""
        return ExtractionPayload(
            file_path=str(file_path),
            sha256=hashes["sha256"],
            md5=hashes["md5"],
            status=result.get("status", ExtractionStatus.QUARANTINED),
            extracted_text=result.get("text", ""),
            page_count=result.get("page_count", 0),
            error_category=result.get("category"),
            processing_timestamp=datetime.now(timezone.utc).isoformat()
        )

    async def process_file(self, file_path: Path) -> Optional[ExtractionPayload]:
        """Orchestrates extraction with explicit memory boundaries."""
        hashes = await self.compute_hashes(file_path)
        primary = await self.extract_primary(file_path)

        if primary.get("status") == ExtractionStatus.SUCCESS:
            return self.build_payload(file_path, hashes, primary)

        logger.warning(f"Primary extraction failed for {file_path.name}. Routing to fallback tier.")
        fallback = await self.extract_fallback(file_path)

        if fallback.get("status") == ExtractionStatus.FALLBACK_OCR:
            return self.build_payload(file_path, hashes, fallback)

        logger.error(f"Extraction exhausted all tiers for {file_path.name}. Quarantining.")
        return self.build_payload(file_path, hashes, {
            "status": ExtractionStatus.QUARANTINED,
            "category": fallback.get("category", ErrorCategory.UNKNOWN)
        })

    async def worker_loop(self):
        """Stateless consumer loop with bounded concurrency."""
        while not self._shutdown:
            batch = []
            for _ in range(self.batch_size):
                try:
                    item = await asyncio.wait_for(self.queue.get(), timeout=2.0)
                    batch.append(item)
                except asyncio.TimeoutError:
                    break

            if not batch:
                continue

            for file_path in batch:
                try:
                    payload = await self.process_file(file_path)
                    if payload:
                        logger.info(json.dumps(payload.model_dump()))
                except ValidationError as ve:
                    logger.error(f"Schema validation failed: {ve}")
                except Exception as e:
                    logger.exception(f"Unhandled worker exception: {e}")
                finally:
                    self.queue.task_done()

            # Explicit memory reclamation
            gc.collect()

    async def enqueue(self, file_path: Path):
        await self.queue.put(file_path)

    async def shutdown(self):
        self._shutdown = True
        await self.queue.join()