dynamic_category_extractor.py

"""
DynamicCategoryExtractor - Intelligent product categorization

This module analyzes product data to generate dynamic category mappings
and extract meaningful terms from user queries. It provides utilities for
matching queries to product categories based on semantic relevance.

The extractor handles multiple languages (Dutch, French, English) and can
identify category-related terms without relying on hardcoded mappings.
"""
import re
import logging
from typing import Dict, List, Any, Optional, Tuple, Set
from collections import Counter

# Configure logging
logger = logging.getLogger(__name__)

class DynamicCategoryExtractor:
    def __init__(self, integrator):
        """
        Initialize the category extractor
        
        Args:
            integrator: Instance of ProductDataIntegrator
        """
        self.integrator = integrator
        self.term_to_category = {}  # Maps terms to categories
        self.category_indicators = self._initialize_indicators()
        self.category_weights = {}  # Track importance of each category
        
        # Build initial category mapping
        self.build_category_mapping()
    
    def _initialize_indicators(self) -> Dict[str, Dict]:
        """
        Initialize basic category indicators in multiple languages
        
        Returns:
            Dict: Mapping of indicator terms to categories and importance
        """
        indicators = {
            # Mechanical indicators
            'mechanisch': {'category': 'mechanical', 'weight': 5},
            'mécanique': {'category': 'mechanical', 'weight': 5},
            'mechanic': {'category': 'mechanical', 'weight': 5},
            'mechanische sloten': {'category': 'mechanical locks', 'weight': 8},
            'mortise lock': {'category': 'mechanical locks', 'weight': 7},
            'serrure mécanique': {'category': 'mechanical locks', 'weight': 8},
            
            # Electronic indicators
            'elektronisch': {'category': 'electronic', 'weight': 5},
            'elektrisch': {'category': 'electronic', 'weight': 5},
            'électronique': {'category': 'electronic', 'weight': 5},
            'électrique': {'category': 'electronic', 'weight': 5},
            'electronic': {'category': 'electronic', 'weight': 5},
            'elektromechanische sloten': {'category': 'electronic locks', 'weight': 8},
            'electric lock': {'category': 'electronic locks', 'weight': 7},
            'serrure électrique': {'category': 'electronic locks', 'weight': 8},
            
            # Magnet indicators
            'magneet': {'category': 'magnets', 'weight': 6},
            'ventouse': {'category': 'magnets', 'weight': 6},
            'magnet': {'category': 'magnets', 'weight': 6},
            
            # Cylinder indicators
            'cilinder': {'category': 'cylinders', 'weight': 6},
            'cylindre': {'category': 'cylinders', 'weight': 6},
            'cylinder': {'category': 'cylinders', 'weight': 6},
            
            # Door closer indicators
            'deursluiter': {'category': 'door closers', 'weight': 6},
            'ferme-porte': {'category': 'door closers', 'weight': 6},
            'door closer': {'category': 'door closers', 'weight': 6},
            
            # Handle indicators
            'kruk': {'category': 'handles', 'weight': 6},
            'béquille': {'category': 'handles', 'weight': 6},
            'handle': {'category': 'handles', 'weight': 6},
            
            # Hold open indicators
            'openhoud': {'category': 'hold open', 'weight': 6},
            'aimant': {'category': 'hold open', 'weight': 6},
            'hold open': {'category': 'hold open', 'weight': 6},
        }
        
        return indicators
    
    def build_category_mapping(self):
        """
        Build mapping of terms to categories based on product data
        
        This method analyzes all available product data to create a
        dynamic mapping between terms and product categories.
        """
        # Start with our basic indicators
        self.term_to_category = {}
        
        # Get all product categories from integrator
        categories = {}
        for product in self.integrator.unified_data.get("products", []):
            group = product.get("group", "").lower()
            subgroup = product.get("subgroup", "").lower()
            
            if group:
                if group not in categories:
                    categories[group] = {"count": 0, "subgroups": {}}
                
                categories[group]["count"] += 1
                
                if subgroup:
                    if subgroup not in categories[group]["subgroups"]:
                        categories[group]["subgroups"][subgroup] = 0
                    
                    categories[group]["subgroups"][subgroup] += 1
        
        # Calculate weights for each category based on product count
        total_products = sum(category["count"] for category in categories.values())
        
        for category, data in categories.items():
            weight = (data["count"] / max(total_products, 1)) * 10  # 0-10 scale
            self.category_weights[category] = weight
        
        # Process each category to extract terms
        for category, data in categories.items():
            # Extract terms from category name
            category_terms = self._extract_terms(category)
            
            for term in category_terms:
                if term not in self.term_to_category:
                    self.term_to_category[term] = []
                
                # Add category with weight
                self.term_to_category[term].append({
                    "category": category,
                    "weight": self.category_weights.get(category, 5) * 2  # Double weight for direct category name
                })
            
            # Process subgroups
            for subgroup, count in data["subgroups"].items():
                # Extract terms from subgroup name
                subgroup_terms = self._extract_terms(subgroup)
                
                for term in subgroup_terms:
                    if term not in self.term_to_category:
                        self.term_to_category[term] = []
                    
                    # Add category with weight based on subgroup
                    subgroup_weight = (count / max(data["count"], 1)) * self.category_weights.get(category, 5)
                    self.term_to_category[term].append({
                        "category": category,
                        "weight": subgroup_weight
                    })
        
        # Process product descriptions
        self._process_product_descriptions()
        
        # Add our pre-defined indicators
        for term, data in self.category_indicators.items():
            if term not in self.term_to_category:
                self.term_to_category[term] = []
            
            # Add with high weight since these are explicit indicators
            self.term_to_category[term].append({
                "category": data["category"],
                "weight": data["weight"]
            })
        
        logger.info(f"Built category mapping with {len(self.term_to_category)} terms")
    
    def _process_product_descriptions(self):
        """
        Process product descriptions to extract category-related terms
        """
        term_categories = {}  # Maps terms to categories they appear in
        
        # Process each product
        for product in self.integrator.unified_data.get("products", []):
            group = product.get("group", "").lower()
            if not group:
                continue
                
            # Extract terms from description
            description = product.get("description", "").lower()
            if description:
                terms = self._extract_terms(description)
                
                for term in terms:
                    if term not in term_categories:
                        term_categories[term] = Counter()
                    
                    term_categories[term][group] += 1
            
            # Also process long_description if available
            long_description = product.get("long_description", "").lower()
            if long_description:
                terms = self._extract_terms(long_description)
                
                for term in terms:
                    if term not in term_categories:
                        term_categories[term] = Counter()
                    
                    term_categories[term][group] += 1
        
        # Now process the term_categories to update term_to_category
        for term, category_counts in term_categories.items():
            # Calculate total occurrences
            total_occurrences = sum(category_counts.values())
            
            # Only consider terms that appear a reasonable number of times
            if total_occurrences < 3:
                continue
            
            if term not in self.term_to_category:
                self.term_to_category[term] = []
            
            # Add each category with weight proportional to occurrence frequency
            for category, count in category_counts.items():
                # Calculate weight as a function of occurrence frequency and category weight
                frequency_weight = (count / total_occurrences) * 5  # 0-5 scale
                category_weight = self.category_weights.get(category, 3)
                weight = frequency_weight + category_weight
                
                # Add to term_to_category
                self.term_to_category[term].append({
                    "category": category,
                    "weight": weight
                })
    
    def _extract_terms(self, text: str) -> List[str]:
        """
        Extract meaningful terms from text
        
        Args:
            text: Text to extract terms from
            
        Returns:
            List[str]: List of extracted terms
        """
        if not text:
            return []
        
        # Convert to lowercase
        text_lower = text.lower()
        
        # Remove numbers and punctuation
        text_clean = re.sub(r'[0-9.,;:!?()[\]{}]', ' ', text_lower)
        
        # Split into words
        words = re.findall(r'\b\w{3,}\b', text_clean)
        
        # Remove common stopwords in Dutch, French and English
        stopwords = {
            # Dutch stopwords
            'een', 'de', 'het', 'en', 'van', 'in', 'is', 'dat', 'op', 'te', 
            'voor', 'met', 'zijn', 'niet', 'aan', 'er', 'ook', 'als', 'bij',
            'door', 'maar', 'naar', 'dan', 'ze', 'uit', 'wel', 'nog', 'al',
            # French stopwords
            'le', 'la', 'les', 'un', 'une', 'des', 'et', 'est', 'que', 'en',
            'à', 'pour', 'dans', 'ce', 'qui', 'pas', 'sur', 'plus', 'avec',
            'vous', 'au', 'par', 'mais', 'nous', 'sont', 'du', 'comme', 'je',
            # English stopwords
            'the', 'a', 'an', 'and', 'is', 'are', 'was', 'were', 'be', 'been',
            'to', 'of', 'for', 'in', 'on', 'at', 'by', 'with', 'from', 'about'
        }
        
        filtered_words = [word for word in words if word not in stopwords and len(word) >= 3]
        
        # Extract single terms
        terms = set(filtered_words)
        
        # Also extract meaningful bigrams (pairs of words)
        if len(filtered_words) >= 2:
            for i in range(len(filtered_words) - 1):
                bigram = f"{filtered_words[i]} {filtered_words[i+1]}"
                terms.add(bigram)
        
        return list(terms)
    
    def detect_categories(self, query: str) -> List[Dict]:
        """
        Detect relevant product categories from a user query
        
        Args:
            query: User query text
            
        Returns:
            List[Dict]: Sorted list of categories with scores
        """
        # Extract terms from query
        query_terms = self._extract_terms(query)
        
        # Track categories and their scores
        category_scores = {}
        
        # Process each term
        for term in query_terms:
            # Look for exact matches
            if term in self.term_to_category:
                for category_data in self.term_to_category[term]:
                    category = category_data["category"]
                    weight = category_data["weight"]
                    
                    if category not in category_scores:
                        category_scores[category] = 0
                    
                    category_scores[category] += weight
            
            # Also try partial matches for longer terms
            if len(term) > 5:
                for mapping_term, categories in self.term_to_category.items():
                    if mapping_term in term or term in mapping_term:
                        # Partial match, add with reduced weight
                        similarity = len(set(term) & set(mapping_term)) / max(len(term), len(mapping_term))
                        
                        for category_data in categories:
                            category = category_data["category"]
                            weight = category_data["weight"] * similarity * 0.7  # Reduce weight for partial matches
                            
                            if category not in category_scores:
                                category_scores[category] = 0
                            
                            category_scores[category] += weight
        
        # Check for direct type indicators (mechanical vs electronic)
        mechanical_indicators = ['mechanisch', 'mécanique', 'mechanic']
        electronic_indicators = ['elektronisch', 'elektrisch', 'électronique', 'électrique', 'electronic']
        
        query_lower = query.lower()
        is_mechanical = any(indicator in query_lower for indicator in mechanical_indicators)
        is_electronic = any(indicator in query_lower for indicator in electronic_indicators)
        
        # If mechanical is explicitly mentioned, boost mechanical categories
        if is_mechanical:
            for category in list(category_scores.keys()):
                if any(indicator in category for indicator in mechanical_indicators):
                    category_scores[category] *= 1.5  # 50% boost
                elif any(indicator in category for indicator in electronic_indicators):
                    category_scores[category] *= 0.5  # 50% penalty
        
        # If electronic is explicitly mentioned, boost electronic categories
        if is_electronic:
            for category in list(category_scores.keys()):
                if any(indicator in category for indicator in electronic_indicators):
                    category_scores[category] *= 1.5  # 50% boost
                elif any(indicator in category for indicator in mechanical_indicators):
                    category_scores[category] *= 0.5  # 50% penalty
        
        # Convert to sorted list
        result = [
            {"category": category, "score": score}
            for category, score in category_scores.items()
        ]
        
        # Sort by score (highest first)
        result.sort(key=lambda x: x["score"], reverse=True)
        
        return result
    
    def get_category_terms(self, category: str) -> List[str]:
        """
        Get terms associated with a specific category
        
        Args:
            category: Category name
            
        Returns:
            List[str]: Terms associated with the category
        """
        category_lower = category.lower()
        terms = []
        
        for term, categories in self.term_to_category.items():
            for category_data in categories:
                if category_data["category"] == category_lower and category_data["weight"] > 3:
                    terms.append(term)
                    break
        
        return terms
    
    def categorize_product(self, product: Dict) -> List[str]:
        """
        Determine the most likely categories for a product
        
        Args:
            product: Product data dictionary
            
        Returns:
            List[str]: Most likely categories
        """
        # If product already has a group, use it
        if "group" in product and product["group"]:
            return [product["group"].lower()]
        
        # Otherwise, try to categorize based on description
        categories = []
        description = product.get("description", "")
        long_description = product.get("long_description", "")
        
        if description or long_description:
            full_text = f"{description} {long_description}"
            detected = self.detect_categories(full_text)
            
            if detected:
                categories = [c["category"] for c in detected[:3]]  # Top 3 categories
        
        return categories