Classify and Extract Compliance Documents with Unstructured and spaCy

                      
                     
"""
Production implementation for classifying and extracting compliance documents using spaCy.
This architecture provides secure and scalable operations for document processing.
"""
from typing import Dict, Any, List
import os
import logging
import spacy
from spacy.tokens import Doc
from spacy.pipeline import EntityRuler

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class for environment variables.
    """
    nlp_model: str = os.getenv('SPACY_MODEL', 'en_core_web_sm')  # Load spaCy model
    database_url: str = os.getenv('DATABASE_URL')  # Database connection string

# Load spaCy model
nlp = spacy.load(Config.nlp_model)

def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data for document processing.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'documents' not in data or not isinstance(data['documents'], list):
        raise ValueError('Invalid input: documents must be a list')
    return True

def sanitize_fields(doc: str) -> str:
    """Sanitize document fields for processing.
    
    Args:
        doc: Raw document string
    Returns:
        Sanitized document string
    """
    return doc.strip().replace('\n', ' ').replace('\r', '')  # Strip whitespace and newlines

def create_entity_ruler(nlp: spacy.language.Language) -> EntityRuler:
    """Create an entity ruler for specific compliance keywords.
    
    Args:
        nlp: spaCy language model
    Returns:
        EntityRuler object
    """
    ruler = EntityRuler(nlp)
    patterns = [{'label': 'COMPLIANCE', 'pattern': 'GDPR'}, {'label': 'COMPLIANCE', 'pattern': 'HIPAA'}]
    ruler.add_patterns(patterns)  # Adding compliance patterns
    nlp.add_pipe(ruler)  # Add ruler to the pipeline
    return ruler

def process_documents(docs: List[str]) -> List[Dict[str, Any]]:
    """Process a list of documents and extract entities.
    
    Args:
        docs: List of document strings
    Returns:
        List of dictionaries with extracted data
    """
    results = []  # Store results
    ruler = create_entity_ruler(nlp)  # Initialize entity ruler
    for doc in docs:
        sanitized_doc = sanitize_fields(doc)  # Sanitize document
        spacy_doc = nlp(sanitized_doc)  # Process with spaCy
        entities = [(ent.text, ent.label_) for ent in spacy_doc.ents]  # Extract entities
        results.append({'text': sanitized_doc, 'entities': entities})  # Save results
    return results  # Return all extracted data

def save_to_db(data: List[Dict[str, Any]]) -> None:
    """Save processed data to the database.
    
    Args:
        data: Data to save
    Raises:
        Exception: If database operation fails
    """
    # Placeholder for database saving logic
    try:
        logger.info('Saving data to the database...')
        # Simulating a DB save operation
        # db.save(data)
        logger.info('Data saved successfully.')
    except Exception as e:
        logger.error(f'Error saving data to DB: {e}')
        raise  # Rethrow exception for upstream handling

def format_output(results: List[Dict[str, Any]]) -> None:
    """Format output for display or further processing.
    
    Args:
        results: Processed results to format
    """
    for result in results:
        logger.info(f'Document: {result['text']}, Entities: {result['entities']}')  # Log results

class ComplianceDocumentProcessor:
    """Orchestrator class for processing compliance documents.
    
    This class ties together the helper functions for a complete workflow.
    """
    def __init__(self, documents: List[str]):
        self.documents = documents

    def run(self) -> None:
        """Run the document processing workflow.
        """
        try:
            validate_input({'documents': self.documents})  # Validate input
            results = process_documents(self.documents)  # Process documents
            save_to_db(results)  # Save results to DB
            format_output(results)  # Format and display results
        except ValueError as ve:
            logger.error(f'Input validation error: {ve}')  # Log validation errors
        except Exception as e:
            logger.error(f'An error occurred during processing: {e}')  # Log other errors

if __name__ == '__main__':
    # Example usage
    sample_documents = [
        'This document is compliant with GDPR.',
        'This document follows HIPAA regulations.'
    ]
    processor = ComplianceDocumentProcessor(sample_documents)  # Create processor instance
    processor.run()  # Run the processing workflow