Parse Multi-Format Factory Documents into Search Indexes with MarkItDown and LlamaIndex

"""
Production implementation for parsing multi-format factory documents into search indexes.
Provides secure, scalable operations using MarkItDown and LlamaIndex.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import json
import requests
from contextlib import contextmanager
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# Setup logging for monitoring the application
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to manage environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL')

# Create a connection pool for database interactions
engine = create_engine(Config.database_url)
Session = sessionmaker(bind=engine)

@contextmanager
def get_db_session():
    """
    Context manager for database session management.
    
    Yields:
        Session: Database session object.
    """
    session = Session()
    try:
        yield session
    except Exception as e:
        logger.error(f"Database error: {e}")
        session.rollback()
    finally:
        session.close()

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'id' not in data:
        raise ValueError('Missing id')  # Raise error if 'id' is not present
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data
    """
    # Sanitize string fields to prevent injection attacks
    return {key: str(value).strip() for key, value in data.items()}

async def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform records to desired format.
    
    Args:
        records: Raw records to transform
    Returns:
        Transformed records
    """
    # Example transformation logic
    return [{'id': rec['id'], 'content': rec['content'].lower()} for rec in records]

async def process_batch(records: List[Dict[str, Any]]) -> None:
    """Process a batch of records.
    
    Args:
        records: List of records to process
    """
    for record in records:
        # Call to MarkItDown API to parse the content
        response = await call_api(record['content'])
        if response:
            await save_to_db(response)

async def call_api(content: str) -> Dict[str, Any]:
    """Call external API to parse content.
    
    Args:
        content: Content to parse
    Returns:
        Parsed data from the API
    Raises:
        RuntimeError: If API call fails
    """
    try:
        response = requests.post('https://api.markitdown.com/parse', json={'content': content})
        response.raise_for_status()  # Raise error for bad responses
        return response.json()
    except requests.HTTPError as e:
        logger.error(f"API call failed: {e}")
        raise RuntimeError("API call failed")

async def save_to_db(parsed_data: Dict[str, Any]) -> None:
    """Save parsed data to the database.
    
    Args:
        parsed_data: Data to save
    """
    with get_db_session() as session:
        # Example of saving data to the database
        session.execute(text("INSERT INTO documents (id, content) VALUES (:id, :content)"),
                       {'id': parsed_data['id'], 'content': json.dumps(parsed_data['content'])})
        session.commit()  # Commit the transaction

async def format_output(data: Any) -> str:
    """Format output data for presentation.
    
    Args:
        data: Data to format
    Returns:
        Formatted string output
    """
    return json.dumps(data, indent=2)  # Pretty print JSON data

async def handle_errors(func):
    """Decorator to handle errors in async functions.
    
    Args:
        func: Async function to wrap
    """
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            logger.error(f"Error occurred: {e}")
            return None  # Provide a fallback
    return wrapper

class DocumentParser:
    """Main orchestrator for document parsing.
    Handles the flow of data from input to output.
    """
    @handle_errors
    async def parse_documents(self, documents: List[Dict[str, Any]]) -> None:
        """Main parsing workflow.
        
        Args:
            documents: List of documents to parse
        """
        for doc in documents:
            await validate_input(doc)  # Validate each document
            sanitized_doc = await sanitize_fields(doc)
            transformed_data = await transform_records([sanitized_doc])
            await process_batch(transformed_data)
            logger.info(f"Processed document: {doc['id']}")  # Log processed document

if __name__ == '__main__':
    # Example usage
    parser = DocumentParser()
    documents = [{'id': '1', 'content': 'Sample content here.'},
                 {'id': '2', 'content': 'Another sample content.'}]
    await parser.parse_documents(documents)  # Run the parser