#!/usr/bin/env python3 """ Group 5 Pattern Recognition Project - Deployment Version ======================================================= Recipe Recommendation System with Google Drive file loading for deployment. Optimized for Hugging Face Spaces or similar platforms. """ import gradio as gr import torch from transformers import BertTokenizer, BertModel import pickle import os import csv from typing import List, Dict import time import ast import requests import gdown from pathlib import Path # Google Drive file IDs (you'll need to replace these with your actual file IDs) GOOGLE_DRIVE_FILES = { 'torch_recipe_embeddings_231630.pt': '1PSidY1toSfgECXDxa4pGza56Jq6vOq6t', 'tag_based_bert_model.pth': '1LBl7yFs5JFqOsgfn88BF9g83W9mxiBm6', 'RAW_recipes.csv': '1rFJQzg_ErwEpN6WmhQ4jRyiXv6JCINyf', 'recipe_statistics_231630.pkl': '1n8TNT-6EA_usv59CCCU1IXqtuM7i084E', 'recipe_scores_231630.pkl': '1gfPBzghKHOZqgJu4VE9NkandAd6FGjrA' } def download_file_from_drive(file_id: str, destination: str) -> bool: """Download file from Google Drive""" try: print(f"📥 Downloading {destination}...") url = f"https://drive.google.com/uc?id={file_id}" gdown.download(url, destination, quiet=False) return True except Exception as e: print(f"❌ Error downloading {destination}: {e}") return False def ensure_files_downloaded(): """Ensure all required files are downloaded from Google Drive""" print("🔍 Checking required files...") for filename, file_id in GOOGLE_DRIVE_FILES.items(): if not os.path.exists(filename): if file_id == 'YOUR_EMBEDDINGS_FILE_ID_HERE': print(f"⚠️ {filename} not configured for download") continue print(f"📥 Downloading {filename} from Google Drive...") success = download_file_from_drive(file_id, filename) if not success: print(f"❌ Failed to download {filename}") return False print("✅ All files ready!") return True class DeployableRecipeSearch: """ Deployment-ready recipe search system """ def __init__(self): print("🚀 Initializing Recipe Search System...") self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"📱 Device: {self.device}") # Ensure files are downloaded if not ensure_files_downloaded(): print("❌ Failed to download required files") self.is_ready = False return # Load tokenizer and model self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased') self.model = BertModel.from_pretrained('bert-base-uncased') # Load trained model if available if os.path.exists('tag_based_bert_model.pth'): print("🧠 Loading trained BERT model...") self.model.load_state_dict(torch.load('tag_based_bert_model.pth', map_location=self.device)) print("✅ Trained model loaded!") else: print("⚠️ Using pre-trained BERT") self.model.to(self.device) self.model.eval() # Load data self.load_data() print("🎉 Recipe Search System ready!") def safe_literal_eval(self, text): """Safely evaluate string representations of lists""" if not text or text == 'nan' or str(text).lower() == 'nan': return [] try: if isinstance(text, str) and text.startswith('[') and text.endswith(']'): return ast.literal_eval(text) elif isinstance(text, str): return [item.strip() for item in text.split(',') if item.strip()] elif isinstance(text, list): return text else: return [] except: return [] def safe_int(self, value): """Safely convert value to int""" try: return int(float(value)) except: return 0 def load_data(self): """Load all required data""" # Load PyTorch embeddings embeddings_file = 'torch_recipe_embeddings_231630.pt' if os.path.exists(embeddings_file): print(f"📥 Loading embeddings...") self.recipe_embeddings = torch.load(embeddings_file, map_location=self.device) print(f"✅ Loaded {self.recipe_embeddings.shape[0]} embeddings") else: print(f"❌ Embeddings not found") self.is_ready = False return # Load recipes from CSV self.load_recipes_from_csv() # Load statistics and scores self.load_statistics_and_scores() # Check if we have everything we need self.is_ready = all([ self.recipe_embeddings is not None, len(self.recipes) > 0, len(self.recipe_stats) > 0, len(self.recipe_scores) > 0 ]) if self.is_ready: self.fix_recipe_id_mismatches() print("🎯 All data loaded successfully!") else: print("⚠️ Some data missing") def load_recipes_from_csv(self): """Load and filter recipes from CSV""" print("📊 Loading recipes from CSV...") self.recipes = [] if os.path.exists('RAW_recipes.csv'): valid_recipes = [] with open('RAW_recipes.csv', 'r', encoding='utf-8') as file: csv_reader = csv.DictReader(file) for row_idx, row in enumerate(csv_reader): try: # Apply filtering logic name = row.get('name', '') if not name or str(name).lower().strip() in ['', 'nan', 'unknown recipe']: continue name = str(name).lower().strip() tags = self.safe_literal_eval(row.get('tags', '[]')) ingredients = self.safe_literal_eval(row.get('ingredients', '[]')) # Filter conditions if not tags or len(tags) == 0: continue if not ingredients or len(ingredients) == 0: continue if len(name) == 0 or name == 'unknown recipe': continue recipe = { 'id': int(row.get('id', row_idx)), 'name': name, 'minutes': self.safe_int(row.get('minutes', 0)), 'tags': tags, 'ingredients': ingredients, 'n_steps': self.safe_int(row.get('n_steps', 0)), 'description': str(row.get('description', '')).strip() } valid_recipes.append(recipe) if len(valid_recipes) >= 231630: break except Exception as e: continue self.recipes = valid_recipes print(f"✅ Loaded {len(self.recipes)} recipes") else: print("❌ RAW_recipes.csv not found") self.recipes = [] def load_statistics_and_scores(self): """Load recipe statistics and scores""" # Load statistics stats_file = 'recipe_statistics_231630.pkl' try: if os.path.exists(stats_file): with open(stats_file, 'rb') as f: self.recipe_stats = pickle.load(f) print(f"✅ Loaded statistics for {len(self.recipe_stats)} recipes") else: self.recipe_stats = {} for recipe in self.recipes: self.recipe_stats[recipe['id']] = (4.0, 10, 5) except Exception as e: print(f"⚠️ Statistics loading failed: {e}") self.recipe_stats = {} for recipe in self.recipes: self.recipe_stats[recipe['id']] = (4.0, 10, 5) # Load scores scores_file = 'recipe_scores_231630.pkl' try: if os.path.exists(scores_file): with open(scores_file, 'rb') as f: self.recipe_scores = pickle.load(f) print(f"✅ Loaded scores for {len(self.recipe_scores)} recipes") else: self.recipe_scores = {} for recipe in self.recipes: self.recipe_scores[recipe['id']] = 0.5 except Exception as e: print(f"⚠️ Scores loading failed: {e}") self.recipe_scores = {} for recipe in self.recipes: self.recipe_scores[recipe['id']] = 0.5 def fix_recipe_id_mismatches(self): """Filter statistics and scores to match loaded recipes""" loaded_recipe_ids = set(recipe['id'] for recipe in self.recipes) # Filter statistics original_stats_count = len(self.recipe_stats) self.recipe_stats = { recipe_id: stats for recipe_id, stats in self.recipe_stats.items() if recipe_id in loaded_recipe_ids } # Filter scores original_scores_count = len(self.recipe_scores) self.recipe_scores = { recipe_id: score for recipe_id, score in self.recipe_scores.items() if recipe_id in loaded_recipe_ids } print(f"🔧 Aligned data: Stats {original_stats_count}→{len(self.recipe_stats)}, Scores {original_scores_count}→{len(self.recipe_scores)}") def search_recipes(self, query: str, num_results: int = 5, min_rating: float = 3.0) -> str: """Search for recipes and return formatted HTML results""" if not self.is_ready: return """
❌ Search system not ready - files may still be downloading
""" if not query.strip(): return """
⚠️ Please enter a search query
""" try: start_time = time.time() # Tokenize query inputs = self.tokenizer( query, return_tensors='pt', truncation=True, max_length=128, padding='max_length' ).to(self.device) # Get query embedding with torch.no_grad(): outputs = self.model(**inputs) query_embedding = outputs.last_hidden_state[:, 0, :].cpu().flatten() # Calculate similarities recipe_embeddings_normalized = torch.nn.functional.normalize(self.recipe_embeddings, p=2, dim=1) query_embedding_normalized = torch.nn.functional.normalize(query_embedding.unsqueeze(0), p=2, dim=1) similarities = torch.mm(recipe_embeddings_normalized, query_embedding_normalized.t()).flatten() # Get top results top_indices = torch.argsort(similarities, descending=True)[:num_results * 3] results = [] for idx in top_indices: if len(results) >= num_results: break embedding_idx = idx.item() if embedding_idx < len(self.recipes): recipe = self.recipes[embedding_idx] recipe_id = recipe['id'] if recipe_id in self.recipe_stats: avg_rating, num_ratings, unique_users = self.recipe_stats[recipe_id] if avg_rating >= min_rating: similarity_score = similarities[idx].item() popularity_score = self.recipe_scores.get(recipe_id, 0.0) combined_score = 0.7 * similarity_score + 0.3 * popularity_score results.append({ 'name': recipe['name'], 'ingredients': recipe['ingredients'][:8] if isinstance(recipe['ingredients'], list) else [], 'tags': recipe['tags'][:6] if isinstance(recipe['tags'], list) else [], 'minutes': recipe.get('minutes', 0), 'n_steps': recipe.get('n_steps', 0), 'similarity_score': similarity_score, 'popularity_score': popularity_score, 'combined_score': combined_score, 'avg_rating': avg_rating, 'num_ratings': num_ratings, 'recipe_id': recipe_id }) search_time = time.time() - start_time if results: return self.format_results(query, results, search_time) else: return f"""
😔 No recipes found for "{query}" with rating ≥ {min_rating}
""" except Exception as e: return f"""
❌ Search error: {str(e)}
""" def format_results(self, query: str, results: List[Dict], search_time: float) -> str: """Format search results as HTML""" html = f"""

🎯 Found {len(results)} recipes for "{query}"

⚡ Search completed in {search_time:.2f}s

""" for i, recipe in enumerate(results, 1): ingredients = recipe['ingredients'] ingredients_text = ', '.join(ingredients) if ingredients else "No ingredients listed" if len(ingredients_text) > 150: ingredients_text = ingredients_text[:150] + "..." tags = recipe['tags'] tags_html = ' '.join([f'{tag}' for tag in tags]) if tags else "" time_text = f"{recipe['minutes']} min" if recipe['minutes'] > 0 else "Time not specified" recipe_html = f"""

{i}. {recipe['name']}

⏱️ {time_text} | 🔥 {recipe['n_steps']} steps | ⭐ {recipe['avg_rating']:.1f}/5.0 ({recipe['num_ratings']} ratings)
Match: {recipe['similarity_score']:.1%} Score: {recipe['combined_score']:.1%}
{tags_html}
🥘 Ingredients:
{ingredients_text}
""" html += recipe_html return html # Initialize the search system print("🔄 Initializing deployment-ready recipe search system...") try: search_system = DeployableRecipeSearch() except Exception as e: print(f"❌ Initialization failed: {e}") search_system = None def search_interface(query, num_results, min_rating): """Gradio interface function""" if search_system is None: return "
❌ System initialization failed
" return search_system.search_recipes(query, int(num_results), float(min_rating)) # Create Gradio interface with gr.Blocks(title="Group 5 Pattern Recognition Project", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🍽️ Group 5 Pattern Recognition Project ### Advanced Recipe Recommendation using Semantic Search """) with gr.Row(): with gr.Column(scale=1): query_input = gr.Textbox( label="🔍 Search for recipes", placeholder="e.g., 'chicken pasta', 'vegetarian salad', 'chocolate dessert'", lines=1 ) with gr.Row(): num_results = gr.Slider(1, 10, 5, step=1, label="Results") min_rating = gr.Slider(1.0, 5.0, 3.0, step=0.1, label="Min Rating") search_btn = gr.Button("Search Recipes", variant="primary") # Example buttons with gr.Row(): ex1 = gr.Button("🍗 Chicken Pasta", size="sm") ex2 = gr.Button("🥗 Healthy Salad", size="sm") ex3 = gr.Button("🍫 Chocolate Dessert", size="sm") with gr.Column(scale=1): results_output = gr.HTML("""

🔍 Ready to Search

Enter a search query and click "Search Recipes" to see results.

""") # Event handlers search_btn.click(search_interface, [query_input, num_results, min_rating], results_output) query_input.submit(search_interface, [query_input, num_results, min_rating], results_output) # Example buttons ex1.click(lambda: "chicken pasta", outputs=query_input) ex2.click(lambda: "healthy salad", outputs=query_input) ex3.click(lambda: "chocolate dessert", outputs=query_input) if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, # Standard port for Hugging Face Spaces share=False )