knowledge_base.py

"""
Knowledge Base Loader
Loads and validates knowledge_base.json as the single source of truth.
"""
import json
import os
from typing import Dict, List, Optional, Any
from pathlib import Path


class KnowledgeBase:
    """
    Singleton knowledge base loader.
    Loads knowledge_base.json once at startup and provides lookup methods.
    """
    _instance = None
    _kb_data: Optional[Dict[str, Any]] = None
    _initialized = False
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(KnowledgeBase, cls).__new__(cls)
        return cls._instance
    
    def __init__(self):
        if not self._initialized:
            self._load_knowledge_base()
            self._initialized = True
    
    def _load_knowledge_base(self) -> None:
        """Load knowledge_base.json from project root"""
        # Get project root (parent of agentic_support_ai)
        current_dir = Path(__file__).parent
        project_root = current_dir.parent
        kb_path = project_root / "knowledge_base.json"
        
        if not kb_path.exists():
            raise FileNotFoundError(
                f"knowledge_base.json not found at {kb_path}. "
                "This file is required for the system to operate."
            )
        
        try:
            with open(kb_path, 'r', encoding='utf-8') as f:
                self._kb_data = json.load(f)
            self._validate_kb()
        except json.JSONDecodeError as e:
            raise ValueError(f"Invalid JSON in knowledge_base.json: {e}")
        except Exception as e:
            raise RuntimeError(f"Failed to load knowledge_base.json: {e}")
    
    def _validate_kb(self) -> None:
        """Validate knowledge base structure"""
        if not self._kb_data:
            raise ValueError("Knowledge base is empty")
        
        required_keys = ["domain", "categories", "intensity_priority_mapping"]
        for key in required_keys:
            if key not in self._kb_data:
                raise ValueError(f"Missing required key in knowledge_base.json: {key}")
        
        # Validate categories structure
        categories = self._kb_data.get("categories", [])
        if not categories:
            raise ValueError("No categories found in knowledge_base.json")
        
        for cat in categories:
            if "id" not in cat or "name" not in cat or "subcategories" not in cat:
                raise ValueError(f"Invalid category structure: {cat}")
            
            if not cat["subcategories"]:
                raise ValueError(f"Category {cat['name']} has no subcategories")
            
            for subcat in cat["subcategories"]:
                required_subcat_keys = ["id", "name", "intensity", "priority", "security_critical"]
                for key in required_subcat_keys:
                    if key not in subcat:
                        raise ValueError(
                            f"Subcategory {subcat.get('name', 'unknown')} missing required key: {key}"
                        )
    
    def get_domain(self) -> str:
        """Get domain name"""
        return self._kb_data.get("domain", "Banking / FinTech")
    
    def get_categories(self) -> List[Dict[str, Any]]:
        """Get all categories"""
        return self._kb_data.get("categories", [])
    
    def get_category_by_id(self, category_id: str) -> Optional[Dict[str, Any]]:
        """Get category by ID"""
        categories = self.get_categories()
        for cat in categories:
            if cat.get("id") == category_id:
                return cat
        return None
    
    def get_category_by_name(self, category_name: str) -> Optional[Dict[str, Any]]:
        """Get category by name"""
        categories = self.get_categories()
        for cat in categories:
            if cat.get("name") == category_name:
                return cat
        return None
    
    def get_subcategory(self, category_id: str, subcategory_id: str) -> Optional[Dict[str, Any]]:
        """Get subcategory by category ID and subcategory ID"""
        category = self.get_category_by_id(category_id)
        if not category:
            return None
        
        subcategories = category.get("subcategories", [])
        for subcat in subcategories:
            if subcat.get("id") == subcategory_id:
                return subcat
        return None
    
    def get_subcategory_by_name(self, category_name: str, subcategory_name: str) -> Optional[Dict[str, Any]]:
        """Get subcategory by category name and subcategory name"""
        category = self.get_category_by_name(category_name)
        if not category:
            return None
        
        subcategories = category.get("subcategories", [])
        for subcat in subcategories:
            if subcat.get("name") == subcategory_name:
                return subcat
        return None
    
    def is_security_critical(self, category_id: str, subcategory_id: str) -> bool:
        """Check if subcategory is security-critical (triggers kill-switch)"""
        subcat = self.get_subcategory(category_id, subcategory_id)
        if not subcat:
            return False
        return subcat.get("security_critical", False)
    
    def get_intensity(self, category_id: str, subcategory_id: str) -> Optional[int]:
        """Get intensity for subcategory"""
        subcat = self.get_subcategory(category_id, subcategory_id)
        if not subcat:
            return None
        return subcat.get("intensity")
    
    def get_priority(self, category_id: str, subcategory_id: str) -> Optional[str]:
        """Get priority for subcategory"""
        subcat = self.get_subcategory(category_id, subcategory_id)
        if not subcat:
            return None
        return subcat.get("priority")
    
    def get_security_actions(self) -> Dict[str, Any]:
        """Get security critical actions configuration"""
        return self._kb_data.get("security_critical_actions", {
            "immediate_intensity": 4,
            "immediate_priority": "Critical",
            "escalation_required": True,
            "user_advisory": "Please secure your account immediately."
        })
    
    def get_invalid_key_handling(self) -> Dict[str, Any]:
        """Get invalid KB key handling configuration"""
        return self._kb_data.get("invalid_kb_key_handling", {
            "action": "escalate",
            "priority": "Critical",
            "message": "Invalid configuration detected. Case escalated."
        })


# Global singleton instance
_kb_instance: Optional[KnowledgeBase] = None


def get_knowledge_base() -> KnowledgeBase:
    """Get or create knowledge base singleton"""
    global _kb_instance
    if _kb_instance is None:
        _kb_instance = KnowledgeBase()
    return _kb_instance