📊 Business Process Automation: Transform Enterprise Workflows

Business Process Automation (BPA) revolutionizes how organizations operate by transforming manual, paper-based processes into efficient digital workflows that reduce costs, improve accuracy, and accelerate delivery times. Like upgrading from horse-drawn carriages to modern vehicles, BPA modernizes business operations through intelligent automation that connects people, systems, and data seamlessly. Whether you're automating procurement, HR processes, financial operations, or customer service, mastering BPA is essential for digital transformation. Let's explore the comprehensive world of enterprise business process automation! 🏢

The BPA Architecture

Think of Business Process Automation as creating a digital nervous system for your organization - it senses events, routes information, triggers actions, and orchestrates complex workflows across departments and systems. Using workflow engines, business rules, integration platforms, and analytics, BPA transforms isolated manual tasks into connected, intelligent processes. Understanding process modeling, workflow orchestration, and performance optimization is crucial for successful BPA implementation!

Real-World Scenario: The Enterprise Process Transformation 🚀

You're leading a comprehensive BPA initiative for a global corporation that needs to automate employee onboarding across 50 locations, streamline invoice processing handling 10,000+ invoices monthly, automate customer order fulfillment from request to delivery, manage regulatory compliance reporting across jurisdictions, coordinate multi-department approval workflows, integrate with 20+ enterprise systems (ERP, CRM, HRM), provide real-time process visibility to stakeholders, and ensure scalability for business growth. Your solution must handle complex business rules, maintain audit trails, provide exception handling, and deliver measurable ROI. Let's build a comprehensive BPA framework!

# Comprehensive Business Process Automation Framework
# pip install celery redis pandas numpy
# pip install sqlalchemy psycopg2-binary alembic
# pip install pydantic fastapi python-multipart
# pip install camunda-external-task-client-python3
# pip install office365-rest-python-client exchangelib

import os
import json
import asyncio
from typing import Dict, List, Any, Optional, Union, Callable, Type
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from pathlib import Path
from enum import Enum, auto
from abc import ABC, abstractmethod
import uuid
import logging

# Workflow and orchestration
from celery import Celery, chain, group, chord
from celery.result import AsyncResult

# Database and ORM
from sqlalchemy import create_engine, Column, String, Integer, DateTime, JSON, Boolean, ForeignKey, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, relationship

# Data validation
from pydantic import BaseModel, Field, validator

# Business rules engine
import re
from typing import Pattern

# Document processing
import pandas as pd
from datetime import datetime
import hashlib

# Email integration
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

# ==================== Core BPA Models ====================

Base = declarative_base()

class ProcessState(str, Enum):
    """Process execution states."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    ON_HOLD = "on_hold"

class ProcessPriority(int, Enum):
    """Process priority levels."""
    LOW = 1
    NORMAL = 5
    HIGH = 8
    CRITICAL = 10

@dataclass
class ProcessDefinition:
    """Definition of a business process."""
    id: str
    name: str
    description: str
    version: str
    category: str
    owner: str
    steps: List['ProcessStep']
    rules: List['BusinessRule']
    sla_hours: Optional[int] = None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary."""
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'version': self.version,
            'category': self.category,
            'owner': self.owner,
            'steps': [step.to_dict() for step in self.steps],
            'rules': [rule.to_dict() for rule in self.rules],
            'sla_hours': self.sla_hours
        }

@dataclass
class ProcessStep:
    """Individual step in a process."""
    id: str
    name: str
    type: str  # manual, automated, decision, parallel
    handler: Optional[str] = None
    assignee: Optional[str] = None
    timeout_minutes: Optional[int] = None
    retry_count: int = 3
    next_steps: List[str] = field(default_factory=list)
    conditions: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary."""
        return asdict(self)

# ==================== Database Models ====================

class ProcessInstance(Base):
    """Process instance in database."""
    __tablename__ = 'process_instances'
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    process_definition_id = Column(String, nullable=False)
    state = Column(String, default=ProcessState.PENDING.value)
    priority = Column(Integer, default=ProcessPriority.NORMAL.value)
    
    started_at = Column(DateTime, default=datetime.utcnow)
    completed_at = Column(DateTime, nullable=True)
    
    initiator = Column(String, nullable=False)
    current_step = Column(String, nullable=True)
    
    context = Column(JSON, default={})
    error = Column(Text, nullable=True)
    
    # Relationships
    tasks = relationship("Task", back_populates="process_instance")
    audit_logs = relationship("AuditLog", back_populates="process_instance")

class Task(Base):
    """Individual task in a process."""
    __tablename__ = 'tasks'
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    process_instance_id = Column(String, ForeignKey('process_instances.id'))
    
    step_id = Column(String, nullable=False)
    name = Column(String, nullable=False)
    type = Column(String, nullable=False)
    
    state = Column(String, default=ProcessState.PENDING.value)
    assignee = Column(String, nullable=True)
    
    created_at = Column(DateTime, default=datetime.utcnow)
    started_at = Column(DateTime, nullable=True)
    completed_at = Column(DateTime, nullable=True)
    
    input_data = Column(JSON, default={})
    output_data = Column(JSON, default={})
    
    retry_count = Column(Integer, default=0)
    error = Column(Text, nullable=True)
    
    # Relationships
    process_instance = relationship("ProcessInstance", back_populates="tasks")

class AuditLog(Base):
    """Audit log for process activities."""
    __tablename__ = 'audit_logs'
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    process_instance_id = Column(String, ForeignKey('process_instances.id'))
    
    timestamp = Column(DateTime, default=datetime.utcnow)
    action = Column(String, nullable=False)
    user = Column(String, nullable=False)
    
    details = Column(JSON, default={})
    
    # Relationships
    process_instance = relationship("ProcessInstance", back_populates="audit_logs")

# ==================== Business Rules Engine ====================

@dataclass
class BusinessRule:
    """Business rule definition."""
    id: str
    name: str
    description: str
    condition: str  # Python expression
    action: str  # Action to take if condition is true
    priority: int = 5
    
    def evaluate(self, context: Dict[str, Any]) -> bool:
        """Evaluate rule condition."""
        try:
            return eval(self.condition, {"__builtins__": {}}, context)
        except Exception as e:
            logging.error(f"Error evaluating rule {self.name}: {e}")
            return False
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary."""
        return asdict(self)

class RulesEngine:
    """Business rules execution engine."""
    
    def __init__(self):
        self.rules: List[BusinessRule] = []
        self.logger = logging.getLogger(__name__)
    
    def add_rule(self, rule: BusinessRule):
        """Add rule to engine."""
        self.rules.append(rule)
        self.rules.sort(key=lambda r: r.priority, reverse=True)
    
    def evaluate_rules(self, context: Dict[str, Any]) -> List[str]:
        """Evaluate all rules and return actions."""
        actions = []
        
        for rule in self.rules:
            if rule.evaluate(context):
                self.logger.info(f"Rule '{rule.name}' matched")
                actions.append(rule.action)
        
        return actions
    
    def check_condition(self, condition: str, context: Dict[str, Any]) -> bool:
        """Check a single condition."""
        try:
            return eval(condition, {"__builtins__": {}}, context)
        except Exception as e:
            self.logger.error(f"Error checking condition: {e}")
            return False

# ==================== Workflow Engine ====================

class WorkflowEngine:
    """Core workflow execution engine."""
    
    def __init__(self, db_url: str = "sqlite:///bpa.db"):
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        self.SessionLocal = sessionmaker(bind=self.engine)
        self.rules_engine = RulesEngine()
        self.handlers = {}
        self.logger = logging.getLogger(__name__)
    
    def register_handler(self, step_type: str, handler: Callable):
        """Register step handler."""
        self.handlers[step_type] = handler
        self.logger.info(f"Registered handler for {step_type}")
    
    def start_process(
        self,
        process_def: ProcessDefinition,
        initiator: str,
        initial_context: Dict[str, Any] = None
    ) -> str:
        """Start a new process instance."""
        session = self.SessionLocal()
        
        try:
            # Create process instance
            instance = ProcessInstance(
                process_definition_id=process_def.id,
                initiator=initiator,
                context=initial_context or {},
                current_step=process_def.steps[0].id if process_def.steps else None
            )
            
            session.add(instance)
            session.commit()
            
            instance_id = instance.id
            
            # Log process start
            self._audit_log(
                session, instance_id, "PROCESS_STARTED",
                initiator, {"process": process_def.name}
            )
            
            # Execute first step
            if process_def.steps:
                self._execute_step(instance_id, process_def, process_def.steps[0])
            
            return instance_id
            
        except Exception as e:
            session.rollback()
            self.logger.error(f"Error starting process: {e}")
            raise
        finally:
            session.close()
    
    def _execute_step(
        self,
        instance_id: str,
        process_def: ProcessDefinition,
        step: ProcessStep
    ):
        """Execute a process step."""
        session = self.SessionLocal()
        
        try:
            # Get process instance
            instance = session.query(ProcessInstance).filter_by(id=instance_id).first()
            if not instance:
                raise ValueError(f"Process instance {instance_id} not found")
            
            # Update current step
            instance.current_step = step.id
            instance.state = ProcessState.RUNNING.value
            
            # Create task
            task = Task(
                process_instance_id=instance_id,
                step_id=step.id,
                name=step.name,
                type=step.type,
                assignee=step.assignee
            )
            
            session.add(task)
            session.commit()
            
            # Execute based on step type
            if step.type == "automated":
                self._execute_automated_step(task, step, instance)
            elif step.type == "manual":
                self._create_manual_task(task, step, instance)
            elif step.type == "decision":
                self._execute_decision_step(task, step, instance, process_def)
            elif step.type == "parallel":
                self._execute_parallel_steps(task, step, instance, process_def)
            
            session.commit()
            
        except Exception as e:
            session.rollback()
            self.logger.error(f"Error executing step {step.id}: {e}")
            instance.state = ProcessState.FAILED.value
            instance.error = str(e)
            session.commit()
        finally:
            session.close()
    
    def _execute_automated_step(
        self,
        task: Task,
        step: ProcessStep,
        instance: ProcessInstance
    ):
        """Execute automated step."""
        handler = self.handlers.get(step.handler)
        
        if not handler:
            raise ValueError(f"No handler registered for {step.handler}")
        
        try:
            # Execute handler
            task.state = ProcessState.RUNNING.value
            task.started_at = datetime.utcnow()
            
            result = handler(instance.context, task.input_data)
            
            task.output_data = result
            task.state = ProcessState.COMPLETED.value
            task.completed_at = datetime.utcnow()
            
            # Update context
            if isinstance(result, dict):
                instance.context.update(result)
            
        except Exception as e:
            task.state = ProcessState.FAILED.value
            task.error = str(e)
            task.retry_count += 1
            
            if task.retry_count < step.retry_count:
                # Retry
                task.state = ProcessState.PENDING.value
                self._schedule_retry(task, step)
            else:
                raise
    
    def _create_manual_task(
        self,
        task: Task,
        step: ProcessStep,
        instance: ProcessInstance
    ):
        """Create manual task for user."""
        task.state = ProcessState.PENDING.value
        
        # Send notification to assignee
        if task.assignee:
            self._send_task_notification(task, step, instance)
    
    def _execute_decision_step(
        self,
        task: Task,
        step: ProcessStep,
        instance: ProcessInstance,
        process_def: ProcessDefinition
    ):
        """Execute decision step based on conditions."""
        task.state = ProcessState.RUNNING.value
        task.started_at = datetime.utcnow()
        
        # Evaluate conditions
        for next_step_id, condition in step.conditions.items():
            if self.rules_engine.check_condition(condition, instance.context):
                # Find next step
                next_step = self._find_step(process_def, next_step_id)
                if next_step:
                    self._execute_step(instance.id, process_def, next_step)
                    break
        
        task.state = ProcessState.COMPLETED.value
        task.completed_at = datetime.utcnow()
    
    def _execute_parallel_steps(
        self,
        task: Task,
        step: ProcessStep,
        instance: ProcessInstance,
        process_def: ProcessDefinition
    ):
        """Execute parallel steps."""
        # Create tasks for parallel execution
        parallel_tasks = []
        
        for next_step_id in step.next_steps:
            next_step = self._find_step(process_def, next_step_id)
            if next_step:
                # Use Celery for parallel execution
                parallel_tasks.append(
                    execute_step_async.delay(instance.id, process_def.to_dict(), next_step.to_dict())
                )
        
        # Wait for all parallel tasks to complete
        # This would be handled by Celery chord in production
    
    def _find_step(self, process_def: ProcessDefinition, step_id: str) -> Optional[ProcessStep]:
        """Find step by ID."""
        for step in process_def.steps:
            if step.id == step_id:
                return step
        return None
    
    def _schedule_retry(self, task: Task, step: ProcessStep):
        """Schedule task retry."""
        # Use Celery for scheduling
        retry_task.apply_async(
            args=[task.id, step.to_dict()],
            countdown=60 * (2 ** task.retry_count)  # Exponential backoff
        )
    
    def _send_task_notification(self, task: Task, step: ProcessStep, instance: ProcessInstance):
        """Send notification for manual task."""
        # Implementation would send actual email/notification
        self.logger.info(f"Notification sent to {task.assignee} for task {task.id}")
    
    def _audit_log(
        self,
        session: Session,
        instance_id: str,
        action: str,
        user: str,
        details: Dict[str, Any]
    ):
        """Create audit log entry."""
        log = AuditLog(
            process_instance_id=instance_id,
            action=action,
            user=user,
            details=details
        )
        session.add(log)

# ==================== Process Handlers ====================

class ProcessHandlers:
    """Collection of process step handlers."""
    
    @staticmethod
    def validate_invoice(context: Dict, input_data: Dict) -> Dict:
        """Validate invoice data."""
        required_fields = ['invoice_number', 'amount', 'vendor', 'date']
        
        for field in required_fields:
            if field not in input_data:
                raise ValueError(f"Missing required field: {field}")
        
        # Validate amount
        if input_data['amount'] <= 0:
            raise ValueError("Invoice amount must be positive")
        
        # Validate date format
        try:
            datetime.strptime(input_data['date'], '%Y-%m-%d')
        except ValueError:
            raise ValueError("Invalid date format. Use YYYY-MM-DD")
        
        return {
            'invoice_validated': True,
            'validation_timestamp': datetime.utcnow().isoformat()
        }
    
    @staticmethod
    def check_approval_limit(context: Dict, input_data: Dict) -> Dict:
        """Check if approval is needed based on amount."""
        amount = context.get('amount', 0)
        
        approval_required = False
        approver = None
        
        if amount > 10000:
            approval_required = True
            approver = 'cfo@company.com'
        elif amount > 5000:
            approval_required = True
            approver = 'finance_manager@company.com'
        elif amount > 1000:
            approval_required = True
            approver = 'supervisor@company.com'
        
        return {
            'approval_required': approval_required,
            'approver': approver,
            'approval_limit_checked': True
        }
    
    @staticmethod
    def send_to_erp(context: Dict, input_data: Dict) -> Dict:
        """Send data to ERP system."""
        # Simulate ERP integration
        erp_id = hashlib.md5(
            json.dumps(input_data, sort_keys=True).encode()
        ).hexdigest()[:8]
        
        return {
            'erp_id': erp_id,
            'erp_sync_timestamp': datetime.utcnow().isoformat(),
            'erp_status': 'synced'
        }
    
    @staticmethod
    def generate_report(context: Dict, input_data: Dict) -> Dict:
        """Generate report from data."""
        # Create DataFrame from context data
        df = pd.DataFrame([context])
        
        # Generate report (simplified)
        report_path = f"reports/report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        Path("reports").mkdir(exist_ok=True)
        df.to_csv(report_path, index=False)
        
        return {
            'report_generated': True,
            'report_path': report_path,
            'report_rows': len(df)
        }

# ==================== Celery Tasks ====================

celery_app = Celery('bpa_tasks', broker='redis://localhost:6379/0')

@celery_app.task
def execute_step_async(instance_id: str, process_def_dict: Dict, step_dict: Dict):
    """Execute process step asynchronously."""
    engine = WorkflowEngine()
    
    # Reconstruct objects
    process_def = ProcessDefinition(**process_def_dict)
    step = ProcessStep(**step_dict)
    
    engine._execute_step(instance_id, process_def, step)

@celery_app.task
def retry_task(task_id: str, step_dict: Dict):
    """Retry failed task."""
    engine = WorkflowEngine()
    session = engine.SessionLocal()
    
    try:
        task = session.query(Task).filter_by(id=task_id).first()
        if task:
            step = ProcessStep(**step_dict)
            instance = task.process_instance
            engine._execute_automated_step(task, step, instance)
            session.commit()
    finally:
        session.close()

# ==================== Common Business Processes ====================

class CommonProcesses:
    """Library of common business processes."""
    
    @staticmethod
    def create_invoice_approval_process() -> ProcessDefinition:
        """Create invoice approval process."""
        return ProcessDefinition(
            id="invoice_approval_v1",
            name="Invoice Approval Process",
            description="Automated invoice approval workflow",
            version="1.0",
            category="Finance",
            owner="finance@company.com",
            sla_hours=24,
            steps=[
                ProcessStep(
                    id="validate",
                    name="Validate Invoice",
                    type="automated",
                    handler="validate_invoice",
                    retry_count=3
                ),
                ProcessStep(
                    id="check_limit",
                    name="Check Approval Limit",
                    type="automated",
                    handler="check_approval_limit",
                    conditions={
                        "approve": "approval_required == True",
                        "process": "approval_required == False"
                    }
                ),
                ProcessStep(
                    id="approve",
                    name="Approve Invoice",
                    type="manual",
                    assignee="approver",
                    timeout_minutes=1440  # 24 hours
                ),
                ProcessStep(
                    id="process",
                    name="Process Payment",
                    type="automated",
                    handler="send_to_erp"
                ),
                ProcessStep(
                    id="notify",
                    name="Send Notification",
                    type="automated",
                    handler="send_notification"
                )
            ],
            rules=[
                BusinessRule(
                    id="urgent_processing",
                    name="Urgent Invoice Processing",
                    description="Fast-track urgent invoices",
                    condition="context.get('urgent', False) == True",
                    action="set_priority:HIGH",
                    priority=10
                ),
                BusinessRule(
                    id="vendor_blacklist",
                    name="Blacklisted Vendor Check",
                    description="Block blacklisted vendors",
                    condition="context.get('vendor') in ['BadVendor1', 'BadVendor2']",
                    action="reject_invoice",
                    priority=15
                )
            ]
        )
    
    @staticmethod
    def create_employee_onboarding_process() -> ProcessDefinition:
        """Create employee onboarding process."""
        return ProcessDefinition(
            id="employee_onboarding_v1",
            name="Employee Onboarding",
            description="New employee onboarding workflow",
            version="1.0",
            category="HR",
            owner="hr@company.com",
            sla_hours=48,
            steps=[
                ProcessStep(
                    id="create_accounts",
                    name="Create User Accounts",
                    type="parallel",
                    next_steps=["create_email", "create_ad", "create_apps"]
                ),
                ProcessStep(
                    id="create_email",
                    name="Create Email Account",
                    type="automated",
                    handler="create_email_account"
                ),
                ProcessStep(
                    id="create_ad",
                    name="Create AD Account",
                    type="automated",
                    handler="create_ad_account"
                ),
                ProcessStep(
                    id="create_apps",
                    name="Create Application Accounts",
                    type="automated",
                    handler="create_app_accounts"
                ),
                ProcessStep(
                    id="assign_equipment",
                    name="Assign Equipment",
                    type="manual",
                    assignee="it_support@company.com",
                    timeout_minutes=2880  # 48 hours
                ),
                ProcessStep(
                    id="schedule_training",
                    name="Schedule Training",
                    type="automated",
                    handler="schedule_training"
                ),
                ProcessStep(
                    id="send_welcome",
                    name="Send Welcome Package",
                    type="automated",
                    handler="send_welcome_email"
                )
            ],
            rules=[]
        )
    
    @staticmethod
    def create_purchase_order_process() -> ProcessDefinition:
        """Create purchase order process."""
        return ProcessDefinition(
            id="purchase_order_v1",
            name="Purchase Order Process",
            description="Purchase order creation and approval",
            version="1.0",
            category="Procurement",
            owner="procurement@company.com",
            sla_hours=72,
            steps=[
                ProcessStep(
                    id="validate_request",
                    name="Validate Request",
                    type="automated",
                    handler="validate_po_request"
                ),
                ProcessStep(
                    id="check_budget",
                    name="Check Budget",
                    type="automated",
                    handler="check_budget_availability",
                    conditions={
                        "approve": "budget_available == True",
                        "reject": "budget_available == False"
                    }
                ),
                ProcessStep(
                    id="vendor_selection",
                    name="Select Vendor",
                    type="decision",
                    conditions={
                        "preferred_vendor": "context.get('preferred_vendor') is not None",
                        "rfq_process": "context.get('amount') > 50000",
                        "standard_vendor": "True"
                    }
                ),
                ProcessStep(
                    id="create_po",
                    name="Create Purchase Order",
                    type="automated",
                    handler="create_purchase_order"
                ),
                ProcessStep(
                    id="send_to_vendor",
                    name="Send to Vendor",
                    type="automated",
                    handler="send_po_to_vendor"
                )
            ],
            rules=[
                BusinessRule(
                    id="emergency_purchase",
                    name="Emergency Purchase",
                    description="Fast-track emergency purchases",
                    condition="context.get('emergency', False) == True",
                    action="expedite_processing",
                    priority=20
                )
            ]
        )

# ==================== Process Analytics ====================

class ProcessAnalytics:
    """Analytics for business processes."""
    
    def __init__(self, db_url: str = "sqlite:///bpa.db"):
        self.engine = create_engine(db_url)
        self.SessionLocal = sessionmaker(bind=self.engine)
    
    def get_process_metrics(self, process_def_id: str) -> Dict[str, Any]:
        """Get metrics for a process."""
        session = self.SessionLocal()
        
        try:
            instances = session.query(ProcessInstance).filter_by(
                process_definition_id=process_def_id
            ).all()
            
            if not instances:
                return {}
            
            completed = [i for i in instances if i.state == ProcessState.COMPLETED.value]
            failed = [i for i in instances if i.state == ProcessState.FAILED.value]
            
            # Calculate durations
            durations = []
            for instance in completed:
                if instance.completed_at and instance.started_at:
                    duration = (instance.completed_at - instance.started_at).total_seconds() / 3600
                    durations.append(duration)
            
            return {
                'total_instances': len(instances),
                'completed': len(completed),
                'failed': len(failed),
                'success_rate': len(completed) / len(instances) * 100 if instances else 0,
                'avg_duration_hours': sum(durations) / len(durations) if durations else 0,
                'min_duration_hours': min(durations) if durations else 0,
                'max_duration_hours': max(durations) if durations else 0
            }
            
        finally:
            session.close()
    
    def get_bottlenecks(self, process_def_id: str) -> List[Dict[str, Any]]:
        """Identify process bottlenecks."""
        session = self.SessionLocal()
        
        try:
            # Analyze task durations
            tasks = session.query(Task).join(ProcessInstance).filter(
                ProcessInstance.process_definition_id == process_def_id
            ).all()
            
            step_durations = {}
            for task in tasks:
                if task.completed_at and task.started_at:
                    duration = (task.completed_at - task.started_at).total_seconds() / 60
                    
                    if task.step_id not in step_durations:
                        step_durations[task.step_id] = []
                    
                    step_durations[task.step_id].append(duration)
            
            # Calculate averages
            bottlenecks = []
            for step_id, durations in step_durations.items():
                avg_duration = sum(durations) / len(durations)
                bottlenecks.append({
                    'step_id': step_id,
                    'avg_duration_minutes': avg_duration,
                    'task_count': len(durations)
                })
            
            # Sort by duration
            bottlenecks.sort(key=lambda x: x['avg_duration_minutes'], reverse=True)
            
            return bottlenecks[:5]  # Top 5 bottlenecks
            
        finally:
            session.close()

# Example usage
if __name__ == "__main__":
    print("📊 Business Process Automation Examples\n")
    
    # Example 1: Process categories
    print("1️⃣ Common Business Process Categories:")
    categories = [
        ("Finance", "Invoice processing, expense management, budgeting"),
        ("HR", "Onboarding, offboarding, leave management, performance"),
        ("Sales", "Lead management, quote-to-cash, order fulfillment"),
        ("Operations", "Inventory, supply chain, quality control"),
        ("IT", "Service requests, incident management, provisioning"),
        ("Customer Service", "Ticketing, complaints, feedback"),
        ("Compliance", "Regulatory reporting, auditing, risk management"),
        ("Marketing", "Campaign management, content approval, lead nurturing")
    ]
    for category, examples in categories:
        print(f"   {category}: {examples}")
    
    # Example 2: Process components
    print("\n2️⃣ BPA Components:")
    components = [
        "Workflow Engine - Orchestrates process execution",
        "Rules Engine - Evaluates business rules and conditions",
        "Task Manager - Manages manual and automated tasks",
        "Integration Layer - Connects to enterprise systems",
        "Monitoring - Tracks process performance",
        "Analytics - Provides insights and optimization"
    ]
    for component in components:
        print(f"   • {component}")
    
    # Example 3: Create sample process
    print("\n3️⃣ Sample Invoice Approval Process:")
    invoice_process = CommonProcesses.create_invoice_approval_process()
    print(f"   Process: {invoice_process.name}")
    print(f"   Steps: {len(invoice_process.steps)}")
    print(f"   Rules: {len(invoice_process.rules)}")
    print(f"   SLA: {invoice_process.sla_hours} hours")
    
    # Example 4: Initialize workflow engine
    print("\n4️⃣ Workflow Engine Initialization:")
    engine = WorkflowEngine()
    
    # Register handlers
    engine.register_handler('validate_invoice', ProcessHandlers.validate_invoice)
    engine.register_handler('check_approval_limit', ProcessHandlers.check_approval_limit)
    engine.register_handler('send_to_erp', ProcessHandlers.send_to_erp)
    
    print("   Registered handlers:")
    for handler_name in engine.handlers.keys():
        print(f"   • {handler_name}")
    
    # Example 5: Business rules
    print("\n5️⃣ Business Rules Examples:")
    
    rules_engine = RulesEngine()
    
    # Add sample rules
    rules = [
        BusinessRule(
            id="high_value",
            name="High Value Transaction",
            description="Flag high value transactions",
            condition="amount > 10000",
            action="require_cfo_approval",
            priority=10
        ),
        BusinessRule(
            id="urgent",
            name="Urgent Processing",
            description="Expedite urgent requests",
            condition="priority == 'urgent'",
            action="fast_track",
            priority=15
        )
    ]
    
    for rule in rules:
        rules_engine.add_rule(rule)
        print(f"   {rule.name}: {rule.condition} → {rule.action}")
    
    # Example 6: Process metrics
    print("\n6️⃣ Process Metrics:")
    metrics = [
        "Cycle Time - Total time from start to completion",
        "Throughput - Number of processes completed per time",
        "Success Rate - Percentage of successful completions",
        "SLA Compliance - Percentage meeting SLA",
        "Cost per Process - Average cost to complete",
        "Error Rate - Percentage of failed processes",
        "Automation Rate - Percentage of automated steps"
    ]
    for metric in metrics:
        print(f"   • {metric}")
    
    # Example 7: Integration points
    print("\n7️⃣ Common Integration Points:")
    integrations = [
        "ERP Systems (SAP, Oracle, Microsoft Dynamics)",
        "CRM Systems (Salesforce, HubSpot)",
        "HR Systems (Workday, ADP)",
        "Email Systems (Exchange, Gmail)",
        "Document Management (SharePoint, Box)",
        "Databases (SQL Server, Oracle, PostgreSQL)",
        "APIs (REST, SOAP, GraphQL)",
        "File Systems (FTP, SFTP, Network Shares)"
    ]
    for integration in integrations:
        print(f"   • {integration}")
    
    # Example 8: Optimization strategies
    print("\n8️⃣ Process Optimization Strategies:")
    strategies = [
        "Eliminate unnecessary steps",
        "Automate manual tasks",
        "Parallelize independent steps",
        "Implement intelligent routing",
        "Use predictive analytics",
        "Optimize approval chains",
        "Implement exception handling",
        "Cache frequently accessed data"
    ]
    for strategy in strategies:
        print(f"   • {strategy}")
    
    # Example 9: Best practices
    print("\n9️⃣ BPA Best Practices:")
    practices = [
        "🎯 Start with high-impact, high-volume processes",
        "📝 Document current process before automating",
        "🔄 Design for exception handling",
        "📊 Define clear metrics and KPIs",
        "👥 Involve stakeholders in design",
        "🧪 Test thoroughly with edge cases",
        "📈 Monitor and optimize continuously",
        "🔒 Implement proper security and compliance",
        "📚 Maintain documentation and training",
        "🚀 Plan for scalability"
    ]
    for practice in practices:
        print(f"   {practice}")
    
    # Example 10: ROI calculation
    print("\n🔟 BPA ROI Example:")
    
    # Sample ROI calculation
    manual_time = 30  # minutes per process
    processes_per_month = 1000
    labor_cost_per_hour = 50
    
    manual_cost = (manual_time / 60) * processes_per_month * labor_cost_per_hour * 12
    
    automated_time = 5  # minutes per process
    automation_cost = 50000  # Development cost
    maintenance_cost = 5000  # Annual
    
    automated_cost = automation_cost + maintenance_cost + \
                    ((automated_time / 60) * processes_per_month * labor_cost_per_hour * 12)
    
    savings = manual_cost - automated_cost
    roi = (savings / automation_cost) * 100
    
    print(f"   Manual Cost/Year: ${manual_cost:,.2f}")
    print(f"   Automated Cost/Year: ${automated_cost:,.2f}")
    print(f"   Annual Savings: ${savings:,.2f}")
    print(f"   ROI: {roi:.1f}%")
    print(f"   Payback Period: {automation_cost / (savings/12):.1f} months")
    
    print("\n✅ Business process automation demonstration complete!")

Key Takeaways and Best Practices 🎯

• Process Selection: Start with high-volume, repetitive processes for maximum ROI.
• Stakeholder Involvement: Engage process owners and users in design.
• Exception Handling: Design for edge cases and error scenarios.
• Integration: Ensure seamless connection with existing systems.
• Monitoring: Track KPIs and process performance continuously.
• Compliance: Maintain audit trails and ensure regulatory compliance.
• Scalability: Design processes to handle growing volumes.
• Continuous Improvement: Optimize based on analytics and feedback.

Business Process Automation Best Practices 📋

Mastering Business Process Automation enables you to transform manual, inefficient workflows into streamlined, digital processes that deliver consistent results. You can now design complex workflows, implement business rules engines, integrate with enterprise systems, and optimize processes based on analytics. Whether you're automating financial operations, HR processes, or customer workflows, these BPA skills are essential for digital transformation! 🚀