""" AI-Powered Product Recommendations Engine Provides intelligent product suggestions across the platform """ from django.db.models import Count, Q, Avg from django.core.cache import cache from collections import Counter from products.models import Product from decimal import Decimal import random class ProductRecommender: """ AI-powered recommendation engine for products """ def get_similar_products(self, product, limit=4): """ Get similar products based on category, price, and attributes Uses collaborative filtering approach """ if not product: return Product.objects.none() effective_price = product.sale_price or product.price base_qs = Product.objects.filter( Q(vendor__isnull=True) | Q(vendor__status='ACTIVE'), category=product.category, is_active=True ).exclude(id=product.id) if effective_price is not None: price_min = effective_price * Decimal("0.7") price_max = effective_price * Decimal("1.3") base_qs = base_qs.filter(price__gte=price_min, price__lte=price_max) similar = base_qs.select_related('category', 'vendor').order_by('-sales_count')[:limit] return similar def get_frequently_bought_together(self, product_id, limit=4): """ Find products frequently bought together with this product Uses association rule mining """ from orders.models import OrderItem # Cache key for performance cache_key = 'fbt_' + str(product_id) cached_result = cache.get(cache_key) if cached_result: return cached_result # Get orders containing this product orders_with_product = OrderItem.objects.filter( product_id=product_id ).values_list('order_id', flat=True)[:100] # Limit for performance if not orders_with_product: return Product.objects.none() # Get other products from those orders other_products = OrderItem.objects.filter( order_id__in=orders_with_product ).exclude( product_id=product_id ).values_list('product_id', flat=True) # Count frequency product_counts = Counter(other_products) top_product_ids = [pid for pid, count in product_counts.most_common(limit)] # Get the products result = Product.objects.filter( Q(vendor__isnull=True) | Q(vendor__status='ACTIVE'), id__in=top_product_ids, is_active=True ).select_related('category', 'vendor')[:limit] # Cache for 1 hour cache.set(cache_key, result, 3600) return result def get_personalized_recommendations(self, user, limit=8): """ Get personalized recommendations based on user's browsing and purchase history Uses hybrid recommendation (content-based + collaborative) """ if not user or not user.is_authenticated: # Return trending products for anonymous users return self.get_trending_products(limit) from orders.models import Order # Get user's purchase history purchased_products = Product.objects.filter( orderitem__order__user=user, orderitem__order__status__in=['PENDING', 'PROCESSING', 'SHIPPED', 'DELIVERED'] ).distinct() # Analyze user's favorite categories favorite_categories = purchased_products.values('category').annotate( count=Count('category') ).order_by('-count')[:3] category_ids = [cat['category'] for cat in favorite_categories] # Get recommendations from favorite categories recommendations = Product.objects.filter( Q(vendor__isnull=True) | Q(vendor__status='ACTIVE'), category_id__in=category_ids, is_active=True ).exclude( id__in=purchased_products.values_list('id', flat=True) ).annotate( avg_rating=Avg('reviews__rating') ).order_by('-avg_rating', '-sales_count')[:limit] # If not enough recommendations, fill with trending if recommendations.count() < limit: trending = self.get_trending_products(limit - recommendations.count()) recommendations = list(recommendations) + list(trending) return recommendations[:limit] def get_trending_products(self, limit=8): """ Get currently trending products based on recent sales and views """ cache_key = 'trending_products_' + str(limit) cached = cache.get(cache_key) if cached: return cached # Score = (sales_count * 2) + (views_count * 0.1) # This weighs sales more heavily than views trending = Product.objects.filter( Q(vendor__isnull=True) | Q(vendor__status='ACTIVE'), is_active=True ).select_related('category', 'vendor').order_by( '-sales_count', '-views_count' )[:limit] # Cache for 30 minutes cache.set(cache_key, trending, 1800) return trending def get_you_may_also_like(self, user, current_product, limit=4): """ Combine multiple recommendation strategies """ recommendations = [] # Strategy 1: Similar products (50%) similar = list(self.get_similar_products(current_product, limit=2)) recommendations.extend(similar) # Strategy 2: Frequently bought together (30%) if current_product: fbt = list(self.get_frequently_bought_together(current_product.id, limit=1)) recommendations.extend(fbt) # Strategy 3: Personalized (20%) if user and user.is_authenticated: personalized = list(self.get_personalized_recommendations(user, limit=1)) recommendations.extend(personalized) # Remove duplicates and limit seen = set() unique_recommendations = [] for product in recommendations: if product.id not in seen: seen.add(product.id) unique_recommendations.append(product) return unique_recommendations[:limit] class SmartSearch: """ AI-enhanced search with typo correction and synonym expansion """ # Common typos and corrections TYPO_CORRECTIONS = { 'wireles': 'wireless', 'headphon': 'headphone', 'labtop': 'laptop', 'computr': 'computer', 'phon': 'phone', 'mobil': 'mobile', 'tshirt': 't-shirt', 'shooes': 'shoes', 'accesories': 'accessories', 'electonic': 'electronic', } # Synonyms for query expansion SYNONYMS = { 'phone': ['mobile', 'smartphone', 'cellphone'], 'laptop': ['notebook', 'computer', 'pc'], 'shirt': ['t-shirt', 'tee', 'top'], 'pants': ['trousers', 'jeans'], 'shoes': ['footwear', 'sneakers'], 'headphones': ['earphones', 'earbuds', 'headset'], 'watch': ['timepiece', 'smartwatch'], 'bag': ['backpack', 'purse', 'handbag'], } def search(self, query, filters=None): """ Enhanced search with AI features """ if not query: return Product.objects.none() # Step 1: Correct typos corrected_query = self.correct_typos(query) # Step 2: Expand with synonyms search_terms = self.expand_synonyms(corrected_query) # Step 3: Build query q_objects = Q() for term in search_terms: q_objects |= Q(name__icontains=term) q_objects |= Q(description__icontains=term) q_objects |= Q(short_description__icontains=term) q_objects |= Q(category__name__icontains=term) q_objects |= Q(brand__icontains=term) # Filter by vendor status q_objects &= (Q(vendor__isnull=True) | Q(vendor__status='ACTIVE')) q_objects &= Q(is_active=True) products = Product.objects.filter(q_objects).select_related( 'category', 'vendor' ).distinct() # Apply additional filters if provided if filters: if 'category' in filters: products = products.filter(category__slug=filters['category']) if 'min_price' in filters: products = products.filter(price__gte=filters['min_price']) if 'max_price' in filters: products = products.filter(price__lte=filters['max_price']) # Sort by relevance (simplified scoring) # Products with search term in name get higher priority return products.order_by('-is_featured', '-sales_count') def correct_typos(self, query): """ Correct common typos in search query """ words = query.lower().split() corrected_words = [] for word in words: # Check if word is a known typo corrected = self.TYPO_CORRECTIONS.get(word, word) corrected_words.append(corrected) return ' '.join(corrected_words) def expand_synonyms(self, query): """ Expand query with synonyms for better coverage """ words = query.lower().split() all_terms = [query] # Include original query for word in words: if word in self.SYNONYMS: # Add synonyms all_terms.extend(self.SYNONYMS[word]) return all_terms # Singleton instances recommender = ProductRecommender() smart_search = SmartSearch()