Lesson 16.12: Document Ingestion & Content Extraction

# Create directory and virtual environment
mkdir ai-document-ingestion
cd ai-document-ingestion
python -m venv .venv

# Activate environment (Windows)
.venv\Scripts\activate
# Or Linux/macOS: source .venv/bin/activate

# Install core production dependencies
pip install pypdf==5.1.0 python-docx==1.1.2 fastapi==0.115.0 uvicorn==0.31.0 python-multipart==0.0.12 pillow==10.4.0 pytesseract==0.3.13 pytest==8.3.3

from io import BytesIO
from pypdf import PdfReader
from typing import List, Dict, Any

MAX_PDF_PAGES = 300  # Memory safety cap on document length

def extract_pdf_pages(pdf_bytes: bytes) -> List[Dict[str, Any]]:
    """
    Extracts text per page from a digital PDF byte array.
    Rejects encrypted files and tracks 1-indexed page boundaries for citations.
    """
    reader = PdfReader(BytesIO(pdf_bytes))
    
    # Check for password-protected or encrypted PDFs
    if reader.is_encrypted:
        raise ValueError("Encrypted or password-protected PDFs are not supported.")
        
    page_count = len(reader.pages)
    if page_count == 0:
        raise ValueError("The uploaded PDF contains zero pages.")
        
    if page_count > MAX_PDF_PAGES:
        raise ValueError(f"Document contains {page_count} pages, exceeding the maximum limit of {MAX_PDF_PAGES}.")
        
    extracted_pages = []
    for page_number, page in enumerate(reader.pages, start=1):
        # extraction_mode="layout" provides better horizontal spacing alignment
        raw_text = page.extract_text(extraction_mode="layout") or page.extract_text() or ""
        extracted_pages.append({
            "unit_type": "page",
            "unit_index": page_number,
            "text": raw_text.strip(),
            "character_count": len(raw_text.strip())
        })
        
    return extracted_pages

from io import BytesIO
from docx import Document
from docx.text.paragraph import Paragraph
from docx.table import Table
from typing import List, Dict, Any

def extract_docx_blocks(docx_bytes: bytes) -> List[Dict[str, Any]]:
    """
    Extracts paragraphs and tables sequentially using iter_inner_content().
    This preserves the natural reading flow of the document.
    """
    doc = Document(BytesIO(docx_bytes))
    blocks = []
    
    for block_index, item in enumerate(doc.iter_inner_content(), start=1):
        if isinstance(item, Paragraph):
            text = item.text.strip()
            if text:
                blocks.append({
                    "unit_type": "paragraph",
                    "unit_index": block_index,
                    "style": item.style.name if item.style else "Normal",
                    "text": text
                })
        elif isinstance(item, Table):
            table_rows = []
            for row in item.rows:
                cells = [cell.text.strip() for cell in row.cells]
                table_rows.append(cells)
                
            if table_rows:
                blocks.append({
                    "unit_type": "table",
                    "unit_index": block_index,
                    "rows": table_rows
                })
                
    return blocks

from PIL import Image
import pytesseract
from io import BytesIO

def extract_image_ocr(image_bytes: bytes, lang: str = "ben+eng") -> str:
    """
    Extracts text from image byte streams in both Bengali and English.
    Specifying lang="ben+eng" instructs Tesseract to scan for both character sets concurrently.
    """
    try:
        image = Image.open(BytesIO(image_bytes))
        
        # Convert to RGB to ensure uniform color depth for binarization
        if image.mode != "RGB":
            image = image.convert("RGB")
            
        extracted_text = pytesseract.image_to_string(image, lang=lang)
        return extracted_text.strip()
    except Exception as exc:
        raise RuntimeError(f"OCR execution failure: {str(exc)}") from exc

import re
import unicodedata

def normalize_text(text: str) -> str:
    """
    Cleans and normalizes extracted text while preserving semantic meaning:
    1. Unicode NFC canonical composition.
    2. Universal line break standardization (\n).
    3. Removal of non-breaking and zero-width spaces.
    4. Consolidation of excessive spaces and blank lines.
    """
    if not text:
        return ""
        
    # 1. Unicode NFC Canonical Composition
    text = unicodedata.normalize("NFC", text)
    
    # 2. Universal newline standardization
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    
    # 3. Strip non-breaking and zero-width characters
    text = text.replace("\xa0", " ").replace("\u200b", "")
    
    # 4. Collapse horizontal whitespace
    text = re.sub(r"[ \t]+", " ", text)
    
    # 5. Trim leading and trailing spaces per line
    text = re.sub(r" *\n *", "\n", text)
    
    # 6. Compress 3+ consecutive newlines into two
    text = re.sub(r"\n{3,}", "\n\n", text)
    
    return text.strip()

from pydantic import BaseModel
from typing import List, Optional, Literal

class DocumentUnit(BaseModel):
    unit_type: Literal["page", "paragraph", "table"]
    unit_index: int
    text: Optional[str] = None
    rows: Optional[List[List[str]]] = None
    style: Optional[str] = None

class DocumentMetadata(BaseModel):
    original_filename: str
    file_type: str
    content_hash: str
    page_count: Optional[int] = None
    unit_count: int
    character_count: int

class AIReadyDocument(BaseModel):
    document_id: str
    tenant_id: str
    status: Literal["extracted", "needs_ocr_or_review"]
    metadata: DocumentMetadata
    units: List[DocumentUnit]
    full_text: str

import hashlib
from pathlib import Path
from app.normalization import normalize_text
from app.extractors import extract_pdf_pages, extract_docx_blocks

def extract_document(filename: str, file_bytes: bytes) -> dict:
    """
    Routes files to format extractors and outputs a normalized, unified payload.
    """
    ext = Path(filename).suffix.lower()
    content_hash = hashlib.sha256(file_bytes).hexdigest()
    
    if ext == ".pdf":
        raw_units = extract_pdf_pages(file_bytes)
        file_type = "pdf"
        page_count = len(raw_units)
    elif ext == ".docx":
        raw_units = extract_docx_blocks(file_bytes)
        file_type = "docx"
        page_count = None
    else:
        raise ValueError(f"Unsupported file format: {ext}. Only .pdf and .docx are supported.")
        
    cleaned_units = []
    text_pieces = []
    
    for u in raw_units:
        if u["unit_type"] == "table":
            cleaned_rows = []
            for row in u["rows"]:
                cleaned_rows.append([normalize_text(cell) for cell in row])
            cleaned_units.append({
                "unit_type": "table",
                "unit_index": u["unit_index"],
                "rows": cleaned_rows
            })
            for r in cleaned_rows:
                text_pieces.append(" | ".join(r))
        else:
            norm = normalize_text(u.get("text", ""))
            cleaned_units.append({
                "unit_type": u["unit_type"],
                "unit_index": u["unit_index"],
                "style": u.get("style"),
                "text": norm
            })
            if norm:
                text_pieces.append(norm)
                
    full_text = "\n\n".join(text_pieces).strip()
    
    return {
        "file_type": file_type,
        "content_hash": content_hash,
        "units": cleaned_units,
        "full_text": full_text,
        "metadata": {
            "original_filename": filename,
            "page_count": page_count,
            "unit_count": len(cleaned_units),
            "character_count": len(full_text)
        }
    }

from fastapi import FastAPI, UploadFile, File, Header, HTTPException, status
import uuid
from app.extractors import extract_document

app = FastAPI(title="Byteion AI Document Ingestion Service", version="1.0.0")

MAX_UPLOAD_SIZE = 10 * 1024 * 1024  # 10 Megabytes
CHUNK_BUFFER_SIZE = 1024 * 1024     # 1 Megabyte buffer

async def read_limited_file(upload: UploadFile, max_bytes: int = MAX_UPLOAD_SIZE) -> bytes:
    """Stream reader with real-time memory exhaustion protection."""
    chunks = []
    current_size = 0
    while True:
        chunk = await upload.read(CHUNK_BUFFER_SIZE)
        if not chunk:
            break
        current_size += len(chunk)
        if current_size > max_bytes:
            raise HTTPException(
                status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                detail=f"Uploaded file exceeds maximum limit of {max_bytes // (1024*1024)} MB."
            )
        chunks.append(chunk)
    return b"".join(chunks)

@app.post("/documents/ingest")
async def ingest_document(
    file: UploadFile = File(...),
    x_tenant_id: str = Header(..., description="Secure multi-tenant identifier")
):
    if not file.filename:
        raise HTTPException(status_code=400, detail="Filename parameter is required.")
        
    filename = file.filename
    lowered = filename.lower()
    
    # Extension validation
    if not lowered.endswith((".pdf", ".docx")):
        raise HTTPException(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            detail="Only .pdf and .docx file extensions are accepted."
        )
        
    try:
        file_bytes = await read_limited_file(file)
        if len(file_bytes) == 0:
            raise HTTPException(status_code=400, detail="Uploaded file is empty.")
            
        # Verify PDF magic byte signature
        if lowered.endswith(".pdf") and not file_bytes.startswith(b"%PDF-"):
            raise HTTPException(
                status_code=400, 
                detail="Invalid PDF format: file magic byte signature mismatch."
            )
            
        doc_result = extract_document(filename, file_bytes)
        
        # Check if extractable text exists
        if not doc_result["full_text"].strip():
            return {
                "document_id": f"doc_{uuid.uuid4().hex[:12]}",
                "tenant_id": x_tenant_id,
                "status": "needs_ocr_or_review",
                "message": "No extractable text found. Document may be a scanned image.",
                "metadata": doc_result["metadata"]
            }
            
        return {
            "document_id": f"doc_{uuid.uuid4().hex[:12]}",
            "tenant_id": x_tenant_id,
            "status": "extracted",
            "file_type": doc_result["file_type"],
            "metadata": doc_result["metadata"],
            "text_preview": doc_result["full_text"][:500] + "..." if len(doc_result["full_text"]) > 500 else doc_result["full_text"]
        }
        
    except HTTPException:
        raise
    except ValueError as val_err:
        raise HTTPException(status_code=422, detail=str(val_err))
    except Exception as exc:
        raise HTTPException(status_code=500, detail="Internal document processing error.")
    finally:
        await file.close()

import uuid
from typing import Dict, Any
from app.extractors import extract_document
from app.schemas import AIReadyDocument, DocumentMetadata

class DocumentIngestionService:
    @staticmethod
    def process_file(filename: str, file_bytes: bytes, tenant_id: str) -> AIReadyDocument:
        """
        Orchestrates document extraction, status determination, and model hydration.
        """
        result = extract_document(filename, file_bytes)
        full_text = result["full_text"]
        
        status = "extracted" if full_text.strip() else "needs_ocr_or_review"
        
        metadata = DocumentMetadata(
            original_filename=filename,
            file_type=result["file_type"],
            content_hash=result["content_hash"],
            page_count=result["metadata"]["page_count"],
            unit_count=result["metadata"]["unit_count"],
            character_count=result["metadata"]["character_count"]
        )
        
        return AIReadyDocument(
            document_id=f"doc_{uuid.uuid4().hex[:12]}",
            tenant_id=tenant_id,
            status=status,
            metadata=metadata,
            units=result["units"],
            full_text=full_text
        )

import pytest
from app.normalization import normalize_text
from app.extractors import extract_document
from io import BytesIO
from docx import Document

def test_unicode_nfc_and_whitespace_normalization():
    raw = "  Employee    Leave Policy \r\n\r\n\r\n\r\nAnnual Leave:  20 days  "
    clean = normalize_text(raw)
    assert clean == "Employee Leave Policy\n\nAnnual Leave: 20 days"
    assert "20" in clean  # Ensure numbers are never stripped

def test_docx_table_and_paragraph_structure():
    doc = Document()
    doc.add_heading("Leave Policy", level=1)
    doc.add_paragraph("Employees get 20 days.")
    table = doc.add_table(rows=2, cols=2)
    table.cell(0, 0).text = "Type"
    table.cell(0, 1).text = "Days"
    table.cell(1, 0).text = "Annual"
    table.cell(1, 1).text = "20"
    
    buf = BytesIO()
    doc.save(buf)
    file_bytes = buf.getvalue()
    
    result = extract_document("test.docx", file_bytes)
    assert result["file_type"] == "docx"
    assert len(result["units"]) == 3  # 1 heading, 1 paragraph, 1 table
    assert result["units"][2]["unit_type"] == "table"
    assert result["units"][2]["rows"][0] == ["Type", "Days"]

def test_unsupported_file_extension():
    with pytest.raises(ValueError, match="Unsupported file format"):
        extract_document("malicious.exe", b"fake binary data")