document-parsing

name	document-parsing
description	Parse PDF, DOCX, XLSX, and PPTX documents in a microservices context. Use when building the Media Processor service (Phase 9) for document text extraction. Covers BytesIO streaming, standardized JSON output, async wrappers for CPU-bound operations. NOT for desktop automation, file editing, or document creation.

Document Parsing for Microservices

Parse office documents in memory and return structured JSON for AI consumption.

Quick Start

from io import BytesIO
from document_parsing import parse_document, DocumentContent

# Parse from bytes (e.g., from S3 or HTTP upload)
content: DocumentContent = await parse_document(
    file_bytes=uploaded_file.read(),
    filename="report.pdf",
)

# Structured output ready for LLM
print(content.text)        # Full extracted text
print(content.metadata)    # Title, author, page count, etc.
print(content.tables)      # Extracted tables as list[list[str]]

---

Architecture: Media Processor Context

┌─────────────────┐     HTTP/arq      ┌─────────────────┐
│ Chat Orchestrator│ ────────────────► │ Media Processor │
│ (Phase 5)        │                   │ (Phase 9)       │
└─────────────────┘                   └────────┬────────┘
                                               │
                                    ┌──────────┴──────────┐
                                    │                     │
                                    ▼                     ▼
                             Parse Documents        Generate Images
                             (This Skill)           (Vertex AI)
                                    │
                                    ▼
                             ┌─────────────┐
                             │ S3 / MinIO  │
                             │ (File Store)│
                             └─────────────┘

The Media Processor receives files via:

1. HTTP Upload - Direct file upload in request body
2. S3 Reference - Key to fetch from object storage
3. arq Task - Async job with file bytes or S3 key

It returns structured JSON, never raw files.

---

Library Selection

Format	Library	Why
PDF	pdfplumber	Best table extraction, pure Python, BytesIO support
DOCX	python-docx	Official library, BytesIO support, preserves structure
XLSX	openpyxl	Feature-complete, BytesIO support, handles formulas
PPTX	python-pptx	Official library, BytesIO support, slide extraction

Dependencies:

# pyproject.toml
[project]
dependencies = [
    "pdfplumber>=0.10.0",
    "python-docx>=1.1.0",
    "openpyxl>=3.1.0",
    "python-pptx>=0.6.23",
]

---

Standardized Output Schema

All parsers return the same structure for consistent downstream processing.

# schemas/document.py
from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime

class TableData(BaseModel):
    """Extracted table with headers and rows."""
    headers: list[str] = Field(default_factory=list)
    rows: list[list[str]] = Field(default_factory=list)
    page: int | None = None  # Page number (PDF/PPTX)
    sheet: str | None = None  # Sheet name (XLSX)

class DocumentMetadata(BaseModel):
    """Document metadata extracted from file properties."""
    title: str | None = None
    author: str | None = None
    created: datetime | None = None
    modified: datetime | None = None
    page_count: int | None = None
    word_count: int | None = None

class DocumentContent(BaseModel):
    """Unified output from all document parsers."""
    format: Literal["pdf", "docx", "xlsx", "pptx"]
    filename: str
    text: str  # Full extracted text
    metadata: DocumentMetadata
    tables: list[TableData] = Field(default_factory=list)
    sections: list[str] = Field(default_factory=list)  # Headings/slide titles
    errors: list[str] = Field(default_factory=list)  # Non-fatal parse warnings

---

Parser Implementations

PDF Parser (pdfplumber)

# parsers/pdf.py
import pdfplumber
from io import BytesIO

from schemas.document import DocumentContent, DocumentMetadata, TableData

def parse_pdf(file_bytes: bytes, filename: str) -> DocumentContent:
    """Parse PDF from bytes, extract text and tables."""

    buffer = BytesIO(file_bytes)
    text_parts: list[str] = []
    tables: list[TableData] = []
    errors: list[str] = []

    with pdfplumber.open(buffer) as pdf:
        metadata = DocumentMetadata(
            title=pdf.metadata.get("Title"),
            author=pdf.metadata.get("Author"),
            page_count=len(pdf.pages),
        )

        for page_num, page in enumerate(pdf.pages, start=1):
            # Extract text
            try:
                page_text = page.extract_text() or ""
                text_parts.append(page_text)
            except Exception as e:
                errors.append(f"Page {page_num} text extraction failed: {e}")

            # Extract tables
            try:
                page_tables = page.extract_tables()
                for table in page_tables:
                    if table and len(table) > 1:
                        tables.append(TableData(
                            headers=table[0] if table[0] else [],
                            rows=table[1:],
                            page=page_num,
                        ))
            except Exception as e:
                errors.append(f"Page {page_num} table extraction failed: {e}")

    full_text = "\n\n".join(text_parts)

    return DocumentContent(
        format="pdf",
        filename=filename,
        text=full_text,
        metadata=metadata,
        tables=tables,
        errors=errors,
    )

DOCX Parser (python-docx)

# parsers/docx.py
from docx import Document
from docx.opc.exceptions import PackageNotFoundError
from io import BytesIO

from schemas.document import DocumentContent, DocumentMetadata, TableData

def parse_docx(file_bytes: bytes, filename: str) -> DocumentContent:
    """Parse DOCX from bytes, extract text, headings, and tables."""

    buffer = BytesIO(file_bytes)

    try:
        doc = Document(buffer)
    except PackageNotFoundError:
        return DocumentContent(
            format="docx",
            filename=filename,
            text="",
            metadata=DocumentMetadata(),
            errors=["Invalid or corrupted DOCX file"],
        )

    # Extract metadata from core properties
    props = doc.core_properties
    metadata = DocumentMetadata(
        title=props.title,
        author=props.author,
        created=props.created,
        modified=props.modified,
    )

    # Extract text and headings
    text_parts: list[str] = []
    sections: list[str] = []

    for para in doc.paragraphs:
        text_parts.append(para.text)

        # Track headings for document structure
        if para.style and para.style.name.startswith("Heading"):
            sections.append(para.text)

    # Extract tables
    tables: list[TableData] = []
    for table in doc.tables:
        rows = []
        for row in table.rows:
            cells = [cell.text.strip() for cell in row.cells]
            rows.append(cells)

        if rows:
            tables.append(TableData(
                headers=rows[0] if rows else [],
                rows=rows[1:] if len(rows) > 1 else [],
            ))

    full_text = "\n".join(text_parts)
    word_count = len(full_text.split())
    metadata.word_count = word_count

    return DocumentContent(
        format="docx",
        filename=filename,
        text=full_text,
        metadata=metadata,
        tables=tables,
        sections=sections,
    )

XLSX Parser (openpyxl)

# parsers/xlsx.py
from openpyxl import load_workbook
from openpyxl.utils.exceptions import InvalidFileException
from io import BytesIO

from schemas.document import DocumentContent, DocumentMetadata, TableData

def parse_xlsx(file_bytes: bytes, filename: str) -> DocumentContent:
    """Parse XLSX from bytes, extract sheet data as tables."""

    buffer = BytesIO(file_bytes)

    try:
        # data_only=True to get calculated values, not formulas
        wb = load_workbook(buffer, data_only=True, read_only=True)
    except InvalidFileException:
        return DocumentContent(
            format="xlsx",
            filename=filename,
            text="",
            metadata=DocumentMetadata(),
            errors=["Invalid or corrupted XLSX file"],
        )

    # Extract metadata
    props = wb.properties
    metadata = DocumentMetadata(
        title=props.title,
        author=props.creator,
        created=props.created,
        modified=props.modified,
    )

    # Extract each sheet as a table
    tables: list[TableData] = []
    text_parts: list[str] = []

    for sheet_name in wb.sheetnames:
        sheet = wb[sheet_name]
        rows: list[list[str]] = []

        for row in sheet.iter_rows(values_only=True):
            # Convert all values to strings, handle None
            str_row = [str(cell) if cell is not None else "" for cell in row]

            # Skip completely empty rows
            if any(cell.strip() for cell in str_row):
                rows.append(str_row)

        if rows:
            tables.append(TableData(
                headers=rows[0],
                rows=rows[1:],
                sheet=sheet_name,
            ))

            # Also add to text representation
            text_parts.append(f"=== Sheet: {sheet_name} ===")
            for row in rows[:50]:  # Limit text extraction
                text_parts.append("\t".join(row))

    wb.close()

    return DocumentContent(
        format="xlsx",
        filename=filename,
        text="\n".join(text_parts),
        metadata=metadata,
        tables=tables,
        sections=[name for name in wb.sheetnames],
    )

PPTX Parser (python-pptx)

# parsers/pptx.py
from pptx import Presentation
from pptx.util import Inches
from io import BytesIO

from schemas.document import DocumentContent, DocumentMetadata, TableData

def parse_pptx(file_bytes: bytes, filename: str) -> DocumentContent:
    """Parse PPTX from bytes, extract slide text and tables."""

    buffer = BytesIO(file_bytes)

    try:
        prs = Presentation(buffer)
    except Exception as e:
        return DocumentContent(
            format="pptx",
            filename=filename,
            text="",
            metadata=DocumentMetadata(),
            errors=[f"Invalid or corrupted PPTX file: {e}"],
        )

    # Extract metadata
    props = prs.core_properties
    metadata = DocumentMetadata(
        title=props.title,
        author=props.author,
        created=props.created,
        modified=props.modified,
        page_count=len(prs.slides),
    )

    text_parts: list[str] = []
    sections: list[str] = []  # Slide titles
    tables: list[TableData] = []

    for slide_num, slide in enumerate(prs.slides, start=1):
        slide_texts: list[str] = []
        slide_title = None

        for shape in slide.shapes:
            # Extract text from text frames
            if shape.has_text_frame:
                for para in shape.text_frame.paragraphs:
                    text = para.text.strip()
                    if text:
                        slide_texts.append(text)

                        # First text is usually the title
                        if slide_title is None:
                            slide_title = text

            # Extract tables
            if shape.has_table:
                table = shape.table
                rows: list[list[str]] = []

                for row in table.rows:
                    cells = [cell.text.strip() for cell in row.cells]
                    rows.append(cells)

                if rows:
                    tables.append(TableData(
                        headers=rows[0],
                        rows=rows[1:],
                        page=slide_num,
                    ))

        if slide_title:
            sections.append(slide_title)

        if slide_texts:
            text_parts.append(f"--- Slide {slide_num} ---")
            text_parts.extend(slide_texts)

    return DocumentContent(
        format="pptx",
        filename=filename,
        text="\n".join(text_parts),
        metadata=metadata,
        tables=tables,
        sections=sections,
    )

---

Async Wrappers for FastAPI

Document parsing is CPU-bound and blocks the event loop. Use run_in_executor or anyio.to_thread.

# service.py
import asyncio
from concurrent.futures import ProcessPoolExecutor
from functools import partial
from typing import Callable

from schemas.document import DocumentContent
from parsers.pdf import parse_pdf
from parsers.docx import parse_docx
from parsers.xlsx import parse_xlsx
from parsers.pptx import parse_pptx

# Process pool for CPU-bound parsing (survives across requests)
_executor = ProcessPoolExecutor(max_workers=4)

PARSERS: dict[str, Callable[[bytes, str], DocumentContent]] = {
    ".pdf": parse_pdf,
    ".docx": parse_docx,
    ".xlsx": parse_xlsx,
    ".pptx": parse_pptx,
}

def get_extension(filename: str) -> str:
    """Extract lowercase extension from filename."""
    return "." + filename.rsplit(".", 1)[-1].lower() if "." in filename else ""


async def parse_document(file_bytes: bytes, filename: str) -> DocumentContent:
    """
    Parse document asynchronously without blocking the event loop.

    Uses ProcessPoolExecutor to run CPU-bound parsing in separate process.
    """
    ext = get_extension(filename)

    parser = PARSERS.get(ext)
    if parser is None:
        return DocumentContent(
            format="pdf",  # Default
            filename=filename,
            text="",
            metadata={},
            errors=[f"Unsupported file format: {ext}"],
        )

    loop = asyncio.get_running_loop()

    # Run parser in process pool to avoid blocking
    result = await loop.run_in_executor(
        _executor,
        partial(parser, file_bytes, filename),
    )

    return result

---

S3/MinIO Integration

Fetch files from object storage before parsing.

# storage.py
import aioboto3
from contextlib import asynccontextmanager

from config import settings

@asynccontextmanager
async def get_s3_client():
    """Get async S3 client (works with MinIO too)."""
    session = aioboto3.Session()
    async with session.client(
        "s3",
        endpoint_url=settings.s3_endpoint,  # e.g., "http://minio:9000"
        aws_access_key_id=settings.s3_access_key,
        aws_secret_access_key=settings.s3_secret_key,
    ) as client:
        yield client


async def fetch_and_parse(bucket: str, key: str) -> DocumentContent:
    """Fetch file from S3 and parse it."""
    async with get_s3_client() as s3:
        response = await s3.get_object(Bucket=bucket, Key=key)
        file_bytes = await response["Body"].read()

    # Extract filename from key
    filename = key.rsplit("/", 1)[-1]

    return await parse_document(file_bytes, filename)

---

FastAPI Endpoints

# routes/documents.py
from fastapi import APIRouter, UploadFile, File, HTTPException
from pydantic import BaseModel

from schemas.document import DocumentContent
from service import parse_document
from storage import fetch_and_parse

router = APIRouter(prefix="/documents", tags=["documents"])


class S3Reference(BaseModel):
    bucket: str
    key: str


@router.post("/parse", response_model=DocumentContent)
async def parse_uploaded_document(file: UploadFile = File(...)):
    """Parse an uploaded document and return structured content."""

    if file.size and file.size > 50 * 1024 * 1024:  # 50MB limit
        raise HTTPException(400, "File too large (max 50MB)")

    file_bytes = await file.read()

    return await parse_document(file_bytes, file.filename or "unknown")


@router.post("/parse-s3", response_model=DocumentContent)
async def parse_s3_document(ref: S3Reference):
    """Parse a document from S3/MinIO and return structured content."""

    try:
        return await fetch_and_parse(ref.bucket, ref.key)
    except Exception as e:
        raise HTTPException(404, f"Failed to fetch from S3: {e}")

---

arq Task Worker

For async processing via task queue.

# tasks.py
from arq import func
from arq.connections import RedisSettings

from service import parse_document
from storage import fetch_and_parse


async def parse_document_task(ctx: dict, file_bytes: bytes, filename: str) -> dict:
    """arq task: Parse document from bytes."""
    result = await parse_document(file_bytes, filename)
    return result.model_dump()


async def parse_s3_task(ctx: dict, bucket: str, key: str) -> dict:
    """arq task: Parse document from S3."""
    result = await fetch_and_parse(bucket, key)
    return result.model_dump()


class WorkerSettings:
    functions = [
        func(parse_document_task, name="parse_document"),
        func(parse_s3_task, name="parse_s3"),
    ]
    redis_settings = RedisSettings.from_dsn("redis://redis:6379/0")
    max_jobs = 10

---

Testing

# tests/test_parsers.py
import pytest
from pathlib import Path

from service import parse_document

FIXTURES = Path(__file__).parent / "fixtures"


@pytest.mark.asyncio
async def test_parse_pdf():
    file_bytes = (FIXTURES / "sample.pdf").read_bytes()
    result = await parse_document(file_bytes, "sample.pdf")

    assert result.format == "pdf"
    assert len(result.text) > 0
    assert result.metadata.page_count is not None


@pytest.mark.asyncio
async def test_parse_docx():
    file_bytes = (FIXTURES / "sample.docx").read_bytes()
    result = await parse_document(file_bytes, "sample.docx")

    assert result.format == "docx"
    assert "Hello" in result.text  # Expected content


@pytest.mark.asyncio
async def test_parse_xlsx():
    file_bytes = (FIXTURES / "sample.xlsx").read_bytes()
    result = await parse_document(file_bytes, "sample.xlsx")

    assert result.format == "xlsx"
    assert len(result.tables) > 0


@pytest.mark.asyncio
async def test_parse_unsupported():
    result = await parse_document(b"not a real file", "file.xyz")

    assert "Unsupported" in result.errors[0]

---

Anti-Patterns

1. Using file paths - Always use BytesIO; containers don't have persistent filesystems
2. Blocking the event loop - Always use run_in_executor for parsing
3. Missing error handling - Corrupted files are common; always return partial results
4. Unbounded file sizes - Set limits to prevent OOM
5. Returning raw files - Always return structured JSON

---

Skill Relationships

Skill	Relationship
scaffolding-fastapi-sqlmodel	Media Processor service structure
redis-pattern	arq task queue for async parsing
adapter-interface	Could wrap parsers as adapters for swappability

---

Verification

Run: python scripts/verify.py

Expected: [OK] document-parsing skill ready

References

• pdfplumber docs
• python-docx docs
• openpyxl docs
• python-pptx docs