compliance_plugin.py

"""
Compliance Plugin - Production Example

Demonstrates production plugin architecture with:
- Dependency injection (policy database, email service)
- Input validation with Pydantic
- Error handling and metrics
- Multiple kernel functions
- Cross-plugin composition

KB Reference: Use Cases → Financial Services Compliance
"""

from typing import Dict, Optional, List
from pydantic import BaseModel, Field, validator
from semantic_kernel.functions import kernel_function
from .base_plugin import BaseProductionPlugin, PluginConfig
import asyncio


class PolicyDatabase:
    """Simulated policy database (in production, use real DB)"""

    def __init__(self):
        self.policies = {
            "refund": {
                "title": "Refund Policy",
                "content": "Full refunds within 30 days of purchase",
                "approval_required": False,
                "version": "2.0"
            },
            "expense": {
                "title": "Expense Approval Policy",
                "content": "Expenses under $500: auto-approved. $500-$5000: manager approval. Over $5000: director approval",
                "approval_required": True,
                "version": "1.5"
            },
            "data_access": {
                "title": "Data Access Policy",
                "content": "Employee data access requires manager approval and audit logging",
                "approval_required": True,
                "version": "3.0"
            }
        }

    async def get_policy(self, policy_type: str) -> Optional[Dict]:
        """Get policy by type"""
        # Simulate async database call
        await asyncio.sleep(0.01)
        return self.policies.get(policy_type.lower())

    async def search_policies(self, query: str) -> List[Dict]:
        """Search policies by keyword"""
        await asyncio.sleep(0.01)
        results = []
        query_lower = query.lower()

        for policy_type, policy in self.policies.items():
            if query_lower in policy["content"].lower() or query_lower in policy["title"].lower():
                results.append({
                    "type": policy_type,
                    **policy
                })

        return results


class EmailService:
    """Simulated email service (in production, use SendGrid, Azure Communication Services)"""

    async def send_email(
        self,
        to: str,
        subject: str,
        body: str,
        cc: Optional[List[str]] = None
    ) -> bool:
        """Send email notification"""
        # Simulate email sending
        await asyncio.sleep(0.01)
        print(f"📧 Email sent to {to}: {subject}")
        return True


class ComplianceQueryInput(BaseModel):
    """Input validation for compliance queries"""
    query: str = Field(..., min_length=5, max_length=500)
    user_id: str = Field(..., min_length=3)
    user_clearance: str = Field(default="standard")

    @validator('user_clearance')
    def validate_clearance(cls, v):
        allowed = ["standard", "confidential", "secret"]
        if v not in allowed:
            raise ValueError(f"Clearance must be one of {allowed}")
        return v


class ApprovalRequest(BaseModel):
    """Input validation for approval requests"""
    request_type: str = Field(..., description="Type of request (refund, expense, data_access)")
    amount: Optional[float] = Field(None, ge=0, description="Amount for expense requests")
    user_id: str = Field(..., min_length=3)
    justification: str = Field(..., min_length=10, max_length=1000)


class CompliancePluginConfig(PluginConfig):
    """Extended configuration for compliance plugin"""
    max_query_length: int = Field(default=500)
    email_notifications: bool = Field(default=True)
    audit_logging: bool = Field(default=True)


class CompliancePlugin(BaseProductionPlugin):
    """
    Production compliance plugin for financial services.

    Features:
    - Policy lookup and search
    - Approval workflow automation
    - Email notifications
    - Audit logging
    - Input validation

    Dependencies:
    - PolicyDatabase: Policy storage and retrieval
    - EmailService: Notification delivery

    Example:
        config = CompliancePluginConfig(plugin_name="Compliance")
        policy_db = PolicyDatabase()
        email_svc = EmailService()

        plugin = CompliancePlugin(
            config=config,
            policy_database=policy_db,
            email_service=email_svc
        )

        result = await plugin.check_policy(query="What is the refund policy?")
    """

    def __init__(
        self,
        config: CompliancePluginConfig,
        policy_database: PolicyDatabase,
        email_service: EmailService
    ):
        super().__init__(config)
        self.policy_db = policy_database
        self.email_service = email_service

    async def _initialize_impl(self):
        """Initialize compliance plugin"""
        self.logger.info("Loading compliance policies...")
        # In production: load policies from database, validate connections
        await asyncio.sleep(0.1)  # Simulate loading
        self.logger.info("Compliance policies loaded")

    @kernel_function(
        name="check_policy",
        description="Check company policy for a specific action or question"
    )
    async def check_policy(self, query: str, user_id: str = "system") -> str:
        """
        Check compliance policy.

        Args:
            query: Policy question or action to check
            user_id: User making the request

        Returns:
            Policy information or guidance
        """
        def _execute():
            # Validate input
            input_data = ComplianceQueryInput(
                query=query,
                user_id=user_id
            )

            # Log query for audit
            if self.config.audit_logging:
                self.logger.info(f"Policy query from {user_id}: {query}")

            # Search for relevant policies
            loop = asyncio.get_event_loop()
            policies = loop.run_until_complete(
                self.policy_db.search_policies(query)
            )

            if not policies:
                return "❌ No matching policy found. Please contact your compliance officer."

            # Format response
            if len(policies) == 1:
                policy = policies[0]
                return f"✓ Policy found: {policy['title']}\n\n{policy['content']}\n\n(Version {policy['version']})"
            else:
                result = f"Found {len(policies)} relevant policies:\n\n"
                for i, policy in enumerate(policies, 1):
                    result += f"{i}. {policy['title']}\n   {policy['content']}\n\n"
                return result

        return self.execute_with_metrics("check_policy", _execute)

    @kernel_function(
        name="get_approval_requirements",
        description="Get approval requirements for a request type and amount"
    )
    def get_approval_requirements(
        self,
        request_type: str,
        amount: str = "0"
    ) -> str:
        """
        Get approval requirements.

        Args:
            request_type: Type of request (refund, expense, data_access)
            amount: Amount (for expense requests)

        Returns:
            Approval requirements
        """
        def _execute():
            # Parse amount
            try:
                amount_value = float(amount.replace("$", "").replace(",", ""))
            except ValueError:
                amount_value = 0.0

            # Determine approval chain
            if request_type.lower() == "expense":
                if amount_value < 500:
                    return "✓ Auto-approved (under $500)\nNo approval required."
                elif amount_value < 5000:
                    return "⚠️ Manager approval required\nApprox. 1-2 business days"
                else:
                    return "⚠️ Director approval required\nManager → Director chain\nApprox. 3-5 business days"

            elif request_type.lower() == "refund":
                return "✓ Auto-approved\nRefunds processed within 5-7 business days"

            elif request_type.lower() == "data_access":
                return "⚠️ Manager approval + Audit logging required\nApprox. 2-3 business days"

            else:
                return f"❌ Unknown request type: {request_type}"

        return self.execute_with_metrics("get_approval_requirements", _execute)

    @kernel_function(
        name="submit_approval_request",
        description="Submit an approval request with automatic routing"
    )
    async def submit_approval_request(
        self,
        request_type: str,
        amount: str = "0",
        user_id: str = "user_001",
        justification: str = "No justification provided"
    ) -> str:
        """
        Submit approval request.

        Args:
            request_type: Type of request
            amount: Amount (if applicable)
            user_id: Requesting user ID
            justification: Request justification

        Returns:
            Submission status and next steps
        """
        def _execute():
            # Validate input
            try:
                amount_value = float(amount.replace("$", "").replace(",", ""))
            except ValueError:
                amount_value = 0.0

            request = ApprovalRequest(
                request_type=request_type,
                amount=amount_value,
                user_id=user_id,
                justification=justification
            )

            # Log for audit
            self.logger.info(
                f"Approval request: {request_type} - ${amount_value} - User: {user_id}"
            )

            # Determine routing
            if request_type.lower() == "expense" and amount_value < 500:
                approval_status = "APPROVED"
                next_steps = "Request auto-approved. Processing immediately."
            else:
                approval_status = "PENDING"
                next_steps = "Request submitted for approval. You will receive email notification."

                # Send email notification
                if self.config.email_notifications:
                    loop = asyncio.get_event_loop()
                    loop.run_until_complete(
                        self.email_service.send_email(
                            to=f"{user_id}@company.com",
                            subject=f"Approval Request Submitted: {request_type}",
                            body=f"Your {request_type} request has been submitted for approval.\n\n"
                                 f"Amount: ${amount_value}\n"
                                 f"Justification: {justification}\n\n"
                                 f"You will be notified once approved."
                        )
                    )

            return f"""
📋 Approval Request Submitted

Request ID: APR-{user_id}-{hash(justification) % 10000:04d}
Type: {request_type}
Amount: ${amount_value:.2f}
Status: {approval_status}

{next_steps}
"""

        return self.execute_with_metrics("submit_approval_request", _execute)

    @kernel_function(
        name="validate_and_notify",
        description="Validate request against policy and send notification (composite function)"
    )
    async def validate_and_notify(
        self,
        request_type: str,
        amount: str = "0",
        user_id: str = "user_001"
    ) -> str:
        """
        Composite function: Validate + Submit + Notify

        Demonstrates plugin function composition.
        """
        def _execute():
            # Step 1: Get requirements
            requirements = self.get_approval_requirements(request_type, amount)

            # Step 2: Submit request
            loop = asyncio.get_event_loop()
            submission = loop.run_until_complete(
                self.submit_approval_request(
                    request_type=request_type,
                    amount=amount,
                    user_id=user_id,
                    justification="Auto-generated request"
                )
            )

            return f"""
{requirements}

---

{submission}
"""

        return self.execute_with_metrics("validate_and_notify", _execute)